// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package typeutil

import (
	"encoding/binary"
	"encoding/json"
	"fmt"
	"math"
	"math/rand"
	"reflect"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
	"google.golang.org/protobuf/proto"

	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
	"github.com/milvus-io/milvus/pkg/v3/common"
	"github.com/milvus-io/milvus/pkg/v3/util/merr"
)

func TestSchema(t *testing.T) {
	schema := &schemapb.CollectionSchema{
		Name:        "testColl",
		Description: "",
		AutoID:      false,
		Fields: []*schemapb.FieldSchema{
			{
				FieldID:      100,
				Name:         "field_int8",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_Int8,
			},
			{
				FieldID:      101,
				Name:         "field_int16",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_Int16,
			},
			{
				FieldID:      102,
				Name:         "field_int32",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_Int32,
			},
			{
				FieldID:      103,
				Name:         "field_int64",
				IsPrimaryKey: true,
				Description:  "",
				DataType:     schemapb.DataType_Int64,
			},
			{
				FieldID:      104,
				Name:         "field_float",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_Float,
			},
			{
				FieldID:      105,
				Name:         "field_double",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_Double,
			},
			{
				FieldID:      106,
				Name:         "field_string",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_String,
				TypeParams: []*commonpb.KeyValuePair{
					{
						Key:   common.MaxLengthKey,
						Value: "125",
					},
				},
			},
			{
				FieldID:      107,
				Name:         "field_float_vector",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_FloatVector,
				TypeParams: []*commonpb.KeyValuePair{
					{
						Key:   common.DimKey,
						Value: "128",
					},
				},
				IndexParams: []*commonpb.KeyValuePair{
					{
						Key:   "nlist",
						Value: "128",
					},
				},
			},
			{
				FieldID:      108,
				Name:         "field_binary_vector",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_BinaryVector,
				TypeParams: []*commonpb.KeyValuePair{
					{
						Key:   common.DimKey,
						Value: "128",
					},
				},
			},
			{
				FieldID:     109,
				Name:        "field_array",
				DataType:    schemapb.DataType_Array,
				ElementType: schemapb.DataType_Int32,
			},
			{
				FieldID:  110,
				Name:     "field_json",
				DataType: schemapb.DataType_JSON,
			},
			{
				FieldID:      111,
				Name:         "field_float16_vector",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_Float16Vector,
				TypeParams: []*commonpb.KeyValuePair{
					{
						Key:   common.DimKey,
						Value: "128",
					},
				},
			},
			{
				FieldID:      112,
				Name:         "field_bfloat16_vector",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_BFloat16Vector,
				TypeParams: []*commonpb.KeyValuePair{
					{
						Key:   common.DimKey,
						Value: "128",
					},
				},
			},
			// Do not test on sparse float vector field.
			{
				FieldID:      113,
				Name:         "field_int8_vector",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_Int8Vector,
				TypeParams: []*commonpb.KeyValuePair{
					{
						Key:   common.DimKey,
						Value: "128",
					},
				},
			},
			{
				FieldID:      114,
				Name:         "field_geometry",
				IsPrimaryKey: false,
				DataType:     schemapb.DataType_Geometry,
			},
		},
	}

	t.Run("EstimateSizePerRecord", func(t *testing.T) {
		size, err := EstimateSizePerRecord(schema)
		assert.Equal(t, 2731, size)
		assert.NoError(t, err)
	})

	t.Run("SchemaHelper", func(t *testing.T) {
		_, err := CreateSchemaHelper(nil)
		assert.Error(t, err)

		helper, err := CreateSchemaHelper(schema)
		assert.NoError(t, err)

		field, err := helper.GetPrimaryKeyField()
		assert.NoError(t, err)
		assert.Equal(t, "field_int64", field.Name)

		field1, err := helper.GetFieldFromName("field_int8")
		assert.NoError(t, err)
		assert.Equal(t, "field_int8", field1.Name)

		field2, err := helper.GetFieldFromID(102)
		assert.NoError(t, err)
		assert.Equal(t, "field_int32", field2.Name)

		dim, err := helper.GetVectorDimFromID(107)
		assert.NoError(t, err)
		assert.Equal(t, 128, dim)

		dim, err = helper.GetVectorDimFromID(108)
		assert.NoError(t, err)
		assert.Equal(t, 128, dim)

		_, err = helper.GetVectorDimFromID(103)
		assert.Error(t, err)

		dim, err = helper.GetVectorDimFromID(111)
		assert.NoError(t, err)
		assert.Equal(t, 128, dim)

		dim, err = helper.GetVectorDimFromID(112)
		assert.NoError(t, err)
		assert.Equal(t, 128, dim)

		dim, err = helper.GetVectorDimFromID(113)
		assert.NoError(t, err)
		assert.Equal(t, 128, dim)
	})

	t.Run("Type", func(t *testing.T) {
		assert.False(t, IsVectorType(schemapb.DataType_Bool))
		assert.False(t, IsVectorType(schemapb.DataType_Int8))
		assert.False(t, IsVectorType(schemapb.DataType_Int16))
		assert.False(t, IsVectorType(schemapb.DataType_Int32))
		assert.False(t, IsVectorType(schemapb.DataType_Int64))
		assert.False(t, IsVectorType(schemapb.DataType_Float))
		assert.False(t, IsVectorType(schemapb.DataType_Double))
		assert.False(t, IsVectorType(schemapb.DataType_String))
		assert.True(t, IsVectorType(schemapb.DataType_BinaryVector))
		assert.True(t, IsVectorType(schemapb.DataType_FloatVector))
		assert.True(t, IsVectorType(schemapb.DataType_Float16Vector))
		assert.True(t, IsVectorType(schemapb.DataType_BFloat16Vector))
		assert.True(t, IsVectorType(schemapb.DataType_SparseFloatVector))
		assert.True(t, IsVectorType(schemapb.DataType_Int8Vector))
		assert.True(t, IsVectorType(schemapb.DataType_ArrayOfVector))

		assert.False(t, IsIntegerType(schemapb.DataType_Bool))
		assert.True(t, IsIntegerType(schemapb.DataType_Int8))
		assert.True(t, IsIntegerType(schemapb.DataType_Int16))
		assert.True(t, IsIntegerType(schemapb.DataType_Int32))
		assert.True(t, IsIntegerType(schemapb.DataType_Int64))
		assert.False(t, IsIntegerType(schemapb.DataType_Float))
		assert.False(t, IsIntegerType(schemapb.DataType_Double))
		assert.False(t, IsIntegerType(schemapb.DataType_String))
		assert.False(t, IsIntegerType(schemapb.DataType_BinaryVector))
		assert.False(t, IsIntegerType(schemapb.DataType_FloatVector))
		assert.False(t, IsIntegerType(schemapb.DataType_Float16Vector))
		assert.False(t, IsIntegerType(schemapb.DataType_BFloat16Vector))
		assert.False(t, IsIntegerType(schemapb.DataType_SparseFloatVector))
		assert.False(t, IsIntegerType(schemapb.DataType_Int8Vector))

		assert.False(t, IsFloatingType(schemapb.DataType_Bool))
		assert.False(t, IsFloatingType(schemapb.DataType_Int8))
		assert.False(t, IsFloatingType(schemapb.DataType_Int16))
		assert.False(t, IsFloatingType(schemapb.DataType_Int32))
		assert.False(t, IsFloatingType(schemapb.DataType_Int64))
		assert.True(t, IsFloatingType(schemapb.DataType_Float))
		assert.True(t, IsFloatingType(schemapb.DataType_Double))
		assert.False(t, IsFloatingType(schemapb.DataType_String))
		assert.False(t, IsFloatingType(schemapb.DataType_BinaryVector))
		assert.False(t, IsFloatingType(schemapb.DataType_FloatVector))
		assert.False(t, IsFloatingType(schemapb.DataType_Float16Vector))
		assert.False(t, IsFloatingType(schemapb.DataType_BFloat16Vector))
		assert.False(t, IsFloatingType(schemapb.DataType_SparseFloatVector))
		assert.False(t, IsFloatingType(schemapb.DataType_Int8Vector))

		assert.False(t, IsSparseFloatVectorType(schemapb.DataType_Bool))
		assert.False(t, IsSparseFloatVectorType(schemapb.DataType_Int8))
		assert.False(t, IsSparseFloatVectorType(schemapb.DataType_Int16))
		assert.False(t, IsSparseFloatVectorType(schemapb.DataType_Int32))
		assert.False(t, IsSparseFloatVectorType(schemapb.DataType_Int64))
		assert.False(t, IsSparseFloatVectorType(schemapb.DataType_Float))
		assert.False(t, IsSparseFloatVectorType(schemapb.DataType_Double))
		assert.False(t, IsSparseFloatVectorType(schemapb.DataType_String))
		assert.False(t, IsSparseFloatVectorType(schemapb.DataType_BinaryVector))
		assert.False(t, IsSparseFloatVectorType(schemapb.DataType_FloatVector))
		assert.False(t, IsSparseFloatVectorType(schemapb.DataType_Float16Vector))
		assert.False(t, IsSparseFloatVectorType(schemapb.DataType_BFloat16Vector))
		assert.True(t, IsSparseFloatVectorType(schemapb.DataType_SparseFloatVector))
		assert.False(t, IsSparseFloatVectorType(schemapb.DataType_Int8Vector))

		assert.True(t, IsVectorArrayType(schemapb.DataType_ArrayOfVector))
	})
}

func TestHasTextField(t *testing.T) {
	assert.False(t, HasTextField(nil))
	assert.False(t, HasTextField(&schemapb.CollectionSchema{}))

	schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{FieldID: 100, DataType: schemapb.DataType_Int64},
			{FieldID: 101, DataType: schemapb.DataType_Text},
		},
	}
	assert.True(t, HasTextField(schema))
}

func TestValidateTextRequiresStorageV3(t *testing.T) {
	textSchema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{FieldID: 100, DataType: schemapb.DataType_Int64},
			{FieldID: 101, DataType: schemapb.DataType_Text},
		},
	}
	ordinarySchema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{FieldID: 100, DataType: schemapb.DataType_Int64},
		},
	}

	assert.NoError(t, ValidateTextRequiresStorageV3(nil, false))
	assert.NoError(t, ValidateTextRequiresStorageV3(ordinarySchema, false))
	assert.Error(t, ValidateTextRequiresStorageV3(textSchema, false))
	assert.NoError(t, ValidateTextRequiresStorageV3(textSchema, true))
}

func TestAllowGrowingSourceFlush(t *testing.T) {
	ordinarySchema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{FieldID: 100, DataType: schemapb.DataType_Int64},
		},
	}
	textSchema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{FieldID: 100, DataType: schemapb.DataType_Int64},
			{FieldID: 101, DataType: schemapb.DataType_Text},
		},
	}

	assert.False(t, AllowGrowingSourceFlush(ordinarySchema, false, true))
	assert.False(t, AllowGrowingSourceFlush(textSchema, false, true))
	assert.False(t, AllowGrowingSourceFlush(ordinarySchema, true, false))
	assert.True(t, AllowGrowingSourceFlush(ordinarySchema, true, true))
	assert.False(t, AllowGrowingSourceFlush(textSchema, true, false))
	assert.True(t, AllowGrowingSourceFlush(textSchema, true, true))
	assert.False(t, AllowGrowingSourceFlush(nil, true, false))
	assert.True(t, AllowGrowingSourceFlush(nil, true, true))
	assert.Equal(t,
		AllowGrowingSourceFlush(textSchema, true, false),
		UseGrowingSourceFlush(textSchema, true, false))
}

func TestSchema_GetVectorFieldSchemas(t *testing.T) {
	schemaNormal := &schemapb.CollectionSchema{
		Name:        "testColl",
		Description: "",
		AutoID:      false,
		Fields: []*schemapb.FieldSchema{
			{
				FieldID:      100,
				Name:         "field_int64",
				IsPrimaryKey: true,
				Description:  "",
				DataType:     5,
			},
			{
				FieldID:      107,
				Name:         "field_float_vector",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     101,
				TypeParams: []*commonpb.KeyValuePair{
					{
						Key:   common.DimKey,
						Value: "128",
					},
				},
			},
		},
	}

	t.Run("GetVectorFieldSchemas", func(t *testing.T) {
		fieldSchema := GetVectorFieldSchemas(schemaNormal)
		assert.Equal(t, 1, len(fieldSchema))
		assert.Equal(t, "field_float_vector", fieldSchema[0].Name)
	})

	schemaSparse := &schemapb.CollectionSchema{
		Name:        "testColl",
		Description: "",
		AutoID:      false,
		Fields: []*schemapb.FieldSchema{
			{
				FieldID:      100,
				Name:         "field_int64",
				IsPrimaryKey: true,
				Description:  "",
				DataType:     5,
			},
			{
				FieldID:      107,
				Name:         "field_sparse_float_vector",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     104,
				TypeParams:   []*commonpb.KeyValuePair{},
			},
		},
	}

	t.Run("GetSparseFloatVectorFieldSchema", func(t *testing.T) {
		fieldSchema := GetVectorFieldSchemas(schemaSparse)
		assert.Equal(t, 1, len(fieldSchema))
		assert.Equal(t, "field_sparse_float_vector", fieldSchema[0].Name)
	})

	schemaInvalid := &schemapb.CollectionSchema{
		Name:        "testColl",
		Description: "",
		AutoID:      false,
		Fields: []*schemapb.FieldSchema{
			{
				FieldID:      100,
				Name:         "field_int64",
				IsPrimaryKey: true,
				Description:  "",
				DataType:     5,
			},
		},
	}

	t.Run("GetVectorFieldSchemaInvalid", func(t *testing.T) {
		res := GetVectorFieldSchemas(schemaInvalid)
		assert.Equal(t, 0, len(res))
	})
}

func TestSchemaHelper_GetDynamicField(t *testing.T) {
	t.Run("with_dynamic_schema", func(t *testing.T) {
		sch := &schemapb.CollectionSchema{
			Name:        "testColl",
			Description: "",
			AutoID:      false,
			Fields: []*schemapb.FieldSchema{
				{
					FieldID:      100,
					Name:         "field_int64",
					IsPrimaryKey: true,
					DataType:     schemapb.DataType_Int64,
				},
				{
					FieldID:  101,
					Name:     "field_float_vector",
					DataType: schemapb.DataType_FloatVector,
					TypeParams: []*commonpb.KeyValuePair{
						{
							Key:   common.DimKey,
							Value: "128",
						},
					},
				},
				{
					FieldID:   102,
					Name:      "$meta",
					DataType:  schemapb.DataType_JSON,
					IsDynamic: true,
				},
			},
		}

		helper, err := CreateSchemaHelper(sch)
		require.NoError(t, err)

		f, err := helper.GetDynamicField()
		assert.NoError(t, err)
		assert.NotNil(t, f)
		assert.EqualValues(t, 102, f.FieldID)
	})

	t.Run("without_dynamic_schema", func(t *testing.T) {
		sch := &schemapb.CollectionSchema{
			Name:        "testColl",
			Description: "",
			AutoID:      false,
			Fields: []*schemapb.FieldSchema{
				{
					FieldID:      100,
					Name:         "field_int64",
					IsPrimaryKey: true,
					DataType:     schemapb.DataType_Int64,
				},
				{
					FieldID:  101,
					Name:     "field_float_vector",
					DataType: schemapb.DataType_FloatVector,
					TypeParams: []*commonpb.KeyValuePair{
						{
							Key:   common.DimKey,
							Value: "128",
						},
					},
				},
			},
		}

		helper, err := CreateSchemaHelper(sch)
		require.NoError(t, err)

		_, err = helper.GetDynamicField()
		assert.Error(t, err)
	})

	t.Run("multiple_dynamic_fields", func(t *testing.T) {
		sch := &schemapb.CollectionSchema{
			Name:        "testColl",
			Description: "",
			AutoID:      false,
			Fields: []*schemapb.FieldSchema{
				{
					FieldID:      100,
					Name:         "field_int64",
					IsPrimaryKey: true,
					DataType:     schemapb.DataType_Int64,
				},
				{
					FieldID:  101,
					Name:     "field_float_vector",
					DataType: schemapb.DataType_FloatVector,
					TypeParams: []*commonpb.KeyValuePair{
						{
							Key:   common.DimKey,
							Value: "128",
						},
					},
				},
				{
					FieldID:   102,
					Name:      "$meta",
					DataType:  schemapb.DataType_JSON,
					IsDynamic: true,
				},
				{
					FieldID:   103,
					Name:      "other_json",
					DataType:  schemapb.DataType_JSON,
					IsDynamic: true,
				},
			},
		}

		_, err := CreateSchemaHelper(sch)
		assert.Error(t, err)
	})
}

func TestSchemaHelper_GetClusteringKeyField(t *testing.T) {
	t.Run("with_clustering_key", func(t *testing.T) {
		sch := &schemapb.CollectionSchema{
			Name:        "testColl",
			Description: "",
			AutoID:      false,
			Fields: []*schemapb.FieldSchema{
				{
					FieldID:      100,
					Name:         "field_int64",
					IsPrimaryKey: true,
					DataType:     schemapb.DataType_Int64,
				},
				{
					FieldID:  101,
					Name:     "field_float_vector",
					DataType: schemapb.DataType_FloatVector,
					TypeParams: []*commonpb.KeyValuePair{
						{
							Key:   common.DimKey,
							Value: "128",
						},
					},
				},
				{
					FieldID:         102,
					Name:            "group",
					DataType:        schemapb.DataType_Int64,
					IsClusteringKey: true,
				},
			},
		}

		helper, err := CreateSchemaHelper(sch)
		require.NoError(t, err)

		f, err := helper.GetClusteringKeyField()
		assert.NoError(t, err)
		assert.NotNil(t, f)
		assert.EqualValues(t, 102, f.FieldID)
	})

	t.Run("without_clusteriny_key_schema", func(t *testing.T) {
		sch := &schemapb.CollectionSchema{
			Name:        "testColl",
			Description: "",
			AutoID:      false,
			Fields: []*schemapb.FieldSchema{
				{
					FieldID:      100,
					Name:         "field_int64",
					IsPrimaryKey: true,
					DataType:     schemapb.DataType_Int64,
				},
				{
					FieldID:  101,
					Name:     "field_float_vector",
					DataType: schemapb.DataType_FloatVector,
					TypeParams: []*commonpb.KeyValuePair{
						{
							Key:   common.DimKey,
							Value: "128",
						},
					},
				},
			},
		}

		helper, err := CreateSchemaHelper(sch)
		require.NoError(t, err)

		_, err = helper.GetClusteringKeyField()
		assert.Error(t, err)
	})

	t.Run("multiple_dynamic_fields", func(t *testing.T) {
		sch := &schemapb.CollectionSchema{
			Name:        "testColl",
			Description: "",
			AutoID:      false,
			Fields: []*schemapb.FieldSchema{
				{
					FieldID:      100,
					Name:         "field_int64",
					IsPrimaryKey: true,
					DataType:     schemapb.DataType_Int64,
				},
				{
					FieldID:  101,
					Name:     "field_float_vector",
					DataType: schemapb.DataType_FloatVector,
					TypeParams: []*commonpb.KeyValuePair{
						{
							Key:   common.DimKey,
							Value: "128",
						},
					},
				},
				{
					FieldID:         102,
					Name:            "group",
					DataType:        schemapb.DataType_Int64,
					IsClusteringKey: true,
				},
				{
					FieldID:         103,
					Name:            "batch",
					DataType:        schemapb.DataType_VarChar,
					IsClusteringKey: true,
				},
			},
		}

		_, err := CreateSchemaHelper(sch)
		assert.Error(t, err)
	})
}

func TestSchema_invalid(t *testing.T) {
	t.Run("Duplicate field name", func(t *testing.T) {
		schema := &schemapb.CollectionSchema{
			Name:        "testColl",
			Description: "",
			AutoID:      false,
			Fields: []*schemapb.FieldSchema{
				{
					FieldID:      100,
					Name:         "field_int8",
					IsPrimaryKey: false,
					Description:  "",
					DataType:     2,
				},
				{
					FieldID:      101,
					Name:         "field_int8",
					IsPrimaryKey: false,
					Description:  "",
					DataType:     3,
				},
			},
		}
		_, err := CreateSchemaHelper(schema)
		assert.Error(t, err)
		assert.ErrorIs(t, err, merr.ErrParameterInvalid)
		assert.Contains(t, err.Error(), "duplicated fieldName: field_int8")
	})
	t.Run("Duplicate field id", func(t *testing.T) {
		schema := &schemapb.CollectionSchema{
			Name:        "testColl",
			Description: "",
			AutoID:      false,
			Fields: []*schemapb.FieldSchema{
				{
					FieldID:      100,
					Name:         "field_int8",
					IsPrimaryKey: false,
					Description:  "",
					DataType:     2,
				},
				{
					FieldID:      100,
					Name:         "field_int16",
					IsPrimaryKey: false,
					Description:  "",
					DataType:     3,
				},
			},
		}
		_, err := CreateSchemaHelper(schema)
		assert.Error(t, err)
		assert.ErrorIs(t, err, merr.ErrParameterInvalid)
		assert.Contains(t, err.Error(), "duplicated fieldID: 100")
	})
	t.Run("Duplicated primary key", func(t *testing.T) {
		schema := &schemapb.CollectionSchema{
			Name:        "testColl",
			Description: "",
			AutoID:      false,
			Fields: []*schemapb.FieldSchema{
				{
					FieldID:      100,
					Name:         "field_int8",
					IsPrimaryKey: true,
					Description:  "",
					DataType:     2,
				},
				{
					FieldID:      101,
					Name:         "field_int16",
					IsPrimaryKey: true,
					Description:  "",
					DataType:     3,
				},
			},
		}
		_, err := CreateSchemaHelper(schema)
		assert.Error(t, err)
		assert.ErrorIs(t, err, merr.ErrParameterInvalid)
		assert.Contains(t, err.Error(), "primary key is not unique")
	})
	t.Run("field not exist", func(t *testing.T) {
		schema := &schemapb.CollectionSchema{
			Name:        "testColl",
			Description: "",
			AutoID:      false,
			Fields: []*schemapb.FieldSchema{
				{
					FieldID:      100,
					Name:         "field_int8",
					IsPrimaryKey: false,
					Description:  "",
					DataType:     2,
				},
			},
		}
		helper, err := CreateSchemaHelper(schema)
		assert.NoError(t, err)

		_, err = helper.GetPrimaryKeyField()
		assert.Error(t, err)
		assert.ErrorIs(t, err, merr.ErrParameterInvalid)
		assert.Contains(t, err.Error(), "failed to get primary key field: no primary in schema")

		_, err = helper.GetFieldFromName("none")
		assert.Error(t, err)
		assert.ErrorIs(t, err, merr.ErrParameterInvalid)
		assert.Contains(t, err.Error(), "failed to get field schema by name: fieldName(none) not found")

		_, err = helper.GetFieldFromID(101)
		assert.Error(t, err)
		assert.ErrorIs(t, err, merr.ErrParameterInvalid)
		assert.Contains(t, err.Error(), "fieldID(101) not found")
	})
	t.Run("vector dim not exist", func(t *testing.T) {
		schema := &schemapb.CollectionSchema{
			Name:        "testColl",
			Description: "",
			AutoID:      false,
			Fields: []*schemapb.FieldSchema{
				{
					FieldID:      103,
					Name:         "field_int64",
					IsPrimaryKey: true,
					Description:  "",
					DataType:     5,
				},
				{
					FieldID:      107,
					Name:         "field_float_vector",
					IsPrimaryKey: false,
					Description:  "",
					DataType:     101,
				},
			},
		}
		helper, err := CreateSchemaHelper(schema)
		assert.NoError(t, err)

		_, err = helper.GetVectorDimFromID(100)
		assert.Error(t, err)

		_, err = helper.GetVectorDimFromID(103)
		assert.Error(t, err)

		_, err = helper.GetVectorDimFromID(107)
		assert.Error(t, err)
	})
}

func genFieldData(fieldName string, fieldID int64, fieldType schemapb.DataType, fieldValue interface{}, dim int64) *schemapb.FieldData {
	var fieldData *schemapb.FieldData
	switch fieldType {
	case schemapb.DataType_Bool:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_Bool,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_BoolData{
						BoolData: &schemapb.BoolArray{
							Data: fieldValue.([]bool),
						},
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_Int8:
		data := []int32{}
		for _, v := range fieldValue.([]int8) {
			data = append(data, int32(v))
		}
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_Int8,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{
							Data: data,
						},
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_Int16:
		data := []int32{}
		for _, v := range fieldValue.([]int16) {
			data = append(data, int32(v))
		}
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_Int16,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{
							Data: data,
						},
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_Int32:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_Int32,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{
							Data: fieldValue.([]int32),
						},
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_Int64:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_Int64,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_LongData{
						LongData: &schemapb.LongArray{
							Data: fieldValue.([]int64),
						},
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_Float:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_Float,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_FloatData{
						FloatData: &schemapb.FloatArray{
							Data: fieldValue.([]float32),
						},
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_Double:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_Double,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_DoubleData{
						DoubleData: &schemapb.DoubleArray{
							Data: fieldValue.([]float64),
						},
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_String:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_String,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_StringData{
						StringData: &schemapb.StringArray{
							Data: fieldValue.([]string),
						},
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_VarChar:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_VarChar,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_StringData{
						StringData: &schemapb.StringArray{
							Data: fieldValue.([]string),
						},
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_BinaryVector:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_BinaryVector,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_BinaryVector{
						BinaryVector: fieldValue.([]byte),
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_FloatVector:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_FloatVector,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_FloatVector{
						FloatVector: &schemapb.FloatArray{
							Data: fieldValue.([]float32),
						},
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_Float16Vector:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_Float16Vector,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Float16Vector{
						Float16Vector: fieldValue.([]byte),
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_BFloat16Vector:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_BFloat16Vector,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Bfloat16Vector{
						Bfloat16Vector: fieldValue.([]byte),
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_SparseFloatVector:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_SparseFloatVector,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_SparseFloatVector{
						SparseFloatVector: &schemapb.SparseFloatArray{
							Dim:      dim,
							Contents: [][]byte{fieldValue.([]byte)},
						},
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_Int8Vector:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_Int8Vector,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Int8Vector{
						Int8Vector: fieldValue.([]byte),
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_Array:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_Array,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_ArrayData{
						ArrayData: &schemapb.ArrayArray{
							Data:        fieldValue.([]*schemapb.ScalarField),
							ElementType: schemapb.DataType_Int32,
						},
					},
				},
			},
			FieldId: fieldID,
		}

	case schemapb.DataType_JSON:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_JSON,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_JsonData{
						JsonData: &schemapb.JSONArray{
							Data: fieldValue.([][]byte),
						},
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_ArrayOfVector:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_ArrayOfVector,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_VectorArray{
						VectorArray: &schemapb.VectorArray{
							Data:        fieldValue.([]*schemapb.VectorField),
							ElementType: schemapb.DataType_FloatVector,
						},
					},
				},
			},
			FieldId: fieldID,
		}
	case schemapb.DataType_Geometry:
		fieldData = &schemapb.FieldData{
			Type:      schemapb.DataType_Geometry,
			FieldName: fieldName,
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_GeometryData{
						GeometryData: &schemapb.GeometryArray{
							Data: fieldValue.([][]byte),
						},
					},
				},
			},
			FieldId: fieldID,
		}
	default:
		fmt.Printf("not supported field type: %s", fieldType.String())
	}

	return fieldData
}

func TestAppendFieldData(t *testing.T) {
	const (
		Dim                        = 8
		BoolFieldName              = "BoolField"
		Int32FieldName             = "Int32Field"
		Int64FieldName             = "Int64Field"
		FloatFieldName             = "FloatField"
		DoubleFieldName            = "DoubleField"
		BinaryVectorFieldName      = "BinaryVectorField"
		FloatVectorFieldName       = "FloatVectorField"
		Float16VectorFieldName     = "Float16VectorField"
		BFloat16VectorFieldName    = "BFloat16VectorField"
		ArrayFieldName             = "ArrayField"
		SparseFloatVectorFieldName = "SparseFloatVectorField"
		Int8VectorFieldName        = "Int8VectorField"
		VectorArrayFieldName       = "VectorArrayField"
		GeometryFieldName          = "GeometryField"
		BoolFieldID                = common.StartOfUserFieldID + 1
		Int32FieldID               = common.StartOfUserFieldID + 2
		Int64FieldID               = common.StartOfUserFieldID + 3
		FloatFieldID               = common.StartOfUserFieldID + 4
		DoubleFieldID              = common.StartOfUserFieldID + 5
		BinaryVectorFieldID        = common.StartOfUserFieldID + 6
		FloatVectorFieldID         = common.StartOfUserFieldID + 7
		Float16VectorFieldID       = common.StartOfUserFieldID + 8
		BFloat16VectorFieldID      = common.StartOfUserFieldID + 9
		ArrayFieldID               = common.StartOfUserFieldID + 10
		SparseFloatVectorFieldID   = common.StartOfUserFieldID + 11
		Int8VectorFieldID          = common.StartOfUserFieldID + 12
		VectorArrayFieldID         = common.StartOfUserFieldID + 13
		GeometryFieldID            = common.StartOfUserFieldID + 14
	)
	BoolArray := []bool{true, false}
	Int32Array := []int32{1, 2}
	Int64Array := []int64{11, 22}
	FloatArray := []float32{1.0, 2.0}
	DoubleArray := []float64{11.0, 22.0}
	BinaryVector := []byte{0x12, 0x34}
	FloatVector := []float32{1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 11.0, 22.0, 33.0, 44.0, 55.0, 66.0, 77.0, 88.0}
	Float16Vector := []byte{
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff,
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff,
	}
	BFloat16Vector := []byte{
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff,
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff,
	}
	Int8Vector := []byte{
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff,
	}
	ArrayArray := []*schemapb.ScalarField{
		{
			Data: &schemapb.ScalarField_IntData{
				IntData: &schemapb.IntArray{
					Data: []int32{1, 2, 3},
				},
			},
		},
		{
			Data: &schemapb.ScalarField_IntData{
				IntData: &schemapb.IntArray{
					Data: []int32{4, 5, 6},
				},
			},
		},
	}
	SparseFloatVector := &schemapb.SparseFloatArray{
		Dim: 231,
		Contents: [][]byte{
			CreateSparseFloatRow([]uint32{}, []float32{}),
			CreateSparseFloatRow([]uint32{60, 80, 230}, []float32{2.1, 2.2, 2.3}),
		},
	}
	VectorArray := []*schemapb.VectorField{
		{
			Dim: Dim,
			Data: &schemapb.VectorField_FloatVector{
				FloatVector: &schemapb.FloatArray{
					Data: FloatVector,
				},
			},
		},
		{
			Dim: Dim,
			Data: &schemapb.VectorField_FloatVector{
				FloatVector: &schemapb.FloatArray{
					Data: FloatVector,
				},
			},
		},
	}

	result := make([]*schemapb.FieldData, 14)
	// POINT (30.123 -10.456) and LINESTRING (30.123 -10.456, 10.789 30.123, -40.567 40.890)
	GeometryArray := [][]byte{
		{0x01, 0x01, 0x00, 0x00, 0x00, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A, 0x0D, 0x1B, 0x4F, 0x4F, 0x9A, 0x3D, 0x40},
		{0x01, 0x02, 0x00, 0x00, 0x00, 0x03, 0x00, 0x00, 0x00, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A, 0x0D, 0x1B, 0x4F, 0x4F, 0x9A, 0x3D, 0x40, 0x03, 0xA6, 0xB4, 0xA6, 0xA4, 0xD2, 0xC5, 0xC0, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A},
	}

	var fieldDataArray1 []*schemapb.FieldData
	fieldDataArray1 = append(fieldDataArray1, genFieldData(BoolFieldName, BoolFieldID, schemapb.DataType_Bool, BoolArray[0:1], 1))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(Int32FieldName, Int32FieldID, schemapb.DataType_Int32, Int32Array[0:1], 1))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(Int64FieldName, Int64FieldID, schemapb.DataType_Int64, Int64Array[0:1], 1))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(FloatFieldName, FloatFieldID, schemapb.DataType_Float, FloatArray[0:1], 1))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(DoubleFieldName, DoubleFieldID, schemapb.DataType_Double, DoubleArray[0:1], 1))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(BinaryVectorFieldName, BinaryVectorFieldID, schemapb.DataType_BinaryVector, BinaryVector[0:Dim/8], Dim))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(FloatVectorFieldName, FloatVectorFieldID, schemapb.DataType_FloatVector, FloatVector[0:Dim], Dim))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(Float16VectorFieldName, Float16VectorFieldID, schemapb.DataType_Float16Vector, Float16Vector[0:Dim*2], Dim))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(BFloat16VectorFieldName, BFloat16VectorFieldID, schemapb.DataType_BFloat16Vector, BFloat16Vector[0:Dim*2], Dim))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(ArrayFieldName, ArrayFieldID, schemapb.DataType_Array, ArrayArray[0:1], 1))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(SparseFloatVectorFieldName, SparseFloatVectorFieldID, schemapb.DataType_SparseFloatVector, SparseFloatVector.Contents[0], SparseFloatVector.Dim))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(Int8VectorFieldName, Int8VectorFieldID, schemapb.DataType_Int8Vector, Int8Vector[0:Dim], Dim))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(VectorArrayFieldName, VectorArrayFieldID, schemapb.DataType_ArrayOfVector, VectorArray[0:1], Dim))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(GeometryFieldName, GeometryFieldID, schemapb.DataType_Geometry, GeometryArray[0:1], 1))

	var fieldDataArray2 []*schemapb.FieldData
	fieldDataArray2 = append(fieldDataArray2, genFieldData(BoolFieldName, BoolFieldID, schemapb.DataType_Bool, BoolArray[1:2], 1))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(Int32FieldName, Int32FieldID, schemapb.DataType_Int32, Int32Array[1:2], 1))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(Int64FieldName, Int64FieldID, schemapb.DataType_Int64, Int64Array[1:2], 1))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(FloatFieldName, FloatFieldID, schemapb.DataType_Float, FloatArray[1:2], 1))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(DoubleFieldName, DoubleFieldID, schemapb.DataType_Double, DoubleArray[1:2], 1))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(BinaryVectorFieldName, BinaryVectorFieldID, schemapb.DataType_BinaryVector, BinaryVector[Dim/8:2*Dim/8], Dim))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(FloatVectorFieldName, FloatVectorFieldID, schemapb.DataType_FloatVector, FloatVector[Dim:2*Dim], Dim))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(Float16VectorFieldName, Float16VectorFieldID, schemapb.DataType_Float16Vector, Float16Vector[2*Dim:4*Dim], Dim))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(BFloat16VectorFieldName, BFloat16VectorFieldID, schemapb.DataType_BFloat16Vector, BFloat16Vector[2*Dim:4*Dim], Dim))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(ArrayFieldName, ArrayFieldID, schemapb.DataType_Array, ArrayArray[1:2], 1))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(SparseFloatVectorFieldName, SparseFloatVectorFieldID, schemapb.DataType_SparseFloatVector, SparseFloatVector.Contents[1], SparseFloatVector.Dim))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(Int8VectorFieldName, Int8VectorFieldID, schemapb.DataType_Int8Vector, Int8Vector[Dim:2*Dim], Dim))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(VectorArrayFieldName, VectorArrayFieldID, schemapb.DataType_ArrayOfVector, VectorArray[1:2], Dim))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(GeometryFieldName, GeometryFieldID, schemapb.DataType_Geometry, GeometryArray[1:2], 1))

	AppendFieldData(result, fieldDataArray1, 0)
	AppendFieldData(result, fieldDataArray2, 0)

	assert.Equal(t, BoolArray, result[0].GetScalars().GetBoolData().Data)
	assert.Equal(t, Int32Array, result[1].GetScalars().GetIntData().Data)
	assert.Equal(t, Int64Array, result[2].GetScalars().GetLongData().Data)
	assert.Equal(t, FloatArray, result[3].GetScalars().GetFloatData().Data)
	assert.Equal(t, DoubleArray, result[4].GetScalars().GetDoubleData().Data)
	assert.Equal(t, BinaryVector, result[5].GetVectors().Data.(*schemapb.VectorField_BinaryVector).BinaryVector)
	assert.Equal(t, FloatVector, result[6].GetVectors().GetFloatVector().Data)
	assert.Equal(t, Float16Vector, result[7].GetVectors().Data.(*schemapb.VectorField_Float16Vector).Float16Vector)
	assert.Equal(t, BFloat16Vector, result[8].GetVectors().Data.(*schemapb.VectorField_Bfloat16Vector).Bfloat16Vector)
	assert.Equal(t, ArrayArray, result[9].GetScalars().GetArrayData().Data)
	assert.Equal(t, SparseFloatVector, result[10].GetVectors().GetSparseFloatVector())
	assert.Equal(t, Int8Vector, result[11].GetVectors().Data.(*schemapb.VectorField_Int8Vector).Int8Vector)
	assert.Equal(t, VectorArray, result[12].GetVectors().GetVectorArray().Data)
	assert.Equal(t, GeometryArray, result[13].GetScalars().GetGeometryData().Data)
}

func TestDeleteFieldData(t *testing.T) {
	const (
		Dim                        = 8
		BoolFieldName              = "BoolField"
		Int32FieldName             = "Int32Field"
		Int64FieldName             = "Int64Field"
		FloatFieldName             = "FloatField"
		DoubleFieldName            = "DoubleField"
		JSONFieldName              = "JSONField"
		GeometryFieldName          = "GeometryField"
		BinaryVectorFieldName      = "BinaryVectorField"
		FloatVectorFieldName       = "FloatVectorField"
		Float16VectorFieldName     = "Float16VectorField"
		BFloat16VectorFieldName    = "BFloat16VectorField"
		SparseFloatVectorFieldName = "SparseFloatVectorField"
		Int8VectorFieldName        = "Int8VectorField"
	)

	const (
		BoolFieldID = common.StartOfUserFieldID + iota
		Int32FieldID
		Int64FieldID
		FloatFieldID
		DoubleFieldID
		JSONFieldID
		GeometryFiledID
		BinaryVectorFieldID
		FloatVectorFieldID
		Float16VectorFieldID
		BFloat16VectorFieldID
		SparseFloatVectorFieldID
		Int8VectorFieldID
	)
	BoolArray := []bool{true, false}
	Int32Array := []int32{1, 2}
	Int64Array := []int64{11, 22}
	FloatArray := []float32{1.0, 2.0}
	DoubleArray := []float64{11.0, 22.0}
	JSONArray := [][]byte{[]byte("{\"hello\":0}"), []byte("{\"key\":1}")}
	// POINT (30.123 -10.456) and LINESTRING (30.123 -10.456, 10.789 30.123, -40.567 40.890)
	GeometryArray := [][]byte{
		{0x01, 0x01, 0x00, 0x00, 0x00, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A, 0x0D, 0x1B, 0x4F, 0x4F, 0x9A, 0x3D, 0x40},
		{0x01, 0x02, 0x00, 0x00, 0x00, 0x03, 0x00, 0x00, 0x00, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A, 0x0D, 0x1B, 0x4F, 0x4F, 0x9A, 0x3D, 0x40, 0x03, 0xA6, 0xB4, 0xA6, 0xA4, 0xD2, 0xC5, 0xC0, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A},
	}
	BinaryVector := []byte{0x12, 0x34}
	FloatVector := []float32{1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 11.0, 22.0, 33.0, 44.0, 55.0, 66.0, 77.0, 88.0}
	Float16Vector := []byte{
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff,
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff,
	}
	BFloat16Vector := []byte{
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff,
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff,
	}
	SparseFloatVector := &schemapb.SparseFloatArray{
		Dim: 231,
		Contents: [][]byte{
			CreateSparseFloatRow([]uint32{30, 41, 52}, []float32{1.1, 1.2, 1.3}),
			CreateSparseFloatRow([]uint32{60, 80, 230}, []float32{2.1, 2.2, 2.3}),
		},
	}
	Int8Vector := []byte{
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff,
	}

	result1 := make([]*schemapb.FieldData, 13)
	result2 := make([]*schemapb.FieldData, 13)
	var fieldDataArray1 []*schemapb.FieldData
	fieldDataArray1 = append(fieldDataArray1, genFieldData(BoolFieldName, BoolFieldID, schemapb.DataType_Bool, BoolArray[0:1], 1))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(Int32FieldName, Int32FieldID, schemapb.DataType_Int32, Int32Array[0:1], 1))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(Int64FieldName, Int64FieldID, schemapb.DataType_Int64, Int64Array[0:1], 1))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(FloatFieldName, FloatFieldID, schemapb.DataType_Float, FloatArray[0:1], 1))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(DoubleFieldName, DoubleFieldID, schemapb.DataType_Double, DoubleArray[0:1], 1))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(JSONFieldName, JSONFieldID, schemapb.DataType_JSON, JSONArray[0:1], 1))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(GeometryFieldName, GeometryFiledID, schemapb.DataType_Geometry, GeometryArray[0:1], 1))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(BinaryVectorFieldName, BinaryVectorFieldID, schemapb.DataType_BinaryVector, BinaryVector[0:Dim/8], Dim))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(FloatVectorFieldName, FloatVectorFieldID, schemapb.DataType_FloatVector, FloatVector[0:Dim], Dim))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(Float16VectorFieldName, Float16VectorFieldID, schemapb.DataType_Float16Vector, Float16Vector[0:2*Dim], Dim))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(BFloat16VectorFieldName, BFloat16VectorFieldID, schemapb.DataType_BFloat16Vector, BFloat16Vector[0:2*Dim], Dim))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(SparseFloatVectorFieldName, SparseFloatVectorFieldID, schemapb.DataType_SparseFloatVector, SparseFloatVector.Contents[0], SparseFloatVector.Dim))
	fieldDataArray1 = append(fieldDataArray1, genFieldData(Int8VectorFieldName, Int8VectorFieldID, schemapb.DataType_Int8Vector, Int8Vector[0:Dim], Dim))

	var fieldDataArray2 []*schemapb.FieldData
	fieldDataArray2 = append(fieldDataArray2, genFieldData(BoolFieldName, BoolFieldID, schemapb.DataType_Bool, BoolArray[1:2], 1))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(Int32FieldName, Int32FieldID, schemapb.DataType_Int32, Int32Array[1:2], 1))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(Int64FieldName, Int64FieldID, schemapb.DataType_Int64, Int64Array[1:2], 1))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(FloatFieldName, FloatFieldID, schemapb.DataType_Float, FloatArray[1:2], 1))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(DoubleFieldName, DoubleFieldID, schemapb.DataType_Double, DoubleArray[1:2], 1))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(JSONFieldName, JSONFieldID, schemapb.DataType_JSON, JSONArray[1:2], 1))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(GeometryFieldName, GeometryFiledID, schemapb.DataType_Geometry, GeometryArray[1:2], 1))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(BinaryVectorFieldName, BinaryVectorFieldID, schemapb.DataType_BinaryVector, BinaryVector[Dim/8:2*Dim/8], Dim))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(FloatVectorFieldName, FloatVectorFieldID, schemapb.DataType_FloatVector, FloatVector[Dim:2*Dim], Dim))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(Float16VectorFieldName, Float16VectorFieldID, schemapb.DataType_Float16Vector, Float16Vector[2*Dim:4*Dim], Dim))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(BFloat16VectorFieldName, BFloat16VectorFieldID, schemapb.DataType_BFloat16Vector, BFloat16Vector[2*Dim:4*Dim], Dim))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(SparseFloatVectorFieldName, SparseFloatVectorFieldID, schemapb.DataType_SparseFloatVector, SparseFloatVector.Contents[1], SparseFloatVector.Dim))
	fieldDataArray2 = append(fieldDataArray2, genFieldData(Int8VectorFieldName, Int8VectorFieldID, schemapb.DataType_Int8Vector, Int8Vector[Dim:2*Dim], Dim))

	AppendFieldData(result1, fieldDataArray1, 0)
	AppendFieldData(result1, fieldDataArray2, 0)
	DeleteFieldData(result1)
	assert.Equal(t, BoolArray[0:1], result1[BoolFieldID-common.StartOfUserFieldID].GetScalars().GetBoolData().Data)
	assert.Equal(t, Int32Array[0:1], result1[Int32FieldID-common.StartOfUserFieldID].GetScalars().GetIntData().Data)
	assert.Equal(t, Int64Array[0:1], result1[Int64FieldID-common.StartOfUserFieldID].GetScalars().GetLongData().Data)
	assert.Equal(t, FloatArray[0:1], result1[FloatFieldID-common.StartOfUserFieldID].GetScalars().GetFloatData().Data)
	assert.Equal(t, DoubleArray[0:1], result1[DoubleFieldID-common.StartOfUserFieldID].GetScalars().GetDoubleData().Data)
	assert.Equal(t, JSONArray[0:1], result1[JSONFieldID-common.StartOfUserFieldID].GetScalars().GetJsonData().Data)
	assert.Equal(t, GeometryArray[0:1], result1[GeometryFiledID-common.StartOfUserFieldID].GetScalars().GetGeometryData().Data)
	assert.Equal(t, BinaryVector[0:Dim/8], result1[BinaryVectorFieldID-common.StartOfUserFieldID].GetVectors().Data.(*schemapb.VectorField_BinaryVector).BinaryVector)
	assert.Equal(t, FloatVector[0:Dim], result1[FloatVectorFieldID-common.StartOfUserFieldID].GetVectors().GetFloatVector().Data)
	assert.Equal(t, Float16Vector[0:2*Dim], result1[Float16VectorFieldID-common.StartOfUserFieldID].GetVectors().Data.(*schemapb.VectorField_Float16Vector).Float16Vector)
	assert.Equal(t, BFloat16Vector[0:2*Dim], result1[BFloat16VectorFieldID-common.StartOfUserFieldID].GetVectors().Data.(*schemapb.VectorField_Bfloat16Vector).Bfloat16Vector)
	tmpSparseFloatVector := proto.Clone(SparseFloatVector).(*schemapb.SparseFloatArray)
	tmpSparseFloatVector.Contents = [][]byte{SparseFloatVector.Contents[0]}
	assert.Equal(t, tmpSparseFloatVector.Contents, result1[SparseFloatVectorFieldID-common.StartOfUserFieldID].GetVectors().GetSparseFloatVector().Contents)
	assert.Equal(t, Int8Vector[0:Dim], result1[Int8VectorFieldID-common.StartOfUserFieldID].GetVectors().Data.(*schemapb.VectorField_Int8Vector).Int8Vector)

	AppendFieldData(result2, fieldDataArray2, 0)
	AppendFieldData(result2, fieldDataArray1, 0)
	DeleteFieldData(result2)
	assert.Equal(t, BoolArray[1:2], result2[BoolFieldID-common.StartOfUserFieldID].GetScalars().GetBoolData().Data)
	assert.Equal(t, Int32Array[1:2], result2[Int32FieldID-common.StartOfUserFieldID].GetScalars().GetIntData().Data)
	assert.Equal(t, Int64Array[1:2], result2[Int64FieldID-common.StartOfUserFieldID].GetScalars().GetLongData().Data)
	assert.Equal(t, FloatArray[1:2], result2[FloatFieldID-common.StartOfUserFieldID].GetScalars().GetFloatData().Data)
	assert.Equal(t, DoubleArray[1:2], result2[DoubleFieldID-common.StartOfUserFieldID].GetScalars().GetDoubleData().Data)
	assert.Equal(t, JSONArray[1:2], result2[JSONFieldID-common.StartOfUserFieldID].GetScalars().GetJsonData().Data)
	assert.Equal(t, GeometryArray[1:2], result2[GeometryFiledID-common.StartOfUserFieldID].GetScalars().GetGeometryData().Data)
	assert.Equal(t, BinaryVector[Dim/8:2*Dim/8], result2[BinaryVectorFieldID-common.StartOfUserFieldID].GetVectors().Data.(*schemapb.VectorField_BinaryVector).BinaryVector)
	assert.Equal(t, FloatVector[Dim:2*Dim], result2[FloatVectorFieldID-common.StartOfUserFieldID].GetVectors().GetFloatVector().Data)
	assert.Equal(t, Float16Vector[2*Dim:4*Dim], result2[Float16VectorFieldID-common.StartOfUserFieldID].GetVectors().Data.(*schemapb.VectorField_Float16Vector).Float16Vector)
	assert.Equal(t, BFloat16Vector[2*Dim:4*Dim], result2[BFloat16VectorFieldID-common.StartOfUserFieldID].GetVectors().Data.(*schemapb.VectorField_Bfloat16Vector).Bfloat16Vector)
	tmpSparseFloatVector = proto.Clone(SparseFloatVector).(*schemapb.SparseFloatArray)
	tmpSparseFloatVector.Contents = [][]byte{SparseFloatVector.Contents[1]}
	assert.EqualExportedValues(t, tmpSparseFloatVector, result2[SparseFloatVectorFieldID-common.StartOfUserFieldID].GetVectors().GetSparseFloatVector())
	assert.Equal(t, Int8Vector[Dim:2*Dim], result2[Int8VectorFieldID-common.StartOfUserFieldID].GetVectors().Data.(*schemapb.VectorField_Int8Vector).Int8Vector)
}

func TestEstimateEntitySize(t *testing.T) {
	samples := []*schemapb.FieldData{
		{
			FieldId:   111,
			FieldName: "float16_vector",
			Type:      schemapb.DataType_Float16Vector,
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim:  64,
					Data: &schemapb.VectorField_Float16Vector{},
				},
			},
		},
		{
			FieldId:   112,
			FieldName: "bfloat16_vector",
			Type:      schemapb.DataType_BFloat16Vector,
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim:  128,
					Data: &schemapb.VectorField_Bfloat16Vector{},
				},
			},
		},
	}
	size, error := EstimateEntitySize(samples, int(0))
	assert.NoError(t, error)
	assert.True(t, size == 384)
}

func TestGetPrimaryFieldSchema(t *testing.T) {
	int64Field := &schemapb.FieldSchema{
		FieldID:  1,
		Name:     "int64Field",
		DataType: schemapb.DataType_Int64,
	}

	floatField := &schemapb.FieldSchema{
		FieldID:  2,
		Name:     "floatField",
		DataType: schemapb.DataType_Float,
	}

	schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{int64Field, floatField},
	}

	// no primary field error
	_, err := GetPrimaryFieldSchema(schema)
	assert.Error(t, err)
	int64Field.IsPrimaryKey = true
	primaryField, err := GetPrimaryFieldSchema(schema)
	assert.NoError(t, err)
	assert.Equal(t, schemapb.DataType_Int64, primaryField.DataType)

	hasPartitionKey := HasPartitionKey(schema)
	assert.False(t, hasPartitionKey)
	int64Field.IsPartitionKey = true
	hasPartitionKey2 := HasPartitionKey(schema)
	assert.True(t, hasPartitionKey2)
}

func TestGetClusterKeyFieldSchema(t *testing.T) {
	int64Field := &schemapb.FieldSchema{
		FieldID:  1,
		Name:     "int64Field",
		DataType: schemapb.DataType_Int64,
	}

	clusterKeyfloatField := &schemapb.FieldSchema{
		FieldID:         2,
		Name:            "floatField",
		DataType:        schemapb.DataType_Float,
		IsClusteringKey: true,
	}

	unClusterKeyfloatField := &schemapb.FieldSchema{
		FieldID:         2,
		Name:            "floatField",
		DataType:        schemapb.DataType_Float,
		IsClusteringKey: false,
	}

	schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{int64Field, clusterKeyfloatField},
	}
	schema2 := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{int64Field, unClusterKeyfloatField},
	}

	hasClusterKey1 := HasClusterKey(schema)
	assert.True(t, hasClusterKey1)
	hasClusterKey2 := HasClusterKey(schema2)
	assert.False(t, hasClusterKey2)
}

func TestGetPK(t *testing.T) {
	type args struct {
		data *schemapb.IDs
		idx  int64
	}
	tests := []struct {
		name string
		args args
		want interface{}
	}{
		{
			args: args{
				data: &schemapb.IDs{
					IdField: &schemapb.IDs_IntId{
						IntId: &schemapb.LongArray{
							Data: []int64{1, 2, 3},
						},
					},
				},
				idx: 5, // > len(data)
			},
			want: nil,
		},
		{
			args: args{
				data: &schemapb.IDs{
					IdField: &schemapb.IDs_IntId{
						IntId: &schemapb.LongArray{
							Data: []int64{1, 2, 3},
						},
					},
				},
				idx: 1,
			},
			want: int64(2),
		},
		{
			args: args{
				data: &schemapb.IDs{
					IdField: &schemapb.IDs_StrId{
						StrId: &schemapb.StringArray{
							Data: []string{"1", "2", "3"},
						},
					},
				},
				idx: 1,
			},
			want: "2",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := GetPK(tt.args.data, tt.args.idx); !reflect.DeepEqual(got, tt.want) {
				t.Errorf("GetPK(%v, %v) = %v, want %v", tt.args.data, tt.args.idx, got, tt.want)
			}
		})
	}
}

func TestAppendPKs(t *testing.T) {
	intPks := &schemapb.IDs{}
	AppendPKs(intPks, int64(1))
	assert.ElementsMatch(t, []int64{1}, intPks.GetIntId().GetData())
	AppendPKs(intPks, int64(2))
	assert.ElementsMatch(t, []int64{1, 2}, intPks.GetIntId().GetData())

	strPks := &schemapb.IDs{}
	AppendPKs(strPks, "1")
	assert.ElementsMatch(t, []string{"1"}, strPks.GetStrId().GetData())
	AppendPKs(strPks, "2")
	assert.ElementsMatch(t, []string{"1", "2"}, strPks.GetStrId().GetData())
}

func TestSwapPK(t *testing.T) {
	intPks := &schemapb.IDs{}
	AppendPKs(intPks, int64(1))
	AppendPKs(intPks, int64(2))
	AppendPKs(intPks, int64(3))
	require.Equal(t, []int64{1, 2, 3}, intPks.GetIntId().GetData())

	SwapPK(intPks, 0, 1)
	assert.Equal(t, []int64{2, 1, 3}, intPks.GetIntId().GetData())
	SwapPK(intPks, 0, 1)
	assert.Equal(t, []int64{1, 2, 3}, intPks.GetIntId().GetData())
	SwapPK(intPks, 0, 2)
	assert.Equal(t, []int64{3, 2, 1}, intPks.GetIntId().GetData())
	SwapPK(intPks, 0, 2)
	assert.Equal(t, []int64{1, 2, 3}, intPks.GetIntId().GetData())
	SwapPK(intPks, 1, 2)
	assert.Equal(t, []int64{1, 3, 2}, intPks.GetIntId().GetData())
	SwapPK(intPks, 1, 2)
	assert.Equal(t, []int64{1, 2, 3}, intPks.GetIntId().GetData())

	strPks := &schemapb.IDs{}
	AppendPKs(strPks, "1")
	AppendPKs(strPks, "2")
	AppendPKs(strPks, "3")

	require.Equal(t, []string{"1", "2", "3"}, strPks.GetStrId().GetData())

	SwapPK(strPks, 0, 1)
	assert.Equal(t, []string{"2", "1", "3"}, strPks.GetStrId().GetData())
	SwapPK(strPks, 0, 1)
	assert.Equal(t, []string{"1", "2", "3"}, strPks.GetStrId().GetData())
	SwapPK(strPks, 0, 2)
	assert.Equal(t, []string{"3", "2", "1"}, strPks.GetStrId().GetData())
	SwapPK(strPks, 0, 2)
	assert.Equal(t, []string{"1", "2", "3"}, strPks.GetStrId().GetData())
	SwapPK(strPks, 1, 2)
	assert.Equal(t, []string{"1", "3", "2"}, strPks.GetStrId().GetData())
	SwapPK(strPks, 1, 2)
	assert.Equal(t, []string{"1", "2", "3"}, strPks.GetStrId().GetData())
}

func TestComparePk(t *testing.T) {
	intPks := &schemapb.IDs{}
	AppendPKs(intPks, int64(1))
	AppendPKs(intPks, int64(2))
	AppendPKs(intPks, int64(3))
	require.Equal(t, []int64{1, 2, 3}, intPks.GetIntId().GetData())

	less := ComparePKInSlice(intPks, 0, 1)
	assert.True(t, less)
	less = ComparePKInSlice(intPks, 0, 2)
	assert.True(t, less)
	less = ComparePKInSlice(intPks, 1, 2)
	assert.True(t, less)

	less = ComparePKInSlice(intPks, 1, 0)
	assert.False(t, less)
	less = ComparePKInSlice(intPks, 2, 0)
	assert.False(t, less)
	less = ComparePKInSlice(intPks, 2, 1)
	assert.False(t, less)

	strPks := &schemapb.IDs{}
	AppendPKs(strPks, "1")
	AppendPKs(strPks, "2")
	AppendPKs(strPks, "3")

	require.Equal(t, []string{"1", "2", "3"}, strPks.GetStrId().GetData())

	less = ComparePKInSlice(strPks, 0, 1)
	assert.True(t, less)
	less = ComparePKInSlice(strPks, 0, 2)
	assert.True(t, less)
	less = ComparePKInSlice(strPks, 1, 2)
	assert.True(t, less)

	less = ComparePKInSlice(strPks, 1, 0)
	assert.False(t, less)
	less = ComparePKInSlice(strPks, 2, 0)
	assert.False(t, less)
	less = ComparePKInSlice(strPks, 2, 1)
	assert.False(t, less)
}

func TestCalcColumnSize(t *testing.T) {
	fieldValues := map[int64]any{
		100: []int8{0, 1},
		101: []int16{0, 1},
		102: []int32{0, 1},
		103: []int64{0, 1},
		104: []float32{0, 1},
		105: []float64{0, 1},
		106: []string{"0", "1"},
		107: []float32{0, 1, 2, 3},
		109: []*schemapb.ScalarField{
			{
				Data: &schemapb.ScalarField_IntData{
					IntData: &schemapb.IntArray{
						Data: []int32{1, 2, 3},
					},
				},
			},
			{
				Data: &schemapb.ScalarField_IntData{
					IntData: &schemapb.IntArray{
						Data: []int32{4, 5, 6},
					},
				},
			},
		},
		110: [][]byte{[]byte(`{"key":"value"}`), []byte(`{"hello":"world"}`)},
		111: [][]byte{
			// POINT (30.123 -10.456) and LINESTRING (30.123 -10.456, 10.789 30.123, -40.567 40.890)
			{0x01, 0x01, 0x00, 0x00, 0x00, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A, 0x0D, 0x1B, 0x4F, 0x4F, 0x9A, 0x3D, 0x40},
			{0x01, 0x02, 0x00, 0x00, 0x00, 0x03, 0x00, 0x00, 0x00, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A, 0x0D, 0x1B, 0x4F, 0x4F, 0x9A, 0x3D, 0x40, 0x03, 0xA6, 0xB4, 0xA6, 0xA4, 0xD2, 0xC5, 0xC0, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A},
		},
	}
	schema := &schemapb.CollectionSchema{
		Name:        "testColl",
		Description: "",
		AutoID:      false,
		Fields: []*schemapb.FieldSchema{
			{
				FieldID:      100,
				Name:         "field_int8",
				IsPrimaryKey: false,
				DataType:     schemapb.DataType_Int8,
			},
			{
				FieldID:      101,
				Name:         "field_int16",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_Int16,
			},
			{
				FieldID:      102,
				Name:         "field_int32",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_Int32,
			},
			{
				FieldID:      103,
				Name:         "field_int64",
				IsPrimaryKey: true,
				Description:  "",
				DataType:     schemapb.DataType_Int64,
			},
			{
				FieldID:      104,
				Name:         "field_float",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_Float,
			},
			{
				FieldID:      105,
				Name:         "field_double",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_Double,
			},
			{
				FieldID:      106,
				Name:         "field_string",
				IsPrimaryKey: false,
				Description:  "",
				DataType:     schemapb.DataType_VarChar,
				TypeParams: []*commonpb.KeyValuePair{
					{
						Key:   common.MaxLengthKey,
						Value: "125",
					},
				},
			},
			{
				FieldID:     109,
				Name:        "field_array",
				DataType:    schemapb.DataType_Array,
				ElementType: schemapb.DataType_Int32,
			},
			{
				FieldID:  110,
				Name:     "field_json",
				DataType: schemapb.DataType_JSON,
			},
			{
				FieldID:  111,
				Name:     "field_geometry",
				DataType: schemapb.DataType_Geometry,
			},
		},
	}

	for _, field := range schema.GetFields() {
		values := fieldValues[field.GetFieldID()]
		fieldData := genFieldData(field.GetName(), field.GetFieldID(), field.GetDataType(), values, 0)
		size := CalcScalarSize(fieldData)
		expected := 0
		switch field.GetDataType() {
		case schemapb.DataType_VarChar:
			data := values.([]string)
			for _, v := range data {
				expected += len(v)
			}
		case schemapb.DataType_Array:
			data := values.([]*schemapb.ScalarField)
			for _, v := range data {
				expected += binary.Size(v.GetIntData().GetData())
			}
		case schemapb.DataType_JSON:
			data := values.([][]byte)
			for _, v := range data {
				expected += len(v)
			}

		case schemapb.DataType_Geometry:
			data := values.([][]byte)
			for _, v := range data {
				expected += len(v)
			}
		default:
			expected = binary.Size(fieldValues[field.GetFieldID()])
		}

		assert.Equal(t, expected, size, field.GetName())
	}
}

func TestGetDataAndGetDataSize(t *testing.T) {
	const (
		Dim       = 8
		fieldName = "field-0"
		fieldID   = 0
	)

	BoolArray := []bool{true, false}
	Int8Array := []int8{1, 2}
	Int16Array := []int16{3, 4}
	Int32Array := []int32{5, 6}
	Int64Array := []int64{11, 22}
	FloatArray := []float32{1.0, 2.0}
	DoubleArray := []float64{11.0, 22.0}
	VarCharArray := []string{"a", "b"}
	BinaryVector := []byte{0x12, 0x34}
	FloatVector := []float32{1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 11.0, 22.0, 33.0, 44.0, 55.0, 66.0, 77.0, 88.0}
	Float16Vector := []byte{
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
	}
	BFloat16Vector := []byte{
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
	}
	SparseFloatVector := &schemapb.SparseFloatArray{
		Dim: 231,
		Contents: [][]byte{
			CreateSparseFloatRow([]uint32{}, []float32{}),
			CreateSparseFloatRow([]uint32{60, 80, 230}, []float32{2.1, 2.2, 2.3}),
		},
	}
	Int8Vector := []byte{
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
	}

	boolData := genFieldData(fieldName, fieldID, schemapb.DataType_Bool, BoolArray, 1)
	int8Data := genFieldData(fieldName, fieldID, schemapb.DataType_Int8, Int8Array, 1)
	int16Data := genFieldData(fieldName, fieldID, schemapb.DataType_Int16, Int16Array, 1)
	int32Data := genFieldData(fieldName, fieldID, schemapb.DataType_Int32, Int32Array, 1)
	int64Data := genFieldData(fieldName, fieldID, schemapb.DataType_Int64, Int64Array, 1)
	floatData := genFieldData(fieldName, fieldID, schemapb.DataType_Float, FloatArray, 1)
	doubleData := genFieldData(fieldName, fieldID, schemapb.DataType_Double, DoubleArray, 1)
	varCharData := genFieldData(fieldName, fieldID, schemapb.DataType_VarChar, VarCharArray, 1)
	binVecData := genFieldData(fieldName, fieldID, schemapb.DataType_BinaryVector, BinaryVector, Dim)
	floatVecData := genFieldData(fieldName, fieldID, schemapb.DataType_FloatVector, FloatVector, Dim)
	float16VecData := genFieldData(fieldName, fieldID, schemapb.DataType_Float16Vector, Float16Vector, Dim)
	bfloat16VecData := genFieldData(fieldName, fieldID, schemapb.DataType_BFloat16Vector, BFloat16Vector, Dim)
	sparseFloatData := genFieldData(fieldName, fieldID, schemapb.DataType_SparseFloatVector, SparseFloatVector.Contents[0], SparseFloatVector.Dim)
	int8VecData := genFieldData(fieldName, fieldID, schemapb.DataType_Int8Vector, Int8Vector, Dim)
	invalidData := &schemapb.FieldData{
		Type: schemapb.DataType_None,
	}

	t.Run("test GetPKSize", func(t *testing.T) {
		int64DataRes := GetPKSize(int64Data)
		varCharDataRes := GetPKSize(varCharData)

		assert.Equal(t, 2, int64DataRes)
		assert.Equal(t, 2, varCharDataRes)
	})

	t.Run("test GetData", func(t *testing.T) {
		boolDataRes := getData(boolData, 0)
		int8DataRes := getData(int8Data, 0)
		int16DataRes := getData(int16Data, 0)
		int32DataRes := getData(int32Data, 0)
		int64DataRes := getData(int64Data, 0)
		floatDataRes := getData(floatData, 0)
		doubleDataRes := getData(doubleData, 0)
		varCharDataRes := getData(varCharData, 0)
		binVecDataRes := getData(binVecData, 0)
		floatVecDataRes := getData(floatVecData, 0)
		float16VecDataRes := getData(float16VecData, 0)
		bfloat16VecDataRes := getData(bfloat16VecData, 0)
		sparseFloatDataRes := getData(sparseFloatData, 0)
		int8VecDataRes := getData(int8VecData, 0)
		invalidDataRes := getData(invalidData, 0)

		assert.Equal(t, BoolArray[0], boolDataRes)
		assert.Equal(t, int32(Int8Array[0]), int8DataRes)
		assert.Equal(t, int32(Int16Array[0]), int16DataRes)
		assert.Equal(t, Int32Array[0], int32DataRes)
		assert.Equal(t, Int64Array[0], int64DataRes)
		assert.Equal(t, FloatArray[0], floatDataRes)
		assert.Equal(t, DoubleArray[0], doubleDataRes)
		assert.Equal(t, VarCharArray[0], varCharDataRes)
		assert.ElementsMatch(t, BinaryVector[:Dim/8], binVecDataRes)
		assert.ElementsMatch(t, FloatVector[:Dim], floatVecDataRes)
		assert.ElementsMatch(t, Float16Vector[:2*Dim], float16VecDataRes)
		assert.ElementsMatch(t, BFloat16Vector[:2*Dim], bfloat16VecDataRes)
		assert.Equal(t, SparseFloatVector.Contents[0], sparseFloatDataRes)
		assert.ElementsMatch(t, Int8Vector[:Dim], int8VecDataRes)
		assert.Nil(t, invalidDataRes)
	})
}

func TestMergeFieldData(t *testing.T) {
	sparseFloatRows := [][]byte{
		// 3 rows for dst
		CreateSparseFloatRow([]uint32{30, 41, 52}, []float32{1.1, 1.2, 1.3}),
		CreateSparseFloatRow([]uint32{60, 80, 230}, []float32{2.1, 2.2, 2.3}),
		CreateSparseFloatRow([]uint32{300, 410, 520}, []float32{1.1, 1.2, 1.3}),
		// 3 rows for src
		CreateSparseFloatRow([]uint32{600, 800, 2300}, []float32{2.1, 2.2, 2.3}),
		CreateSparseFloatRow([]uint32{90, 141, 352}, []float32{1.1, 1.2, 1.3}),
		CreateSparseFloatRow([]uint32{}, []float32{}),
	}

	t.Run("merge data", func(t *testing.T) {
		dstFields := []*schemapb.FieldData{
			genFieldData("int64", 100, schemapb.DataType_Int64, []int64{1, 2, 3}, 1),
			genFieldData("vector", 101, schemapb.DataType_FloatVector, []float32{1, 2, 3}, 1),
			genFieldData("json", 102, schemapb.DataType_JSON, [][]byte{[]byte(`{"key":"value"}`), []byte(`{"hello":"world"}`)}, 1),
			genFieldData("array", 103, schemapb.DataType_Array, []*schemapb.ScalarField{
				{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{
							Data: []int32{1, 2, 3},
						},
					},
				},
			}, 1),
			{
				Type:      schemapb.DataType_Array,
				FieldName: "bytes",
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_BytesData{},
					},
				},
				FieldId: 104,
			},
			{
				Type:      schemapb.DataType_Array,
				FieldName: "bytes",
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_BytesData{
							BytesData: &schemapb.BytesArray{
								Data: [][]byte{[]byte("hello"), []byte("world")},
							},
						},
					},
				},
				FieldId: 105,
			},
			{
				Type:      schemapb.DataType_SparseFloatVector,
				FieldName: "sparseFloat",
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim: 521,
						Data: &schemapb.VectorField_SparseFloatVector{
							SparseFloatVector: &schemapb.SparseFloatArray{
								Dim:      521,
								Contents: sparseFloatRows[:3],
							},
						},
					},
				},
				FieldId: 106,
			},
			genFieldData("float16_vector", 111, schemapb.DataType_Float16Vector, []byte("12345678"), 4),
			genFieldData("bfloat16_vector", 112, schemapb.DataType_BFloat16Vector, []byte("12345678"), 4),
			genFieldData("int8_vector", 113, schemapb.DataType_Int8Vector, []byte("12345678"), 4),
			genFieldData("geometry", 114, schemapb.DataType_Geometry, [][]byte{
				{0x01, 0x01, 0x00, 0x00, 0x00, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A, 0x0D, 0x1B, 0x4F, 0x4F, 0x9A, 0x3D, 0x40},
				{0x01, 0x02, 0x00, 0x00, 0x00, 0x03, 0x00, 0x00, 0x00, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A, 0x0D, 0x1B, 0x4F, 0x4F, 0x9A, 0x3D, 0x40, 0x03, 0xA6, 0xB4, 0xA6, 0xA4, 0xD2, 0xC5, 0xC0, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A},
			}, 1),
		}

		srcFields := []*schemapb.FieldData{
			genFieldData("int64", 100, schemapb.DataType_Int64, []int64{4, 5, 6}, 1),
			genFieldData("vector", 101, schemapb.DataType_FloatVector, []float32{4, 5, 6}, 1),
			genFieldData("json", 102, schemapb.DataType_JSON, [][]byte{[]byte(`{"key":"value"}`), []byte(`{"hello":"world"}`)}, 1),
			genFieldData("array", 103, schemapb.DataType_Array, []*schemapb.ScalarField{
				{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{
							Data: []int32{4, 5, 6},
						},
					},
				},
			}, 1),
			{
				Type:      schemapb.DataType_Array,
				FieldName: "bytes",
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_BytesData{
							BytesData: &schemapb.BytesArray{
								Data: [][]byte{[]byte("hoo"), []byte("foo")},
							},
						},
					},
				},
				FieldId: 104,
			},
			{
				Type:      schemapb.DataType_Array,
				FieldName: "bytes",
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_BytesData{
							BytesData: &schemapb.BytesArray{
								Data: [][]byte{[]byte("hoo")},
							},
						},
					},
				},
				FieldId: 105,
			},
			{
				Type:      schemapb.DataType_SparseFloatVector,
				FieldName: "sparseFloat",
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim: 2301,
						Data: &schemapb.VectorField_SparseFloatVector{
							SparseFloatVector: &schemapb.SparseFloatArray{
								Dim:      2301,
								Contents: sparseFloatRows[3:],
							},
						},
					},
				},
				FieldId: 106,
			},
			genFieldData("float16_vector", 111, schemapb.DataType_Float16Vector, []byte("abcdefgh"), 4),
			genFieldData("bfloat16_vector", 112, schemapb.DataType_BFloat16Vector, []byte("ABCDEFGH"), 4),
			genFieldData("int8_vector", 113, schemapb.DataType_Int8Vector, []byte("abcdefgh"), 4),
			genFieldData("geometry", 114, schemapb.DataType_Geometry, [][]byte{
				{0x01, 0x01, 0x00, 0x00, 0x00, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A, 0x0D, 0x1B, 0x4F, 0x4F, 0x9A, 0x3D, 0x40},
				{0x01, 0x02, 0x00, 0x00, 0x00, 0x03, 0x00, 0x00, 0x00, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A, 0x0D, 0x1B, 0x4F, 0x4F, 0x9A, 0x3D, 0x40, 0x03, 0xA6, 0xB4, 0xA6, 0xA4, 0xD2, 0xC5, 0xC0, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A},
			}, 1),
		}

		err := MergeFieldData(dstFields, srcFields)
		assert.NoError(t, err)

		assert.Equal(t, []int64{1, 2, 3, 4, 5, 6}, dstFields[0].GetScalars().GetLongData().Data)
		assert.Equal(t, []float32{1, 2, 3, 4, 5, 6}, dstFields[1].GetVectors().GetFloatVector().Data)
		assert.Equal(t, [][]byte{[]byte(`{"key":"value"}`), []byte(`{"hello":"world"}`), []byte(`{"key":"value"}`), []byte(`{"hello":"world"}`)},
			dstFields[2].GetScalars().GetJsonData().Data)
		assert.Equal(t, []*schemapb.ScalarField{
			{
				Data: &schemapb.ScalarField_IntData{
					IntData: &schemapb.IntArray{
						Data: []int32{1, 2, 3},
					},
				},
			},
			{
				Data: &schemapb.ScalarField_IntData{
					IntData: &schemapb.IntArray{
						Data: []int32{4, 5, 6},
					},
				},
			},
		},
			dstFields[3].GetScalars().GetArrayData().Data)
		assert.Equal(t, [][]byte{[]byte("hoo"), []byte("foo")}, dstFields[4].GetScalars().GetBytesData().Data)
		assert.Equal(t, [][]byte{[]byte("hello"), []byte("world"), []byte("hoo")}, dstFields[5].GetScalars().GetBytesData().Data)
		assert.Equal(t, &schemapb.SparseFloatArray{
			Dim:      2301,
			Contents: sparseFloatRows,
		}, dstFields[6].GetVectors().GetSparseFloatVector())
		assert.Equal(t, []byte("12345678abcdefgh"), dstFields[7].GetVectors().GetFloat16Vector())
		assert.Equal(t, []byte("12345678ABCDEFGH"), dstFields[8].GetVectors().GetBfloat16Vector())
		assert.Equal(t, []byte("12345678abcdefgh"), dstFields[9].GetVectors().GetInt8Vector())
	})

	t.Run("merge with nil", func(t *testing.T) {
		srcFields := []*schemapb.FieldData{
			genFieldData("int64", 100, schemapb.DataType_Int64, []int64{1, 2, 3}, 1),
			genFieldData("vector", 101, schemapb.DataType_FloatVector, []float32{1, 2, 3}, 1),
			genFieldData("json", 102, schemapb.DataType_JSON, [][]byte{[]byte(`{"key":"value"}`), []byte(`{"hello":"world"}`)}, 1),
			genFieldData("array", 103, schemapb.DataType_Array, []*schemapb.ScalarField{
				{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{
							Data: []int32{1, 2, 3},
						},
					},
				},
			}, 1),
			{
				Type:      schemapb.DataType_SparseFloatVector,
				FieldName: "sparseFloat",
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim: 521,
						Data: &schemapb.VectorField_SparseFloatVector{
							SparseFloatVector: &schemapb.SparseFloatArray{
								Dim:      521,
								Contents: sparseFloatRows[:3],
							},
						},
					},
				},
				FieldId: 104,
			},
			genFieldData("float16_vector", 111, schemapb.DataType_Float16Vector, []byte("12345678"), 4),
			genFieldData("bfloat16_vector", 112, schemapb.DataType_BFloat16Vector, []byte("12345678"), 4),
			genFieldData("int8_vector", 113, schemapb.DataType_Int8Vector, []byte("12345678"), 4),
			genFieldData("geometry", 114, schemapb.DataType_Geometry, [][]byte{
				{0x01, 0x01, 0x00, 0x00, 0x00, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A, 0x0D, 0x1B, 0x4F, 0x4F, 0x9A, 0x3D, 0x40},
				{0x01, 0x02, 0x00, 0x00, 0x00, 0x03, 0x00, 0x00, 0x00, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A, 0x0D, 0x1B, 0x4F, 0x4F, 0x9A, 0x3D, 0x40, 0x03, 0xA6, 0xB4, 0xA6, 0xA4, 0xD2, 0xC5, 0xC0, 0xD2, 0x4A, 0x4D, 0x6A, 0x8B, 0x3C, 0x5C, 0x0A},
			}, 1),
		}

		dstFields := []*schemapb.FieldData{
			{Type: schemapb.DataType_Int64, FieldName: "int64", Field: &schemapb.FieldData_Scalars{Scalars: &schemapb.ScalarField{Data: &schemapb.ScalarField_LongData{}}}, FieldId: 100},
			{Type: schemapb.DataType_FloatVector, FieldName: "vector", Field: &schemapb.FieldData_Vectors{Vectors: &schemapb.VectorField{Data: &schemapb.VectorField_FloatVector{}}}, FieldId: 101},
			{Type: schemapb.DataType_JSON, FieldName: "json", Field: &schemapb.FieldData_Scalars{Scalars: &schemapb.ScalarField{Data: &schemapb.ScalarField_JsonData{}}}, FieldId: 102},
			{Type: schemapb.DataType_Array, FieldName: "array", Field: &schemapb.FieldData_Scalars{Scalars: &schemapb.ScalarField{Data: &schemapb.ScalarField_ArrayData{}}}, FieldId: 103},
			{Type: schemapb.DataType_SparseFloatVector, FieldName: "sparseFloat", Field: &schemapb.FieldData_Vectors{Vectors: &schemapb.VectorField{Data: &schemapb.VectorField_SparseFloatVector{}}}, FieldId: 104},
			{Type: schemapb.DataType_Float16Vector, FieldName: "float16_vector", Field: &schemapb.FieldData_Vectors{Vectors: &schemapb.VectorField{Data: &schemapb.VectorField_Float16Vector{}}}, FieldId: 111},
			{Type: schemapb.DataType_BFloat16Vector, FieldName: "bfloat16_vector", Field: &schemapb.FieldData_Vectors{Vectors: &schemapb.VectorField{Data: &schemapb.VectorField_Bfloat16Vector{}}}, FieldId: 112},
			{Type: schemapb.DataType_Int8Vector, FieldName: "int8_vector", Field: &schemapb.FieldData_Vectors{Vectors: &schemapb.VectorField{Data: &schemapb.VectorField_Int8Vector{}}}, FieldId: 113},
			{Type: schemapb.DataType_Geometry, FieldName: "geometry", Field: &schemapb.FieldData_Scalars{Scalars: &schemapb.ScalarField{Data: &schemapb.ScalarField_GeometryData{}}}, FieldId: 114},
		}

		err := MergeFieldData(dstFields, srcFields)
		assert.NoError(t, err)

		assert.Equal(t, []int64{1, 2, 3}, dstFields[0].GetScalars().GetLongData().Data)
		assert.Equal(t, []float32{1, 2, 3}, dstFields[1].GetVectors().GetFloatVector().Data)
		assert.Equal(t, [][]byte{[]byte(`{"key":"value"}`), []byte(`{"hello":"world"}`)},
			dstFields[2].GetScalars().GetJsonData().Data)
		assert.Equal(t, []*schemapb.ScalarField{
			{
				Data: &schemapb.ScalarField_IntData{
					IntData: &schemapb.IntArray{
						Data: []int32{1, 2, 3},
					},
				},
			},
		},
			dstFields[3].GetScalars().GetArrayData().Data)
		assert.Equal(t, &schemapb.SparseFloatArray{
			Dim:      521,
			Contents: sparseFloatRows[:3],
		}, dstFields[4].GetVectors().GetSparseFloatVector())
		assert.Equal(t, []byte("12345678"), dstFields[5].GetVectors().GetFloat16Vector())
		assert.Equal(t, []byte("12345678"), dstFields[6].GetVectors().GetBfloat16Vector())
		assert.Equal(t, []byte("12345678"), dstFields[7].GetVectors().GetInt8Vector())
	})

	t.Run("error case", func(t *testing.T) {
		emptyField := &schemapb.FieldData{
			Type: schemapb.DataType_None,
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: nil,
				},
			},
		}

		err := MergeFieldData([]*schemapb.FieldData{emptyField}, []*schemapb.FieldData{emptyField})
		assert.Error(t, err)
	})

	t.Run("nullable float vector - all null src", func(t *testing.T) {
		dim := int64(4)
		dstFields := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_FloatVector,
				FieldName: "nullable_vec",
				FieldId:   200,
				ValidData: []bool{true},
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim: dim,
						Data: &schemapb.VectorField_FloatVector{
							FloatVector: &schemapb.FloatArray{
								Data: []float32{1, 2, 3, 4},
							},
						},
					},
				},
			},
		}
		// src: all rows are null -> VectorField.Data = nil
		srcFields := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_FloatVector,
				FieldName: "nullable_vec",
				FieldId:   200,
				ValidData: []bool{false},
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim:  dim,
						Data: nil, // all null -> no FloatVector oneof set
					},
				},
			},
		}

		err := MergeFieldData(dstFields, srcFields)
		assert.NoError(t, err, "MergeFieldData should handle nullable FloatVector with nil VectorField.Data")
	})
}

type FieldDataSuite struct {
	suite.Suite
}

func (s *FieldDataSuite) TestPrepareFieldData() {
	fieldID := int64(100)
	fieldName := "testField"
	topK := int64(100)

	s.Run("bool", func() {
		samples := []*schemapb.FieldData{
			{
				FieldId:   fieldID,
				FieldName: fieldName,
				Type:      schemapb.DataType_Bool,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_BoolData{},
					},
				},
			},
		}

		fields := PrepareResultFieldData(samples, topK)
		s.Require().Len(fields, 1)
		field := fields[0]
		s.Equal(fieldID, field.GetFieldId())
		s.Equal(fieldName, field.GetFieldName())
		s.Equal(schemapb.DataType_Bool, field.GetType())

		s.EqualValues(topK, cap(field.GetScalars().GetBoolData().GetData()))
	})

	s.Run("int", func() {
		dataTypes := []schemapb.DataType{
			schemapb.DataType_Int32,
			schemapb.DataType_Int16,
			schemapb.DataType_Int8,
		}
		for _, dataType := range dataTypes {
			samples := []*schemapb.FieldData{
				{
					FieldId:   fieldID,
					FieldName: fieldName,
					Type:      dataType,
					Field: &schemapb.FieldData_Scalars{
						Scalars: &schemapb.ScalarField{
							Data: &schemapb.ScalarField_IntData{},
						},
					},
				},
			}

			fields := PrepareResultFieldData(samples, topK)
			s.Require().Len(fields, 1)
			field := fields[0]
			s.Equal(fieldID, field.GetFieldId())
			s.Equal(fieldName, field.GetFieldName())
			s.Equal(dataType, field.GetType())

			s.EqualValues(topK, cap(field.GetScalars().GetIntData().GetData()))
		}
	})

	s.Run("long", func() {
		samples := []*schemapb.FieldData{
			{
				FieldId:   fieldID,
				FieldName: fieldName,
				Type:      schemapb.DataType_Int64,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_LongData{},
					},
				},
			},
		}

		fields := PrepareResultFieldData(samples, topK)
		s.Require().Len(fields, 1)
		field := fields[0]
		s.Equal(fieldID, field.GetFieldId())
		s.Equal(fieldName, field.GetFieldName())
		s.Equal(schemapb.DataType_Int64, field.GetType())

		s.EqualValues(topK, cap(field.GetScalars().GetLongData().GetData()))
	})

	s.Run("float", func() {
		samples := []*schemapb.FieldData{
			{
				FieldId:   fieldID,
				FieldName: fieldName,
				Type:      schemapb.DataType_Float,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_FloatData{},
					},
				},
			},
		}

		fields := PrepareResultFieldData(samples, topK)
		s.Require().Len(fields, 1)
		field := fields[0]
		s.Equal(fieldID, field.GetFieldId())
		s.Equal(fieldName, field.GetFieldName())
		s.Equal(schemapb.DataType_Float, field.GetType())

		s.EqualValues(topK, cap(field.GetScalars().GetFloatData().GetData()))
	})

	s.Run("double", func() {
		samples := []*schemapb.FieldData{
			{
				FieldId:   fieldID,
				FieldName: fieldName,
				Type:      schemapb.DataType_Double,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_DoubleData{},
					},
				},
			},
		}

		fields := PrepareResultFieldData(samples, topK)
		s.Require().Len(fields, 1)
		field := fields[0]
		s.Equal(fieldID, field.GetFieldId())
		s.Equal(fieldName, field.GetFieldName())
		s.Equal(schemapb.DataType_Double, field.GetType())

		s.EqualValues(topK, cap(field.GetScalars().GetDoubleData().GetData()))
	})

	s.Run("string", func() {
		dataTypes := []schemapb.DataType{
			schemapb.DataType_VarChar,
			schemapb.DataType_String,
		}
		for _, dataType := range dataTypes {
			samples := []*schemapb.FieldData{
				{
					FieldId:   fieldID,
					FieldName: fieldName,
					Type:      dataType,
					Field: &schemapb.FieldData_Scalars{
						Scalars: &schemapb.ScalarField{
							Data: &schemapb.ScalarField_StringData{},
						},
					},
				},
			}

			fields := PrepareResultFieldData(samples, topK)
			s.Require().Len(fields, 1)
			field := fields[0]
			s.Equal(fieldID, field.GetFieldId())
			s.Equal(fieldName, field.GetFieldName())
			s.Equal(dataType, field.GetType())

			s.EqualValues(topK, cap(field.GetScalars().GetStringData().GetData()))
		}
	})

	s.Run("json", func() {
		samples := []*schemapb.FieldData{
			{
				FieldId:   fieldID,
				FieldName: fieldName,
				Type:      schemapb.DataType_JSON,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_JsonData{},
					},
				},
			},
		}

		fields := PrepareResultFieldData(samples, topK)
		s.Require().Len(fields, 1)
		field := fields[0]
		s.Equal(fieldID, field.GetFieldId())
		s.Equal(fieldName, field.GetFieldName())
		s.Equal(schemapb.DataType_JSON, field.GetType())

		s.EqualValues(topK, cap(field.GetScalars().GetJsonData().GetData()))
	})

	s.Run("array", func() {
		samples := []*schemapb.FieldData{
			{
				FieldId:   fieldID,
				FieldName: fieldName,
				Type:      schemapb.DataType_Array,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_ArrayData{
							ArrayData: &schemapb.ArrayArray{
								ElementType: schemapb.DataType_Bool,
							},
						},
					},
				},
			},
		}

		fields := PrepareResultFieldData(samples, topK)
		s.Require().Len(fields, 1)
		field := fields[0]
		s.Equal(fieldID, field.GetFieldId())
		s.Equal(fieldName, field.GetFieldName())
		s.Equal(schemapb.DataType_Array, field.GetType())

		s.EqualValues(topK, cap(field.GetScalars().GetArrayData().GetData()))
		s.Equal(schemapb.DataType_Bool, field.GetScalars().GetArrayData().GetElementType())
	})

	s.Run("float_vector", func() {
		samples := []*schemapb.FieldData{
			{
				FieldId:   fieldID,
				FieldName: fieldName,
				Type:      schemapb.DataType_FloatVector,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim:  128,
						Data: &schemapb.VectorField_FloatVector{},
					},
				},
			},
		}

		fields := PrepareResultFieldData(samples, topK)
		s.Require().Len(fields, 1)
		field := fields[0]
		s.Equal(fieldID, field.GetFieldId())
		s.Equal(fieldName, field.GetFieldName())
		s.Equal(schemapb.DataType_FloatVector, field.GetType())

		s.EqualValues(128, field.GetVectors().GetDim())
		s.EqualValues(topK*128, cap(field.GetVectors().GetFloatVector().GetData()))
	})

	s.Run("float16_vector", func() {
		samples := []*schemapb.FieldData{
			{
				FieldId:   fieldID,
				FieldName: fieldName,
				Type:      schemapb.DataType_Float16Vector,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim:  128,
						Data: &schemapb.VectorField_Float16Vector{},
					},
				},
			},
		}

		fields := PrepareResultFieldData(samples, topK)
		s.Require().Len(fields, 1)
		field := fields[0]
		s.Equal(fieldID, field.GetFieldId())
		s.Equal(fieldName, field.GetFieldName())
		s.Equal(schemapb.DataType_Float16Vector, field.GetType())

		s.EqualValues(128, field.GetVectors().GetDim())
		s.EqualValues(topK*128*2, cap(field.GetVectors().GetFloat16Vector()))
	})

	s.Run("bfloat16_vector", func() {
		samples := []*schemapb.FieldData{
			{
				FieldId:   fieldID,
				FieldName: fieldName,
				Type:      schemapb.DataType_BFloat16Vector,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim:  128,
						Data: &schemapb.VectorField_Bfloat16Vector{},
					},
				},
			},
		}

		fields := PrepareResultFieldData(samples, topK)
		s.Require().Len(fields, 1)
		field := fields[0]
		s.Equal(fieldID, field.GetFieldId())
		s.Equal(fieldName, field.GetFieldName())
		s.Equal(schemapb.DataType_BFloat16Vector, field.GetType())

		s.EqualValues(128, field.GetVectors().GetDim())
		s.EqualValues(topK*128*2, cap(field.GetVectors().GetBfloat16Vector()))
	})

	s.Run("binary_vector", func() {
		samples := []*schemapb.FieldData{
			{
				FieldId:   fieldID,
				FieldName: fieldName,
				Type:      schemapb.DataType_BinaryVector,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim:  128,
						Data: &schemapb.VectorField_BinaryVector{},
					},
				},
			},
		}

		fields := PrepareResultFieldData(samples, topK)
		s.Require().Len(fields, 1)
		field := fields[0]
		s.Equal(fieldID, field.GetFieldId())
		s.Equal(fieldName, field.GetFieldName())
		s.Equal(schemapb.DataType_BinaryVector, field.GetType())

		s.EqualValues(128, field.GetVectors().GetDim())
		s.EqualValues(topK*128/8, cap(field.GetVectors().GetBinaryVector()))
	})

	s.Run("sparse_float_vector", func() {
		samples := []*schemapb.FieldData{
			{
				FieldId:   fieldID,
				FieldName: fieldName,
				Type:      schemapb.DataType_SparseFloatVector,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim:  128,
						Data: &schemapb.VectorField_SparseFloatVector{},
					},
				},
			},
		}

		fields := PrepareResultFieldData(samples, topK)
		s.Require().Len(fields, 1)
		field := fields[0]
		s.Equal(fieldID, field.GetFieldId())
		s.Equal(fieldName, field.GetFieldName())
		s.Equal(schemapb.DataType_SparseFloatVector, field.GetType())

		s.EqualValues(0, field.GetVectors().GetDim())
		s.EqualValues(topK, cap(field.GetVectors().GetSparseFloatVector().GetContents()))
	})

	s.Run("int8_vector", func() {
		samples := []*schemapb.FieldData{
			{
				FieldId:   fieldID,
				FieldName: fieldName,
				Type:      schemapb.DataType_Int8Vector,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim:  128,
						Data: &schemapb.VectorField_Int8Vector{},
					},
				},
			},
		}

		fields := PrepareResultFieldData(samples, topK)
		s.Require().Len(fields, 1)
		field := fields[0]
		s.Equal(fieldID, field.GetFieldId())
		s.Equal(fieldName, field.GetFieldName())
		s.Equal(schemapb.DataType_Int8Vector, field.GetType())

		s.EqualValues(128, field.GetVectors().GetDim())
		s.EqualValues(topK*128, cap(field.GetVectors().GetInt8Vector()))
	})
}

func TestFieldData(t *testing.T) {
	suite.Run(t, new(FieldDataSuite))
}

func TestValidateSparseFloatRows(t *testing.T) {
	t.Run("valid rows", func(t *testing.T) {
		rows := [][]byte{
			CreateSparseFloatRow([]uint32{1, 3, 5}, []float32{1.0, 2.0, 3.0}),
			CreateSparseFloatRow([]uint32{2, 4, 6}, []float32{4.0, 5.0, 6.0}),
			CreateSparseFloatRow([]uint32{0, 7, 8}, []float32{7.0, 8.0, 9.0}),
			// we allow empty row(including indices with 0 value)
			CreateSparseFloatRow([]uint32{}, []float32{}),
			CreateSparseFloatRow([]uint32{24}, []float32{0}),
			{},
		}
		err := ValidateSparseFloatRows(rows...)
		assert.NoError(t, err)
	})

	t.Run("nil row", func(t *testing.T) {
		err := ValidateSparseFloatRows(nil)
		assert.Error(t, err)
	})

	t.Run("incorrect lengths", func(t *testing.T) {
		rows := [][]byte{
			make([]byte, 10),
		}
		err := ValidateSparseFloatRows(rows...)
		assert.Error(t, err)
	})

	t.Run("unordered index", func(t *testing.T) {
		rows := [][]byte{
			CreateSparseFloatRow([]uint32{100, 2000, 500}, []float32{1.0, 2.0, 3.0}),
		}
		err := ValidateSparseFloatRows(rows...)
		assert.Error(t, err)
	})

	t.Run("same index", func(t *testing.T) {
		rows := [][]byte{
			CreateSparseFloatRow([]uint32{100, 100, 500}, []float32{1.0, 2.0, 3.0}),
		}
		err := ValidateSparseFloatRows(rows...)
		assert.Error(t, err)
	})

	t.Run("negative value", func(t *testing.T) {
		rows := [][]byte{
			CreateSparseFloatRow([]uint32{100, 200, 500}, []float32{-1.0, 2.0, 3.0}),
		}
		err := ValidateSparseFloatRows(rows...)
		assert.Error(t, err)
	})

	t.Run("invalid value", func(t *testing.T) {
		rows := [][]byte{
			CreateSparseFloatRow([]uint32{100, 200, 500}, []float32{float32(math.NaN()), 2.0, 3.0}),
		}
		err := ValidateSparseFloatRows(rows...)
		assert.Error(t, err)

		rows = [][]byte{
			CreateSparseFloatRow([]uint32{100, 200, 500}, []float32{float32(math.Inf(1)), 2.0, 3.0}),
		}
		err = ValidateSparseFloatRows(rows...)
		assert.Error(t, err)

		rows = [][]byte{
			CreateSparseFloatRow([]uint32{100, 200, 500}, []float32{float32(math.Inf(-1)), 2.0, 3.0}),
		}
		err = ValidateSparseFloatRows(rows...)
		assert.Error(t, err)
	})

	t.Run("invalid index", func(t *testing.T) {
		rows := [][]byte{
			CreateSparseFloatRow([]uint32{3, 5, math.MaxUint32}, []float32{1.0, 2.0, 3.0}),
		}
		err := ValidateSparseFloatRows(rows...)
		assert.Error(t, err)
	})

	t.Run("no rows", func(t *testing.T) {
		err := ValidateSparseFloatRows()
		assert.NoError(t, err)
	})
}

func TestParseJsonSparseFloatRow(t *testing.T) {
	t.Run("valid row 1", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{1, 3, 5}, "values": []interface{}{1.0, 2.0, 3.0}}
		res, err := CreateSparseFloatRowFromMap(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1, 3, 5}, []float32{1.0, 2.0, 3.0}), res)
	})

	t.Run("valid row 2", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{3, 1, 5}, "values": []interface{}{1.0, 2.0, 3.0}}
		res, err := CreateSparseFloatRowFromMap(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1, 3, 5}, []float32{2.0, 1.0, 3.0}), res)
	})

	t.Run("valid row 3", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{1, 3, 5}, "values": []interface{}{1, 2, 3}}
		res, err := CreateSparseFloatRowFromMap(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1, 3, 5}, []float32{1.0, 2.0, 3.0}), res)
	})

	t.Run("valid row 4", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{math.MaxInt32 + 1}, "values": []interface{}{1.0}}
		res, err := CreateSparseFloatRowFromMap(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{math.MaxInt32 + 1}, []float32{1.0}), res)
	})

	t.Run("valid row 5", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{}, "values": []interface{}{}}
		res, err := CreateSparseFloatRowFromMap(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{}, []float32{}), res)
	})

	t.Run("valid row 6", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{1}, "values": []interface{}{0}}
		res, err := CreateSparseFloatRowFromMap(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1}, []float32{0}), res)
	})

	t.Run("invalid row 1", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{1, 3, 5}, "values": []interface{}{1.0, 2.0}}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid row 2", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{1}, "values": []interface{}{1.0, 2.0}}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid row 4", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{3}, "values": []interface{}{-0.2}}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid row 5", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{3.1}, "values": []interface{}{0.2}}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid row 6", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{-1}, "values": []interface{}{0.2}}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid row 7", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{math.MaxUint32}, "values": []interface{}{1.0}}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid row 8", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{math.MaxUint32 + 10}, "values": []interface{}{1.0}}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid row 9", func(t *testing.T) {
		row := map[string]interface{}{"indices": []interface{}{10}, "values": []interface{}{float64(math.MaxFloat32) * 2}}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("valid dict row 1", func(t *testing.T) {
		row := map[string]interface{}{"1": 1.0, "3": 2.0, "5": 3.0}
		res, err := CreateSparseFloatRowFromMap(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1, 3, 5}, []float32{1.0, 2.0, 3.0}), res)
	})

	t.Run("valid dict row 2", func(t *testing.T) {
		row := map[string]interface{}{"3": 1.0, "1": 2.0, "5": 3.0}
		res, err := CreateSparseFloatRowFromMap(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1, 3, 5}, []float32{2.0, 1.0, 3.0}), res)
	})

	t.Run("valid dict row 3", func(t *testing.T) {
		row := map[string]interface{}{}
		res, err := CreateSparseFloatRowFromMap(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{}, []float32{}), res)
	})

	t.Run("valid dict row 4", func(t *testing.T) {
		row := map[string]interface{}{"1": 0}
		res, err := CreateSparseFloatRowFromMap(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1}, []float32{0}), res)
	})

	t.Run("invalid dict row 1", func(t *testing.T) {
		row := map[string]interface{}{"a": 1.0, "3": 2.0, "5": 3.0}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 2", func(t *testing.T) {
		row := map[string]interface{}{"1": "a", "3": 2.0, "5": 3.0}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 3", func(t *testing.T) {
		row := map[string]interface{}{"1": "1.0", "3": 2.0, "5": 3.0}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 4", func(t *testing.T) {
		row := map[string]interface{}{"-1": 1.0, "3": 2.0, "5": 3.0}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 5", func(t *testing.T) {
		row := map[string]interface{}{"1": -1.0, "3": 2.0, "5": 3.0}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 7", func(t *testing.T) {
		row := map[string]interface{}{fmt.Sprint(math.MaxUint32): 1.0, "3": 2.0, "5": 3.0}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 8", func(t *testing.T) {
		row := map[string]interface{}{fmt.Sprint(math.MaxUint32 + 10): 1.0, "3": 2.0, "5": 3.0}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 8", func(t *testing.T) {
		row := map[string]interface{}{fmt.Sprint(math.MaxUint32 + 10): 1.0, "3": 2.0, "5": float64(math.MaxFloat32) * 2}
		_, err := CreateSparseFloatRowFromMap(row)
		assert.Error(t, err)
	})
}

func TestParseJsonSparseFloatRowBytes(t *testing.T) {
	t.Run("valid row 1", func(t *testing.T) {
		row := []byte(`{"indices":[1,3,5],"values":[1.0,2.0,3.0]}`)
		res, err := CreateSparseFloatRowFromJSON(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1, 3, 5}, []float32{1.0, 2.0, 3.0}), res)
	})

	t.Run("valid row 2", func(t *testing.T) {
		row := []byte(`{"indices":[3,1,5],"values":[1.0,2.0,3.0]}`)
		res, err := CreateSparseFloatRowFromJSON(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1, 3, 5}, []float32{2.0, 1.0, 3.0}), res)
	})

	t.Run("valid row 3", func(t *testing.T) {
		row := []byte(`{"indices":[1, 3, 5], "values":[1, 2, 3]}`)
		res, err := CreateSparseFloatRowFromJSON(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1, 3, 5}, []float32{1.0, 2.0, 3.0}), res)
	})

	t.Run("valid row 3", func(t *testing.T) {
		row := []byte(`{"indices":[2147483648], "values":[1.0]}`)
		res, err := CreateSparseFloatRowFromJSON(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{math.MaxInt32 + 1}, []float32{1.0}), res)
	})

	t.Run("valid row 4", func(t *testing.T) {
		row := []byte(`{"indices":[], "values":[]}`)
		res, err := CreateSparseFloatRowFromJSON(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{}, []float32{}), res)
	})

	t.Run("valid row 5", func(t *testing.T) {
		row := []byte(`{"indices":[1], "values":[0]}`)
		res, err := CreateSparseFloatRowFromJSON(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1}, []float32{0}), res)
	})

	t.Run("invalid row 1", func(t *testing.T) {
		row := []byte(`{"indices":[1,3,5],"values":[1.0,2.0,3.0`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("invalid row 2", func(t *testing.T) {
		row := []byte(`{"indices":[1,3,5],"values":[1.0,2.0]`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("invalid row 3", func(t *testing.T) {
		row := []byte(`{"indices":[1],"values":[1.0,2.0]`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("invalid row 4", func(t *testing.T) {
		row := []byte(`{"indices":[],"values":[]`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("invalid row 5", func(t *testing.T) {
		row := []byte(`{"indices":[-3],"values":[0.2]`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("invalid row 6", func(t *testing.T) {
		row := []byte(`{"indices":[3],"values":[-0.2]`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("invalid row 7", func(t *testing.T) {
		row := []byte(`{"indices": []interface{}{3.1}, "values": []interface{}{0.2}}`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("valid dict row 1", func(t *testing.T) {
		row := []byte(`{"1": 1.0, "3": 2.0, "5": 3.0}`)
		res, err := CreateSparseFloatRowFromJSON(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1, 3, 5}, []float32{1.0, 2.0, 3.0}), res)
	})

	t.Run("valid dict row 2", func(t *testing.T) {
		row := []byte(`{"3": 1.0, "1": 2.0, "5": 3.0}`)
		res, err := CreateSparseFloatRowFromJSON(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1, 3, 5}, []float32{2.0, 1.0, 3.0}), res)
	})

	t.Run("valid dict row 3", func(t *testing.T) {
		row := []byte(`{}`)
		res, err := CreateSparseFloatRowFromJSON(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{}, []float32{}), res)
	})

	t.Run("valid dict row 4", func(t *testing.T) {
		row := []byte(`{"1": 0}`)
		res, err := CreateSparseFloatRowFromJSON(row)
		assert.NoError(t, err)
		assert.Equal(t, CreateSparseFloatRow([]uint32{1}, []float32{0}), res)
	})

	t.Run("invalid dict row 1", func(t *testing.T) {
		row := []byte(`{"a": 1.0, "3": 2.0, "5": 3.0}`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 2", func(t *testing.T) {
		row := []byte(`{"1": "a", "3": 2.0, "5": 3.0}`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 3", func(t *testing.T) {
		row := []byte(`{"1": "1.0", "3": 2.0, "5": 3.0}`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 4", func(t *testing.T) {
		row := []byte(`{"1": 1.0, "3": 2.0, "5": }`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 5", func(t *testing.T) {
		row := []byte(`{"-1": 1.0, "3": 2.0, "5": 3.0}`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 6", func(t *testing.T) {
		row := []byte(`{"1": -1.0, "3": 2.0, "5": 3.0}`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 7", func(t *testing.T) {
		row := []byte(``)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})

	t.Run("invalid dict row 8", func(t *testing.T) {
		row := []byte(`{"1.1": 1.0, "3": 2.0, "5": 3.0}`)
		_, err := CreateSparseFloatRowFromJSON(row)
		assert.Error(t, err)
	})
}

// test EstimateSparseVectorNNZFromPlaceholderGroup: given a PlaceholderGroup
// with various nq and averageNNZ, test if the estimated number of non-zero
// elements is close to the actual number.
func TestSparsePlaceholderGroupSize(t *testing.T) {
	nqs := []int{1, 100, 10000}
	averageNNZs := []int{1, 4, 16, 64, 256, 1024}
	numCases := 0
	casesWithLargeError := 0
	for _, nq := range nqs {
		for _, averageNNZ := range averageNNZs {
			variants := make([]int, 0)
			for i := 1; i <= averageNNZ/2; i *= 2 {
				variants = append(variants, i)
			}

			for _, variant := range variants {
				numCases++
				contents := make([][]byte, nq)
				contentsSize := 0
				totalNNZ := 0
				for i := range contents {
					// nnz of each row is in range [averageNNZ - variant/2, averageNNZ + variant/2] and at least 1.
					nnz := averageNNZ + variant/2 + rand.Intn(variant)
					if nnz < 1 {
						nnz = 1
					}
					indices := make([]uint32, nnz)
					values := make([]float32, nnz)
					for j := 0; j < nnz; j++ {
						indices[j] = uint32(i*averageNNZ + j)
						values[j] = float32(i*averageNNZ + j)
					}
					contents[i] = CreateSparseFloatRow(indices, values)
					contentsSize += len(contents[i])
					totalNNZ += nnz
				}

				placeholderGroup := &commonpb.PlaceholderGroup{
					Placeholders: []*commonpb.PlaceholderValue{
						{
							Tag:    "$0",
							Type:   commonpb.PlaceholderType_SparseFloatVector,
							Values: contents,
						},
					},
				}
				bytes, _ := proto.Marshal(placeholderGroup)
				estimatedNNZ := EstimateSparseVectorNNZFromPlaceholderGroup(bytes, nq)
				errorRatio := (float64(totalNNZ-estimatedNNZ) / float64(totalNNZ)) * 100
				assert.Less(t, errorRatio, 10.0)
				if errorRatio > 5.0 {
					casesWithLargeError++
				}
				// keep the logs for easy debugging.
				// fmt.Printf("nq: %d, total nnz: %d, overhead bytes: %d, len of bytes: %d\n", nq, totalNNZ, len(bytes)-contentsSize, len(bytes))
				// fmt.Printf("\tnq: %d, total nnz: %d, estimated nnz: %d, diff: %d, error ratio: %f%%\n", nq, totalNNZ, estimatedNNZ, totalNNZ-estimatedNNZ, errorRatio)
			}
		}
	}
	largeErrorRatio := (float64(casesWithLargeError) / float64(numCases)) * 100
	// no more than 2% cases have large error ratio.
	assert.Less(t, largeErrorRatio, 2.0)
}

func TestGetDataIterator(t *testing.T) {
	tests := []struct {
		name  string
		field *schemapb.FieldData
		want  []any
	}{
		{
			name: "empty field",
			field: &schemapb.FieldData{
				Type: schemapb.DataType_Int64,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_LongData{
							LongData: &schemapb.LongArray{},
						},
					},
				},
			},
			want: []any{},
		},
		{
			name: "ints",
			field: &schemapb.FieldData{
				Type: schemapb.DataType_Int64,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_LongData{
							LongData: &schemapb.LongArray{
								Data: []int64{1, 2, 3},
							},
						},
					},
				},
			},
			want: []any{int64(1), int64(2), int64(3)},
		},
		{
			name: "ints with nulls",
			field: &schemapb.FieldData{
				Type: schemapb.DataType_Int64,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_LongData{
							LongData: &schemapb.LongArray{
								Data: []int64{1, 2, 3},
							},
						},
					},
				},
				ValidData: []bool{true, false, true, true},
			},
			want: []any{int64(1), nil, int64(2), int64(3)},
		},
		{
			name: "compact nullable float vector",
			field: &schemapb.FieldData{
				Type: schemapb.DataType_FloatVector,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim: 2,
						Data: &schemapb.VectorField_FloatVector{
							FloatVector: &schemapb.FloatArray{
								Data: []float32{1, 2, 5, 6},
							},
						},
					},
				},
				ValidData: []bool{true, false, true},
			},
			want: []any{[]float32{1, 2}, nil, []float32{5, 6}},
		},
		{
			name: "compact nullable sparse vector",
			field: &schemapb.FieldData{
				Type: schemapb.DataType_SparseFloatVector,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Data: &schemapb.VectorField_SparseFloatVector{
							SparseFloatVector: &schemapb.SparseFloatArray{
								Contents: [][]byte{
									CreateSparseFloatRow([]uint32{1}, []float32{1}),
									CreateSparseFloatRow([]uint32{3}, []float32{3}),
								},
							},
						},
					},
				},
				ValidData: []bool{false, true, false, true},
			},
			want: []any{nil, CreateSparseFloatRow([]uint32{1}, []float32{1}), nil, CreateSparseFloatRow([]uint32{3}, []float32{3})},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			itr := GetDataIterator(tt.field)
			for i, want := range tt.want {
				got := itr(i)
				assert.Equal(t, want, got)
			}
		})
	}
}

func TestUpdateFieldData(t *testing.T) {
	const (
		Dim                  = 8
		BoolFieldName        = "BoolField"
		Int64FieldName       = "Int64Field"
		FloatFieldName       = "FloatField"
		StringFieldName      = "StringField"
		FloatVectorFieldName = "FloatVectorField"
		BoolFieldID          = common.StartOfUserFieldID + 1
		Int64FieldID         = common.StartOfUserFieldID + 2
		FloatFieldID         = common.StartOfUserFieldID + 3
		StringFieldID        = common.StartOfUserFieldID + 4
		FloatVectorFieldID   = common.StartOfUserFieldID + 5
	)

	t.Run("update scalar fields", func(t *testing.T) {
		// Create base data
		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Bool,
				FieldName: BoolFieldName,
				FieldId:   BoolFieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_BoolData{
							BoolData: &schemapb.BoolArray{
								Data: []bool{true, false, true, false},
							},
						},
					},
				},
			},
			{
				Type:      schemapb.DataType_Int64,
				FieldName: Int64FieldName,
				FieldId:   Int64FieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_LongData{
							LongData: &schemapb.LongArray{
								Data: []int64{1, 2, 3, 4},
							},
						},
					},
				},
			},
			{
				Type:      schemapb.DataType_VarChar,
				FieldName: StringFieldName,
				FieldId:   StringFieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_StringData{
							StringData: &schemapb.StringArray{
								Data: []string{"a", "b", "c", "d"},
							},
						},
					},
				},
			},
		}

		// Create update data (only update some fields)
		updateData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Bool,
				FieldName: BoolFieldName,
				FieldId:   BoolFieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_BoolData{
							BoolData: &schemapb.BoolArray{
								Data: []bool{false, true, false, true}, // Updated values
							},
						},
					},
				},
			},
			{
				Type:      schemapb.DataType_VarChar,
				FieldName: StringFieldName,
				FieldId:   StringFieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_StringData{
							StringData: &schemapb.StringArray{
								Data: []string{"x", "y", "z", "w"}, // Updated values
							},
						},
					},
				},
			},
			// Note: Int64Field is not in update, so it should remain unchanged
		}

		// Update index 1
		err := UpdateFieldData(baseData, updateData, 1, 1)
		require.NoError(t, err)

		// Check results
		// Bool field should be updated at index 1
		assert.Equal(t, true, baseData[0].GetScalars().GetBoolData().Data[1])
		// Int64 field should remain unchanged at index 1
		assert.Equal(t, int64(2), baseData[1].GetScalars().GetLongData().Data[1])
		// String field should be updated at index 1
		assert.Equal(t, "y", baseData[2].GetScalars().GetStringData().Data[1])

		// Other indices should remain unchanged
		assert.Equal(t, true, baseData[0].GetScalars().GetBoolData().Data[0])
		assert.Equal(t, int64(1), baseData[1].GetScalars().GetLongData().Data[0])
		assert.Equal(t, "a", baseData[2].GetScalars().GetStringData().Data[0])
	})

	t.Run("update vector fields", func(t *testing.T) {
		// Create base vector data
		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_FloatVector,
				FieldName: FloatVectorFieldName,
				FieldId:   FloatVectorFieldID,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim: Dim,
						Data: &schemapb.VectorField_FloatVector{
							FloatVector: &schemapb.FloatArray{
								Data: []float32{
									1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, // vector 0
									9.0, 10.0, 11.0, 12.0, 13.0, 14.0, 15.0, 16.0, // vector 1
								},
							},
						},
					},
				},
			},
		}

		// Create update data
		updateData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_FloatVector,
				FieldName: FloatVectorFieldName,
				FieldId:   FloatVectorFieldID,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim: Dim,
						Data: &schemapb.VectorField_FloatVector{
							FloatVector: &schemapb.FloatArray{
								Data: []float32{
									100.0, 200.0, 300.0, 400.0, 500.0, 600.0, 700.0, 800.0, // vector 0
									900.0, 1000.0, 1100.0, 1200.0, 1300.0, 1400.0, 1500.0, 1600.0, // vector 1
								},
							},
						},
					},
				},
			},
		}

		// Update index 1 (second vector)
		err := UpdateFieldData(baseData, updateData, 1, 1)
		require.NoError(t, err)

		// Check results
		vectorData := baseData[0].GetVectors().GetFloatVector().Data

		// First vector should remain unchanged
		for i := 0; i < Dim; i++ {
			assert.Equal(t, float32(i+1), vectorData[i])
		}

		// Second vector should be updated
		for i := 0; i < Dim; i++ {
			assert.Equal(t, float32(900+i*100), vectorData[Dim+i])
		}
	})

	t.Run("no update fields", func(t *testing.T) {
		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Int64,
				FieldName: Int64FieldName,
				FieldId:   Int64FieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_LongData{
							LongData: &schemapb.LongArray{
								Data: []int64{1, 2, 3, 4},
							},
						},
					},
				},
			},
		}

		// Empty update data
		updateData := []*schemapb.FieldData{}

		// Update should succeed but change nothing
		err := UpdateFieldData(baseData, updateData, 1, 1)
		require.NoError(t, err)

		// Data should remain unchanged
		assert.Equal(t, int64(2), baseData[0].GetScalars().GetLongData().Data[1])
	})

	t.Run("update with ValidData", func(t *testing.T) {
		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Int64,
				FieldName: Int64FieldName,
				FieldId:   Int64FieldID,
				ValidData: []bool{true, true, true, true},
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_LongData{
							LongData: &schemapb.LongArray{
								Data: []int64{1, 2, 3, 4},
							},
						},
					},
				},
			},
		}

		updateData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Int64,
				FieldName: Int64FieldName,
				FieldId:   Int64FieldID,
				ValidData: []bool{false, false, true, false},
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_LongData{
							LongData: &schemapb.LongArray{
								Data: []int64{0, 0, 30, 0},
							},
						},
					},
				},
			},
		}

		// Update index 1
		err := UpdateFieldData(baseData, updateData, 1, 1)
		require.NoError(t, err)
		err = UpdateFieldData(baseData, updateData, 2, 2)
		require.NoError(t, err)

		// Check that ValidData was updated
		assert.Equal(t, false, baseData[0].ValidData[1])
		// Check that data was updated
		assert.Equal(t, int64(0), baseData[0].GetScalars().GetLongData().Data[1])
		assert.Equal(t, int64(30), baseData[0].GetScalars().GetLongData().Data[2])
	})

	t.Run("update dynamic json field", func(t *testing.T) {
		// Create base data with dynamic JSON field
		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_JSON,
				FieldName: "json_field",
				FieldId:   1,
				IsDynamic: true,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_JsonData{
							JsonData: &schemapb.JSONArray{
								Data: [][]byte{
									[]byte(`{"key1": "value1", "key2": 123}`),
									[]byte(`{"key3": true, "key4": 456.789}`),
								},
							},
						},
					},
				},
			},
		}

		// Create update data
		updateData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_JSON,
				FieldName: "json_field",
				FieldId:   1,
				IsDynamic: true,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_JsonData{
							JsonData: &schemapb.JSONArray{
								Data: [][]byte{
									[]byte(`{"key2": 999, "key5": "new_value"}`),
									[]byte(`{"key4": 111.222, "key6": false}`),
								},
							},
						},
					},
				},
			},
		}

		// Test updating first row
		err := UpdateFieldData(baseData, updateData, 0, 0)
		require.NoError(t, err)

		// Verify first row was correctly merged
		firstRow := baseData[0].GetScalars().GetJsonData().Data[0]
		var result map[string]interface{}
		err = json.Unmarshal(firstRow, &result)
		require.NoError(t, err)

		// Check merged values
		assert.Equal(t, "value1", result["key1"])
		assert.Equal(t, float64(999), result["key2"]) // Updated value
		assert.Equal(t, "new_value", result["key5"])  // New value

		// Test updating second row
		err = UpdateFieldData(baseData, updateData, 1, 1)
		require.NoError(t, err)

		// Verify second row was correctly merged
		secondRow := baseData[0].GetScalars().GetJsonData().Data[1]
		err = json.Unmarshal(secondRow, &result)
		require.NoError(t, err)

		// Check merged values
		assert.Equal(t, true, result["key3"])
		assert.Equal(t, float64(111.222), result["key4"]) // Updated value
		assert.Equal(t, false, result["key6"])            // New value
	})

	t.Run("update non-dynamic json field", func(t *testing.T) {
		// Create base data with non-dynamic JSON field
		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_JSON,
				FieldName: "json_field",
				FieldId:   1,
				IsDynamic: false,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_JsonData{
							JsonData: &schemapb.JSONArray{
								Data: [][]byte{
									[]byte(`{"key1": "value1"}`),
								},
							},
						},
					},
				},
			},
		}

		// Create update data
		updateData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_JSON,
				FieldName: "json_field",
				FieldId:   1,
				IsDynamic: false,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_JsonData{
							JsonData: &schemapb.JSONArray{
								Data: [][]byte{
									[]byte(`{"key2": "value2"}`),
								},
							},
						},
					},
				},
			},
		}

		// Test updating
		err := UpdateFieldData(baseData, updateData, 0, 0)
		require.NoError(t, err)

		// For non-dynamic fields, the update should completely replace the old value
		result := baseData[0].GetScalars().GetJsonData().Data[0]
		assert.Equal(t, []byte(`{"key2": "value2"}`), result)
	})

	t.Run("invalid json data", func(t *testing.T) {
		// Create base data with invalid JSON
		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_JSON,
				FieldName: "json_field",
				FieldId:   1,
				IsDynamic: true,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_JsonData{
							JsonData: &schemapb.JSONArray{
								Data: [][]byte{
									[]byte(`invalid json`),
								},
							},
						},
					},
				},
			},
		}

		updateData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_JSON,
				FieldName: "json_field",
				FieldId:   1,
				IsDynamic: true,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_JsonData{
							JsonData: &schemapb.JSONArray{
								Data: [][]byte{
									[]byte(`{"key": "value"}`),
								},
							},
						},
					},
				},
			},
		}

		// Test updating with invalid base JSON
		err := UpdateFieldData(baseData, updateData, 0, 0)
		require.Error(t, err)
		assert.Contains(t, err.Error(), "failed to unmarshal base json")

		// Create base data with valid JSON but invalid update
		baseData[0].GetScalars().GetJsonData().Data[0] = []byte(`{"key": "value"}`)
		updateData[0].GetScalars().GetJsonData().Data[0] = []byte(`invalid json`)

		// Test updating with invalid update JSON
		err = UpdateFieldData(baseData, updateData, 0, 0)
		require.Error(t, err)
		assert.Contains(t, err.Error(), "failed to unmarshal update json")
	})

	t.Run("nullable field with valid data index mapping", func(t *testing.T) {
		// Test the new logic for nullable fields where updateIdx needs to be mapped to actual data index
		// Scenario: data=[1,2,3], valid_data=[true, false, true]

		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Int64,
				FieldName: "nullable_int_field",
				FieldId:   1,
				ValidData: []bool{true, true, true}, // All base data is valid
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_LongData{
							LongData: &schemapb.LongArray{
								Data: []int64{100, 200, 300},
							},
						},
					},
				},
			},
		}

		updateData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Int64,
				FieldName: "nullable_int_field",
				FieldId:   1,
				ValidData: []bool{true, false, true}, // Only indices 0 and 2 are valid
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_LongData{
							LongData: &schemapb.LongArray{
								Data: []int64{999, 0, 777},
							},
						},
					},
				},
			},
		}

		// Test updating at index 1
		err := UpdateFieldData(baseData, updateData, 0, 1)
		require.NoError(t, err)

		// Since valid_data[1] = false, no data should be updated
		assert.Equal(t, int64(0), baseData[0].GetScalars().GetLongData().Data[0])
		assert.Equal(t, false, baseData[0].ValidData[0])

		// Test updating at index 2
		err = UpdateFieldData(baseData, updateData, 1, 2)
		require.NoError(t, err)

		assert.Equal(t, int64(777), baseData[0].GetScalars().GetLongData().Data[1])
		assert.Equal(t, true, baseData[0].ValidData[1])
	})

	t.Run("nullable field with complex valid data pattern", func(t *testing.T) {
		// Test more complex pattern: data=[1,2,3,4,5], valid_data=[false, true, false, true, false]
		// This tests the index mapping logic more thoroughly

		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Float,
				FieldName: "complex_nullable_field",
				FieldId:   2,
				ValidData: []bool{true, true, true, true, true},
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_FloatData{
							FloatData: &schemapb.FloatArray{
								Data: []float32{1.1, 2.2, 3.3, 4.4, 5.5},
							},
						},
					},
				},
			},
		}

		updateData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Float,
				FieldName: "complex_nullable_field",
				FieldId:   2,
				ValidData: []bool{false, true, false, true, false}, // Only indices 1 and 3 are valid
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_FloatData{
							FloatData: &schemapb.FloatArray{
								Data: []float32{0, 999.9, 0, 888.8, 0},
							},
						},
					},
				},
			},
		}

		// Test updating at index 1
		err := UpdateFieldData(baseData, updateData, 1, 1)
		require.NoError(t, err)
		assert.Equal(t, float32(999.9), baseData[0].GetScalars().GetFloatData().Data[1])
		assert.Equal(t, true, baseData[0].ValidData[0])

		// Test updating at index 3
		err = UpdateFieldData(baseData, updateData, 3, 3)
		require.NoError(t, err)
		assert.Equal(t, float32(888.8), baseData[0].GetScalars().GetFloatData().Data[3])
		assert.Equal(t, true, baseData[0].ValidData[1])

		// Test updating at index 0
		err = UpdateFieldData(baseData, updateData, 2, 2)
		require.NoError(t, err)
		assert.Equal(t, float32(0), baseData[0].GetScalars().GetFloatData().Data[2])
		assert.Equal(t, false, baseData[0].ValidData[2])
	})
}

// TestUpdateFieldData_BoundsChecking tests the enhanced bounds checking for UpdateFieldData function
func TestUpdateFieldData_BoundsChecking(t *testing.T) {
	const (
		Dim = 8
	)

	// Helper function to create test scalar field data
	createScalarFieldData := func(fieldName string, fieldID int64, dataType schemapb.DataType, data interface{}) *schemapb.FieldData {
		fieldData := &schemapb.FieldData{
			Type:      dataType,
			FieldName: fieldName,
			FieldId:   fieldID,
		}

		switch dataType {
		case schemapb.DataType_Bool:
			fieldData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_BoolData{
						BoolData: &schemapb.BoolArray{Data: data.([]bool)},
					},
				},
			}
		case schemapb.DataType_Int32:
			fieldData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{Data: data.([]int32)},
					},
				},
			}
		case schemapb.DataType_Int64:
			fieldData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_LongData{
						LongData: &schemapb.LongArray{Data: data.([]int64)},
					},
				},
			}
		case schemapb.DataType_Float:
			fieldData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_FloatData{
						FloatData: &schemapb.FloatArray{Data: data.([]float32)},
					},
				},
			}
		case schemapb.DataType_Double:
			fieldData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_DoubleData{
						DoubleData: &schemapb.DoubleArray{Data: data.([]float64)},
					},
				},
			}
		case schemapb.DataType_VarChar:
			fieldData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_StringData{
						StringData: &schemapb.StringArray{Data: data.([]string)},
					},
				},
			}
		case schemapb.DataType_Array:
			fieldData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_ArrayData{
						ArrayData: &schemapb.ArrayArray{
							Data:        data.([]*schemapb.ScalarField),
							ElementType: schemapb.DataType_Int32,
						},
					},
				},
			}
		case schemapb.DataType_JSON:
			fieldData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_JsonData{
						JsonData: &schemapb.JSONArray{Data: data.([][]byte)},
					},
				},
			}
		}
		return fieldData
	}

	// Helper function to create test vector field data (unused in this test, but kept for consistency)
	_ = func(fieldName string, fieldID int64, dataType schemapb.DataType, data interface{}, dim int64) *schemapb.FieldData {
		fieldData := &schemapb.FieldData{
			Type:      dataType,
			FieldName: fieldName,
			FieldId:   fieldID,
		}

		switch dataType {
		case schemapb.DataType_BinaryVector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_BinaryVector{
						BinaryVector: data.([]byte),
					},
				},
			}
		case schemapb.DataType_FloatVector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_FloatVector{
						FloatVector: &schemapb.FloatArray{Data: data.([]float32)},
					},
				},
			}
		case schemapb.DataType_Float16Vector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Float16Vector{
						Float16Vector: data.([]byte),
					},
				},
			}
		case schemapb.DataType_BFloat16Vector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Bfloat16Vector{
						Bfloat16Vector: data.([]byte),
					},
				},
			}
		case schemapb.DataType_SparseFloatVector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_SparseFloatVector{
						SparseFloatVector: data.(*schemapb.SparseFloatArray),
					},
				},
			}
		case schemapb.DataType_Int8Vector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Int8Vector{
						Int8Vector: data.([]byte),
					},
				},
			}
		}
		return fieldData
	}

	t.Run("scalar field bounds checking - baseIdx out of bounds", func(t *testing.T) {
		// Test case: baseIdx is out of bounds for base data
		baseData := []*schemapb.FieldData{
			createScalarFieldData("bool_field", 1, schemapb.DataType_Bool, []bool{true, false}),
		}
		updateData := []*schemapb.FieldData{
			createScalarFieldData("bool_field", 1, schemapb.DataType_Bool, []bool{false, true}),
		}

		// baseIdx = 2 is out of bounds (base data only has 2 elements: indices 0,1)
		err := UpdateFieldData(baseData, updateData, 2, 0)
		require.NoError(t, err) // Should not panic, just skip update

		// Original data should remain unchanged
		assert.Equal(t, []bool{true, false}, baseData[0].GetScalars().GetBoolData().Data)
	})

	t.Run("scalar field bounds checking - updateIdx out of bounds", func(t *testing.T) {
		// Test case: updateIdx is out of bounds for update data
		baseData := []*schemapb.FieldData{
			createScalarFieldData("int_field", 1, schemapb.DataType_Int32, []int32{1, 2, 3}),
		}
		updateData := []*schemapb.FieldData{
			createScalarFieldData("int_field", 1, schemapb.DataType_Int32, []int32{10, 20}), // Only 2 elements
		}

		// updateIdx = 2 is out of bounds (update data only has 2 elements: indices 0,1)
		err := UpdateFieldData(baseData, updateData, 1, 2)
		require.NoError(t, err) // Should not panic, just skip update

		// Original data should remain unchanged
		assert.Equal(t, []int32{1, 2, 3}, baseData[0].GetScalars().GetIntData().Data)
	})

	t.Run("scalar field bounds checking - both indices in bounds", func(t *testing.T) {
		// Test case: both indices are in bounds, update should succeed
		baseData := []*schemapb.FieldData{
			createScalarFieldData("long_field", 1, schemapb.DataType_Int64, []int64{1, 2, 3, 4}),
		}
		updateData := []*schemapb.FieldData{
			createScalarFieldData("long_field", 1, schemapb.DataType_Int64, []int64{10, 20, 30, 40}),
		}

		// Both indices are in bounds
		err := UpdateFieldData(baseData, updateData, 2, 1)
		require.NoError(t, err)

		// Data at baseIdx=2 should be updated with data from updateIdx=1
		assert.Equal(t, int64(20), baseData[0].GetScalars().GetLongData().Data[2])
		// Other data should remain unchanged
		assert.Equal(t, int64(1), baseData[0].GetScalars().GetLongData().Data[0])
		assert.Equal(t, int64(2), baseData[0].GetScalars().GetLongData().Data[1])
		assert.Equal(t, int64(4), baseData[0].GetScalars().GetLongData().Data[3])
	})

	t.Run("scalar field nil data checking - baseData is nil", func(t *testing.T) {
		// Test case: baseData is nil
		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Float,
				FieldName: "float_field",
				FieldId:   1,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_FloatData{
							FloatData: nil, // baseData is nil
						},
					},
				},
			},
		}
		updateData := []*schemapb.FieldData{
			createScalarFieldData("float_field", 1, schemapb.DataType_Float, []float32{1.1, 2.2}),
		}

		err := UpdateFieldData(baseData, updateData, 0, 0)
		require.NoError(t, err) // Should not panic, just skip update
	})

	t.Run("scalar field nil data checking - updateData is nil", func(t *testing.T) {
		// Test case: updateData is nil
		baseData := []*schemapb.FieldData{
			createScalarFieldData("double_field", 1, schemapb.DataType_Double, []float64{1.1, 2.2}),
		}
		updateData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Double,
				FieldName: "double_field",
				FieldId:   1,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_DoubleData{
							DoubleData: nil, // updateData is nil
						},
					},
				},
			},
		}

		err := UpdateFieldData(baseData, updateData, 0, 0)
		require.NoError(t, err) // Should not panic, just skip update

		// Original data should remain unchanged
		assert.Equal(t, []float64{1.1, 2.2}, baseData[0].GetScalars().GetDoubleData().Data)
	})

	t.Run("test all scalar field types bounds checking", func(t *testing.T) {
		// Test all scalar field types with bounds checking
		scalarTypes := []struct {
			name       string
			dataType   schemapb.DataType
			baseData   interface{}
			updateData interface{}
		}{
			{
				name:       "BoolData",
				dataType:   schemapb.DataType_Bool,
				baseData:   []bool{true, false, true},
				updateData: []bool{false, true, false},
			},
			{
				name:       "IntData",
				dataType:   schemapb.DataType_Int32,
				baseData:   []int32{1, 2, 3},
				updateData: []int32{10, 20, 30},
			},
			{
				name:       "LongData",
				dataType:   schemapb.DataType_Int64,
				baseData:   []int64{1, 2, 3},
				updateData: []int64{10, 20, 30},
			},
			{
				name:       "FloatData",
				dataType:   schemapb.DataType_Float,
				baseData:   []float32{1.1, 2.2, 3.3},
				updateData: []float32{10.1, 20.2, 30.3},
			},
			{
				name:       "DoubleData",
				dataType:   schemapb.DataType_Double,
				baseData:   []float64{1.1, 2.2, 3.3},
				updateData: []float64{10.1, 20.2, 30.3},
			},
			{
				name:       "StringData",
				dataType:   schemapb.DataType_VarChar,
				baseData:   []string{"a", "b", "c"},
				updateData: []string{"x", "y", "z"},
			},
		}

		for _, testCase := range scalarTypes {
			t.Run(testCase.name, func(t *testing.T) {
				baseData := []*schemapb.FieldData{
					createScalarFieldData("test_field", 1, testCase.dataType, testCase.baseData),
				}
				updateData := []*schemapb.FieldData{
					createScalarFieldData("test_field", 1, testCase.dataType, testCase.updateData),
				}

				// Test bounds checking - baseIdx out of bounds
				err := UpdateFieldData(baseData, updateData, 3, 0) // baseIdx=3 is out of bounds
				require.NoError(t, err)

				// Test bounds checking - updateIdx out of bounds
				err = UpdateFieldData(baseData, updateData, 0, 3) // updateIdx=3 is out of bounds
				require.NoError(t, err)

				// Test successful update
				err = UpdateFieldData(baseData, updateData, 1, 1)
				require.NoError(t, err)

				// Verify the update worked correctly
				switch testCase.dataType {
				case schemapb.DataType_Bool:
					assert.Equal(t, true, baseData[0].GetScalars().GetBoolData().Data[1])
				case schemapb.DataType_Int32:
					assert.Equal(t, int32(20), baseData[0].GetScalars().GetIntData().Data[1])
				case schemapb.DataType_Int64:
					assert.Equal(t, int64(20), baseData[0].GetScalars().GetLongData().Data[1])
				case schemapb.DataType_Float:
					assert.Equal(t, float32(20.2), baseData[0].GetScalars().GetFloatData().Data[1])
				case schemapb.DataType_Double:
					assert.Equal(t, float64(20.2), baseData[0].GetScalars().GetDoubleData().Data[1])
				case schemapb.DataType_VarChar:
					assert.Equal(t, "y", baseData[0].GetScalars().GetStringData().Data[1])
				}
			})
		}
	})
}

// TestUpdateFieldData_VectorBoundsChecking tests the enhanced bounds checking for vector fields in UpdateFieldData function
func TestUpdateFieldData_VectorBoundsChecking(t *testing.T) {
	const (
		Dim = 8
	)

	// Helper function to create test vector field data
	createVectorFieldData := func(fieldName string, fieldID int64, dataType schemapb.DataType, data interface{}, dim int64) *schemapb.FieldData {
		fieldData := &schemapb.FieldData{
			Type:      dataType,
			FieldName: fieldName,
			FieldId:   fieldID,
		}

		switch dataType {
		case schemapb.DataType_BinaryVector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_BinaryVector{
						BinaryVector: data.([]byte),
					},
				},
			}
		case schemapb.DataType_FloatVector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_FloatVector{
						FloatVector: &schemapb.FloatArray{Data: data.([]float32)},
					},
				},
			}
		case schemapb.DataType_Float16Vector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Float16Vector{
						Float16Vector: data.([]byte),
					},
				},
			}
		case schemapb.DataType_BFloat16Vector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Bfloat16Vector{
						Bfloat16Vector: data.([]byte),
					},
				},
			}
		case schemapb.DataType_SparseFloatVector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_SparseFloatVector{
						SparseFloatVector: data.(*schemapb.SparseFloatArray),
					},
				},
			}
		case schemapb.DataType_Int8Vector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Int8Vector{
						Int8Vector: data.([]byte),
					},
				},
			}
		}
		return fieldData
	}

	t.Run("vector field bounds checking - baseIdx out of bounds", func(t *testing.T) {
		// Test case: baseIdx is out of bounds for base vector data
		baseData := []*schemapb.FieldData{
			createVectorFieldData("float_vector_field", 1, schemapb.DataType_FloatVector,
				[]float32{1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0}, Dim), // Only 1 vector
		}
		updateData := []*schemapb.FieldData{
			createVectorFieldData("float_vector_field", 1, schemapb.DataType_FloatVector,
				[]float32{10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0}, Dim),
		}

		// baseIdx = 1 is out of bounds (base data only has 1 vector: index 0)
		err := UpdateFieldData(baseData, updateData, 1, 0)
		require.NoError(t, err) // Should not panic, just skip update

		// Original data should remain unchanged
		originalVector := baseData[0].GetVectors().GetFloatVector().Data
		for i := 0; i < Dim; i++ {
			assert.Equal(t, float32(i+1), originalVector[i])
		}
	})

	t.Run("vector field bounds checking - updateIdx out of bounds", func(t *testing.T) {
		// Test case: updateIdx is out of bounds for update vector data
		baseData := []*schemapb.FieldData{
			createVectorFieldData("binary_vector_field", 1, schemapb.DataType_BinaryVector,
				make([]byte, 2*Dim/8), Dim), // 2 vectors
		}
		updateData := []*schemapb.FieldData{
			createVectorFieldData("binary_vector_field", 1, schemapb.DataType_BinaryVector,
				make([]byte, 1*Dim/8), Dim), // Only 1 vector
		}

		// updateIdx = 1 is out of bounds (update data only has 1 vector: index 0)
		err := UpdateFieldData(baseData, updateData, 0, 1)
		require.NoError(t, err) // Should not panic, just skip update
	})

	t.Run("vector field bounds checking - both indices in bounds", func(t *testing.T) {
		// Test case: both indices are in bounds, update should succeed
		baseData := []*schemapb.FieldData{
			createVectorFieldData("float_vector_field", 1, schemapb.DataType_FloatVector,
				[]float32{
					1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, // vector 0
					9.0, 10.0, 11.0, 12.0, 13.0, 14.0, 15.0, 16.0, // vector 1
				}, Dim),
		}
		updateData := []*schemapb.FieldData{
			createVectorFieldData("float_vector_field", 1, schemapb.DataType_FloatVector,
				[]float32{
					100.0, 200.0, 300.0, 400.0, 500.0, 600.0, 700.0, 800.0, // vector 0
					900.0, 1000.0, 1100.0, 1200.0, 1300.0, 1400.0, 1500.0, 1600.0, // vector 1
				}, Dim),
		}

		// Both indices are in bounds
		err := UpdateFieldData(baseData, updateData, 1, 1)
		require.NoError(t, err)

		// Vector at baseIdx=1 should be updated with vector from updateIdx=1
		updatedVector := baseData[0].GetVectors().GetFloatVector().Data
		for i := 0; i < Dim; i++ {
			assert.Equal(t, float32(900+i*100), updatedVector[Dim+i])
		}
		// Vector at baseIdx=0 should remain unchanged
		for i := 0; i < Dim; i++ {
			assert.Equal(t, float32(i+1), updatedVector[i])
		}
	})

	t.Run("vector field nil data checking - baseData is nil", func(t *testing.T) {
		// Test case: baseData is nil
		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_FloatVector,
				FieldName: "float_vector_field",
				FieldId:   1,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim: Dim,
						Data: &schemapb.VectorField_FloatVector{
							FloatVector: nil, // baseData is nil
						},
					},
				},
			},
		}
		updateData := []*schemapb.FieldData{
			createVectorFieldData("float_vector_field", 1, schemapb.DataType_FloatVector,
				[]float32{1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0}, Dim),
		}

		err := UpdateFieldData(baseData, updateData, 0, 0)
		require.NoError(t, err) // Should not panic, just skip update
	})

	t.Run("vector field nil data checking - updateData is nil", func(t *testing.T) {
		// Test case: updateData is nil
		baseData := []*schemapb.FieldData{
			createVectorFieldData("binary_vector_field", 1, schemapb.DataType_BinaryVector,
				make([]byte, Dim/8), Dim),
		}
		updateData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_BinaryVector,
				FieldName: "binary_vector_field",
				FieldId:   1,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim: Dim,
						Data: &schemapb.VectorField_BinaryVector{
							BinaryVector: nil, // updateData is nil
						},
					},
				},
			},
		}

		err := UpdateFieldData(baseData, updateData, 0, 0)
		require.NoError(t, err) // Should not panic, just skip update
	})

	t.Run("test all vector field types bounds checking", func(t *testing.T) {
		// Test all vector field types with bounds checking
		vectorTypes := []struct {
			name       string
			dataType   schemapb.DataType
			baseData   interface{}
			updateData interface{}
		}{
			{
				name:       "BinaryVector",
				dataType:   schemapb.DataType_BinaryVector,
				baseData:   make([]byte, 2*Dim/8), // 2 vectors
				updateData: make([]byte, 2*Dim/8), // 2 vectors
			},
			{
				name:       "FloatVector",
				dataType:   schemapb.DataType_FloatVector,
				baseData:   []float32{1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0, 11.0, 12.0, 13.0, 14.0, 15.0, 16.0},                 // 2 vectors
				updateData: []float32{10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0, 110.0, 120.0, 130.0, 140.0, 150.0, 160.0}, // 2 vectors
			},
			{
				name:       "Float16Vector",
				dataType:   schemapb.DataType_Float16Vector,
				baseData:   make([]byte, 2*Dim*2), // 2 vectors
				updateData: make([]byte, 2*Dim*2), // 2 vectors
			},
			{
				name:       "BFloat16Vector",
				dataType:   schemapb.DataType_BFloat16Vector,
				baseData:   make([]byte, 2*Dim*2), // 2 vectors
				updateData: make([]byte, 2*Dim*2), // 2 vectors
			},
			{
				name:       "Int8Vector",
				dataType:   schemapb.DataType_Int8Vector,
				baseData:   make([]byte, 2*Dim), // 2 vectors
				updateData: make([]byte, 2*Dim), // 2 vectors
			},
		}

		for _, testCase := range vectorTypes {
			t.Run(testCase.name, func(t *testing.T) {
				baseData := []*schemapb.FieldData{
					createVectorFieldData("test_vector_field", 1, testCase.dataType, testCase.baseData, Dim),
				}
				updateData := []*schemapb.FieldData{
					createVectorFieldData("test_vector_field", 1, testCase.dataType, testCase.updateData, Dim),
				}

				// Test bounds checking - baseIdx out of bounds
				err := UpdateFieldData(baseData, updateData, 2, 0) // baseIdx=2 is out of bounds
				require.NoError(t, err)

				// Test bounds checking - updateIdx out of bounds
				err = UpdateFieldData(baseData, updateData, 0, 2) // updateIdx=2 is out of bounds
				require.NoError(t, err)

				// Test successful update
				err = UpdateFieldData(baseData, updateData, 1, 1)
				require.NoError(t, err)
			})
		}
	})

	t.Run("sparse float vector bounds checking", func(t *testing.T) {
		// Create sparse float vector data
		baseSparseData := &schemapb.SparseFloatArray{
			Dim: 10,
			Contents: [][]byte{
				CreateSparseFloatRow([]uint32{1, 3, 5}, []float32{1.1, 3.3, 5.5}), // vector 0
				CreateSparseFloatRow([]uint32{2, 4, 6}, []float32{2.2, 4.4, 6.6}), // vector 1
			},
		}
		updateSparseData := &schemapb.SparseFloatArray{
			Dim: 10,
			Contents: [][]byte{
				CreateSparseFloatRow([]uint32{7, 9}, []float32{7.7, 9.9}),   // vector 0
				CreateSparseFloatRow([]uint32{8, 10}, []float32{8.8, 10.0}), // vector 1
			},
		}

		baseData := []*schemapb.FieldData{
			createVectorFieldData("sparse_vector_field", 1, schemapb.DataType_SparseFloatVector, baseSparseData, 10),
		}
		updateData := []*schemapb.FieldData{
			createVectorFieldData("sparse_vector_field", 1, schemapb.DataType_SparseFloatVector, updateSparseData, 10),
		}

		// Test bounds checking - baseIdx out of bounds
		err := UpdateFieldData(baseData, updateData, 2, 0) // baseIdx=2 is out of bounds
		require.NoError(t, err)

		// Test bounds checking - updateIdx out of bounds
		err = UpdateFieldData(baseData, updateData, 0, 2) // updateIdx=2 is out of bounds
		require.NoError(t, err)

		// Test successful update
		err = UpdateFieldData(baseData, updateData, 1, 1)
		require.NoError(t, err)

		// Verify the update worked correctly
		updatedContents := baseData[0].GetVectors().GetSparseFloatVector().Contents
		assert.Equal(t, updateSparseData.Contents[1], updatedContents[1])
	})
}

// TestUpdateFieldData_IndexFix tests the index fix for vector fields in UpdateFieldData function
// This test verifies that baseIdx is used correctly for base data indexing instead of updateIdx
func TestUpdateFieldData_IndexFix(t *testing.T) {
	const (
		Dim = 8
	)

	// Helper function to create test vector field data
	createVectorFieldData := func(fieldName string, fieldID int64, dataType schemapb.DataType, data interface{}, dim int64) *schemapb.FieldData {
		fieldData := &schemapb.FieldData{
			Type:      dataType,
			FieldName: fieldName,
			FieldId:   fieldID,
		}

		switch dataType {
		case schemapb.DataType_BinaryVector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_BinaryVector{
						BinaryVector: data.([]byte),
					},
				},
			}
		case schemapb.DataType_FloatVector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_FloatVector{
						FloatVector: &schemapb.FloatArray{Data: data.([]float32)},
					},
				},
			}
		case schemapb.DataType_Float16Vector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Float16Vector{
						Float16Vector: data.([]byte),
					},
				},
			}
		case schemapb.DataType_BFloat16Vector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Bfloat16Vector{
						Bfloat16Vector: data.([]byte),
					},
				},
			}
		case schemapb.DataType_SparseFloatVector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_SparseFloatVector{
						SparseFloatVector: data.(*schemapb.SparseFloatArray),
					},
				},
			}
		case schemapb.DataType_Int8Vector:
			fieldData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Int8Vector{
						Int8Vector: data.([]byte),
					},
				},
			}
		}
		return fieldData
	}

	t.Run("vector field index fix - binary vector", func(t *testing.T) {
		// Test case: Verify that baseIdx is used for base data indexing, not updateIdx
		// Create base data with 3 vectors
		baseData := []*schemapb.FieldData{
			createVectorFieldData("binary_vector_field", 1, schemapb.DataType_BinaryVector,
				make([]byte, 3*Dim/8), Dim), // 3 vectors
		}
		// Create update data with 2 vectors
		updateData := []*schemapb.FieldData{
			createVectorFieldData("binary_vector_field", 1, schemapb.DataType_BinaryVector,
				make([]byte, 2*Dim/8), Dim), // 2 vectors
		}

		// Fill update data with specific pattern
		updateBytes := updateData[0].GetVectors().GetBinaryVector()
		for i := range updateBytes {
			updateBytes[i] = 0xFF // Fill with 0xFF pattern
		}

		// Update baseIdx=2 with updateIdx=1
		// This should update the 3rd vector in base data with the 2nd vector from update data
		err := UpdateFieldData(baseData, updateData, 2, 1)
		require.NoError(t, err)

		// Verify that the correct vector was updated
		baseBytes := baseData[0].GetVectors().GetBinaryVector()
		baseStartIdx := 2 * (Dim / 8)
		updateStartIdx := 1 * (Dim / 8)

		// Check that the 3rd vector in base data matches the 2nd vector from update data
		for i := 0; i < Dim/8; i++ {
			assert.Equal(t, updateBytes[updateStartIdx+i], baseBytes[baseStartIdx+i])
		}
	})

	t.Run("vector field index fix - float vector", func(t *testing.T) {
		// Test case: Verify that baseIdx is used for base data indexing, not updateIdx
		baseData := []*schemapb.FieldData{
			createVectorFieldData("float_vector_field", 1, schemapb.DataType_FloatVector,
				[]float32{
					1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, // vector 0
					9.0, 10.0, 11.0, 12.0, 13.0, 14.0, 15.0, 16.0, // vector 1
					17.0, 18.0, 19.0, 20.0, 21.0, 22.0, 23.0, 24.0, // vector 2
				}, Dim),
		}
		updateData := []*schemapb.FieldData{
			createVectorFieldData("float_vector_field", 1, schemapb.DataType_FloatVector,
				[]float32{
					100.0, 200.0, 300.0, 400.0, 500.0, 600.0, 700.0, 800.0, // vector 0
					900.0, 1000.0, 1100.0, 1200.0, 1300.0, 1400.0, 1500.0, 1600.0, // vector 1
				}, Dim),
		}

		// Update baseIdx=2 with updateIdx=1
		// This should update the 3rd vector in base data with the 2nd vector from update data
		err := UpdateFieldData(baseData, updateData, 2, 1)
		require.NoError(t, err)

		// Verify that the correct vector was updated
		baseVectorData := baseData[0].GetVectors().GetFloatVector().Data
		// Vector 0 should remain unchanged
		for i := 0; i < Dim; i++ {
			assert.Equal(t, float32(i+1), baseVectorData[i])
		}
		// Vector 1 should remain unchanged
		for i := 0; i < Dim; i++ {
			assert.Equal(t, float32(9+i), baseVectorData[Dim+i])
		}
		// Vector 2 should be updated with update vector 1
		for i := 0; i < Dim; i++ {
			assert.Equal(t, float32(900+i*100), baseVectorData[2*Dim+i])
		}
	})

	t.Run("vector field index fix - float16 vector", func(t *testing.T) {
		// Test case: Verify that baseIdx is used for base data indexing, not updateIdx
		baseData := []*schemapb.FieldData{
			createVectorFieldData("float16_vector_field", 1, schemapb.DataType_Float16Vector,
				make([]byte, 3*Dim*2), Dim), // 3 vectors
		}
		updateData := []*schemapb.FieldData{
			createVectorFieldData("float16_vector_field", 1, schemapb.DataType_Float16Vector,
				make([]byte, 2*Dim*2), Dim), // 2 vectors
		}

		// Fill update data with specific pattern
		updateBytes := updateData[0].GetVectors().GetFloat16Vector()
		for i := range updateBytes {
			updateBytes[i] = 0xAA // Fill with 0xAA pattern
		}

		// Update baseIdx=2 with updateIdx=1
		err := UpdateFieldData(baseData, updateData, 2, 1)
		require.NoError(t, err)

		// Verify that the correct vector was updated
		baseBytes := baseData[0].GetVectors().GetFloat16Vector()
		baseStartIdx := 2 * (Dim * 2)
		updateStartIdx := 1 * (Dim * 2)

		// Check that the 3rd vector in base data matches the 2nd vector from update data
		for i := 0; i < Dim*2; i++ {
			assert.Equal(t, updateBytes[updateStartIdx+i], baseBytes[baseStartIdx+i])
		}
	})

	t.Run("vector field index fix - bfloat16 vector", func(t *testing.T) {
		// Test case: Verify that baseIdx is used for base data indexing, not updateIdx
		baseData := []*schemapb.FieldData{
			createVectorFieldData("bfloat16_vector_field", 1, schemapb.DataType_BFloat16Vector,
				make([]byte, 3*Dim*2), Dim), // 3 vectors
		}
		updateData := []*schemapb.FieldData{
			createVectorFieldData("bfloat16_vector_field", 1, schemapb.DataType_BFloat16Vector,
				make([]byte, 2*Dim*2), Dim), // 2 vectors
		}

		// Fill update data with specific pattern
		updateBytes := updateData[0].GetVectors().GetBfloat16Vector()
		for i := range updateBytes {
			updateBytes[i] = 0xBB // Fill with 0xBB pattern
		}

		// Update baseIdx=2 with updateIdx=1
		err := UpdateFieldData(baseData, updateData, 2, 1)
		require.NoError(t, err)

		// Verify that the correct vector was updated
		baseBytes := baseData[0].GetVectors().GetBfloat16Vector()
		baseStartIdx := 2 * (Dim * 2)
		updateStartIdx := 1 * (Dim * 2)

		// Check that the 3rd vector in base data matches the 2nd vector from update data
		for i := 0; i < Dim*2; i++ {
			assert.Equal(t, updateBytes[updateStartIdx+i], baseBytes[baseStartIdx+i])
		}
	})

	t.Run("vector field index fix - int8 vector", func(t *testing.T) {
		// Test case: Verify that baseIdx is used for base data indexing, not updateIdx
		baseData := []*schemapb.FieldData{
			createVectorFieldData("int8_vector_field", 1, schemapb.DataType_Int8Vector,
				make([]byte, 3*Dim), Dim), // 3 vectors
		}
		updateData := []*schemapb.FieldData{
			createVectorFieldData("int8_vector_field", 1, schemapb.DataType_Int8Vector,
				make([]byte, 2*Dim), Dim), // 2 vectors
		}

		// Fill update data with specific pattern
		updateBytes := updateData[0].GetVectors().GetInt8Vector()
		for i := range updateBytes {
			updateBytes[i] = 0xCC // Fill with 0xCC pattern
		}

		// Update baseIdx=2 with updateIdx=1
		err := UpdateFieldData(baseData, updateData, 2, 1)
		require.NoError(t, err)

		// Verify that the correct vector was updated
		baseBytes := baseData[0].GetVectors().GetInt8Vector()
		baseStartIdx := 2 * Dim
		updateStartIdx := 1 * Dim

		// Check that the 3rd vector in base data matches the 2nd vector from update data
		for i := 0; i < Dim; i++ {
			assert.Equal(t, updateBytes[updateStartIdx+i], baseBytes[baseStartIdx+i])
		}
	})

	t.Run("vector field index fix - sparse float vector", func(t *testing.T) {
		// Test case: Verify that baseIdx is used for base data indexing, not updateIdx
		baseSparseData := &schemapb.SparseFloatArray{
			Dim: 10,
			Contents: [][]byte{
				CreateSparseFloatRow([]uint32{1, 3, 5}, []float32{1.1, 3.3, 5.5}), // vector 0
				CreateSparseFloatRow([]uint32{2, 4, 6}, []float32{2.2, 4.4, 6.6}), // vector 1
				CreateSparseFloatRow([]uint32{7, 9}, []float32{7.7, 9.9}),         // vector 2
			},
		}
		updateSparseData := &schemapb.SparseFloatArray{
			Dim: 10,
			Contents: [][]byte{
				CreateSparseFloatRow([]uint32{10, 12}, []float32{10.0, 12.0}), // vector 0
				CreateSparseFloatRow([]uint32{11, 13}, []float32{11.0, 13.0}), // vector 1
			},
		}

		baseData := []*schemapb.FieldData{
			createVectorFieldData("sparse_vector_field", 1, schemapb.DataType_SparseFloatVector, baseSparseData, 10),
		}
		updateData := []*schemapb.FieldData{
			createVectorFieldData("sparse_vector_field", 1, schemapb.DataType_SparseFloatVector, updateSparseData, 10),
		}

		// Update baseIdx=2 with updateIdx=1
		// This should update the 3rd vector in base data with the 2nd vector from update data
		err := UpdateFieldData(baseData, updateData, 2, 1)
		require.NoError(t, err)

		// Verify that the correct vector was updated
		updatedContents := baseData[0].GetVectors().GetSparseFloatVector().Contents
		// Vector 0 should remain unchanged
		assert.Equal(t, baseSparseData.Contents[0], updatedContents[0])
		// Vector 1 should remain unchanged
		assert.Equal(t, baseSparseData.Contents[1], updatedContents[1])
		// Vector 2 should be updated with update vector 1
		assert.Equal(t, updateSparseData.Contents[1], updatedContents[2])
	})
}

func TestGetNeedProcessFunctions(t *testing.T) {
	{
		f, err := GetNeedProcessFunctions([]int64{}, []*schemapb.FunctionSchema{}, false, false)
		assert.Len(t, f, 0)
		assert.NoError(t, err)
	}
	{
		fs := []*schemapb.FunctionSchema{{Name: "test_func", OutputFieldIds: []int64{1}}}
		_, err := GetNeedProcessFunctions([]int64{1, 2}, fs, false, false)
		assert.ErrorContains(t, err, "Insert data has function output field")
		f, err := GetNeedProcessFunctions([]int64{1, 2}, fs, true, false)
		assert.NoError(t, err)
		assert.Len(t, f, 0)
	}
	{
		fs := []*schemapb.FunctionSchema{{Name: "test_func", OutputFieldIds: []int64{1}}}
		_, err := GetNeedProcessFunctions([]int64{1}, fs, false, false)
		assert.ErrorContains(t, err, "Insert data has function output field")
		f, err := GetNeedProcessFunctions([]int64{1}, fs, true, false)
		assert.NoError(t, err)
		assert.Len(t, f, 0)
	}
	{
		fs := []*schemapb.FunctionSchema{{Name: "test_func", OutputFieldIds: []int64{1}}, {Name: "test_func2", OutputFieldIds: []int64{2}}}
		_, err := GetNeedProcessFunctions([]int64{1}, fs, false, false)
		assert.Error(t, err)
		f, err := GetNeedProcessFunctions([]int64{1}, fs, true, false)
		assert.NoError(t, err)
		assert.Len(t, f, 1)
		assert.Equal(t, f[0].Name, "test_func2")
	}
	{
		fs := []*schemapb.FunctionSchema{{Name: "test_func", Type: schemapb.FunctionType_BM25, OutputFieldIds: []int64{1}}}
		_, err := GetNeedProcessFunctions([]int64{1}, fs, true, false)
		assert.ErrorContains(t, err, "attempt to insert bm25 function output field")
	}
	{
		fs := []*schemapb.FunctionSchema{
			{Name: "test_func", InputFieldIds: []int64{1, 2}, OutputFieldIds: []int64{3}},
			{Name: "test_func2", InputFieldIds: []int64{1}, OutputFieldIds: []int64{2}},
		}
		_, err := GetNeedProcessFunctions([]int64{1, 2}, fs, false, true)
		assert.Error(t, err)
		f, err := GetNeedProcessFunctions([]int64{1, 2}, fs, true, true)
		assert.NoError(t, err)
		assert.Len(t, f, 1)
		assert.Equal(t, f[0].Name, "test_func")
	}
}

// Test UpdateFieldData for Geometry and Timestamptz types
func TestUpdateFieldData_GeometryAndTimestamptz(t *testing.T) {
	t.Run("update timestamptz field", func(t *testing.T) {
		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Timestamptz,
				FieldName: "timestamp_field",
				FieldId:   100,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_TimestamptzData{
							TimestamptzData: &schemapb.TimestamptzArray{
								Data: []int64{1000, 2000, 3000},
							},
						},
					},
				},
			},
		}

		updateData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Timestamptz,
				FieldName: "timestamp_field",
				FieldId:   100,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_TimestamptzData{
							TimestamptzData: &schemapb.TimestamptzArray{
								Data: []int64{9999, 8888, 7777},
							},
						},
					},
				},
			},
		}

		err := UpdateFieldData(baseData, updateData, 1, 1)
		assert.NoError(t, err)

		timestampData := baseData[0].GetScalars().GetTimestamptzData().GetData()
		assert.Equal(t, int64(1000), timestampData[0])
		assert.Equal(t, int64(8888), timestampData[1])
		assert.Equal(t, int64(3000), timestampData[2])
	})

	t.Run("update geometry WKT field", func(t *testing.T) {
		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Geometry,
				FieldName: "geometry_field",
				FieldId:   101,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_GeometryWktData{
							GeometryWktData: &schemapb.GeometryWktArray{
								Data: []string{"POINT (0 0)", "POINT (1 1)", "POINT (2 2)"},
							},
						},
					},
				},
			},
		}

		updateData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Geometry,
				FieldName: "geometry_field",
				FieldId:   101,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_GeometryWktData{
							GeometryWktData: &schemapb.GeometryWktArray{
								Data: []string{"POINT (10 10)", "POINT (11 11)", "POINT (12 12)"},
							},
						},
					},
				},
			},
		}

		err := UpdateFieldData(baseData, updateData, 1, 1)
		assert.NoError(t, err)

		geoData := baseData[0].GetScalars().GetGeometryWktData().GetData()
		assert.Equal(t, "POINT (0 0)", geoData[0])
		assert.Equal(t, "POINT (11 11)", geoData[1])
		assert.Equal(t, "POINT (2 2)", geoData[2])
	})

	t.Run("update geometry WKB field", func(t *testing.T) {
		baseData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Geometry,
				FieldName: "geometry_field",
				FieldId:   102,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_GeometryData{
							GeometryData: &schemapb.GeometryArray{
								Data: [][]byte{{0x01, 0x02}, {0x03, 0x04}, {0x05, 0x06}},
							},
						},
					},
				},
			},
		}

		updateData := []*schemapb.FieldData{
			{
				Type:      schemapb.DataType_Geometry,
				FieldName: "geometry_field",
				FieldId:   102,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_GeometryData{
							GeometryData: &schemapb.GeometryArray{
								Data: [][]byte{{0xFF, 0xFE}, {0xFD, 0xFC}, {0xFB, 0xFA}},
							},
						},
					},
				},
			},
		}

		err := UpdateFieldData(baseData, updateData, 2, 2)
		assert.NoError(t, err)

		geoData := baseData[0].GetScalars().GetGeometryData().GetData()
		assert.Equal(t, []byte{0x01, 0x02}, geoData[0])
		assert.Equal(t, []byte{0x03, 0x04}, geoData[1])
		assert.Equal(t, []byte{0xFB, 0xFA}, geoData[2])
	})
}

func TestIsBM25FunctionOutputField(t *testing.T) {
	schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", DataType: schemapb.DataType_VarChar, TypeParams: []*commonpb.KeyValuePair{{Key: "enable_analyzer", Value: "true"}}},
			{Name: "output_field", DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:             "bm25_func",
				Type:             schemapb.FunctionType_BM25,
				InputFieldNames:  []string{"input_field"},
				OutputFieldNames: []string{"output_field"},
			},
		},
	}
	assert.False(t, IsBM25FunctionOutputField(schema.Fields[0], schema))
	assert.True(t, IsBM25FunctionOutputField(schema.Fields[1], schema))

	// Test with field IDs instead of names
	schemaWithIds := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", FieldID: 100, DataType: schemapb.DataType_VarChar, TypeParams: []*commonpb.KeyValuePair{{Key: "enable_analyzer", Value: "true"}}},
			{Name: "output_field", FieldID: 101, DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:           "bm25_func",
				Type:           schemapb.FunctionType_BM25,
				InputFieldIds:  []int64{100},
				OutputFieldIds: []int64{101},
			},
		},
	}
	assert.False(t, IsBM25FunctionOutputField(schemaWithIds.Fields[0], schemaWithIds))
	assert.True(t, IsBM25FunctionOutputField(schemaWithIds.Fields[1], schemaWithIds))

	// Test with non-BM25 function
	nonBM25Schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", DataType: schemapb.DataType_VarChar},
			{Name: "output_field", DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:             "other_func",
				Type:             schemapb.FunctionType_Unknown,
				InputFieldNames:  []string{"input_field"},
				OutputFieldNames: []string{"output_field"},
			},
		},
	}
	assert.False(t, IsBM25FunctionOutputField(nonBM25Schema.Fields[1], nonBM25Schema))

	// Test with non-sparse vector field
	nonSparseSchema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", DataType: schemapb.DataType_VarChar},
			{Name: "output_field", DataType: schemapb.DataType_FloatVector, IsFunctionOutput: true},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:             "bm25_func",
				Type:             schemapb.FunctionType_BM25,
				InputFieldNames:  []string{"input_field"},
				OutputFieldNames: []string{"output_field"},
			},
		},
	}
	assert.False(t, IsBM25FunctionOutputField(nonSparseSchema.Fields[1], nonSparseSchema))

	// Function output ownership depends on the normalized field marker.
	unmarkedOutputSchema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", DataType: schemapb.DataType_VarChar},
			{Name: "output_field", DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: false},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:             "bm25_func",
				Type:             schemapb.FunctionType_BM25,
				InputFieldNames:  []string{"input_field"},
				OutputFieldNames: []string{"output_field"},
			},
		},
	}
	assert.False(t, IsBM25FunctionOutputField(unmarkedOutputSchema.Fields[1], unmarkedOutputSchema))
}

func TestIsBm25FunctionInputField(t *testing.T) {
	schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{FieldID: 100, Name: "input_field", DataType: schemapb.DataType_VarChar, TypeParams: []*commonpb.KeyValuePair{{Key: "enable_analyzer", Value: "true"}}},
			{FieldID: 101, Name: "output_field", DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:             "bm25_func",
				Type:             schemapb.FunctionType_BM25,
				InputFieldIds:    []int64{100},
				InputFieldNames:  []string{"input_field"},
				OutputFieldIds:   []int64{101},
				OutputFieldNames: []string{"output_field"},
			},
		},
	}
	assert.True(t, IsBm25FunctionInputField(schema, schema.Fields[0]))
	assert.False(t, IsBm25FunctionInputField(schema, schema.Fields[1]))

	// CreateCollection validation runs before field IDs are assigned, so the
	// helper must fall back to input field names when IDs are unavailable.
	schemaWithoutIds := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", DataType: schemapb.DataType_VarChar},
			{Name: "output_field", DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:             "bm25_func",
				Type:             schemapb.FunctionType_BM25,
				InputFieldNames:  []string{"input_field"},
				OutputFieldNames: []string{"output_field"},
			},
		},
	}
	assert.True(t, IsBm25FunctionInputField(schemaWithoutIds, schemaWithoutIds.Fields[0]))
	assert.False(t, IsBm25FunctionInputField(schemaWithoutIds, schemaWithoutIds.Fields[1]))

	// Test with multiple functions, only one is BM25
	multipleSchema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{FieldID: 100, Name: "input_field1", DataType: schemapb.DataType_VarChar},
			{FieldID: 101, Name: "input_field2", DataType: schemapb.DataType_VarChar},
			{FieldID: 102, Name: "output_field1", DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true},
			{FieldID: 103, Name: "output_field2", DataType: schemapb.DataType_FloatVector, IsFunctionOutput: true},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:             "bm25_func",
				Type:             schemapb.FunctionType_BM25,
				InputFieldIds:    []int64{100},
				InputFieldNames:  []string{"input_field1"},
				OutputFieldIds:   []int64{102},
				OutputFieldNames: []string{"output_field1"},
			},
			{
				Name:             "other_func",
				Type:             schemapb.FunctionType_Unknown,
				InputFieldIds:    []int64{101},
				InputFieldNames:  []string{"input_field2"},
				OutputFieldIds:   []int64{103},
				OutputFieldNames: []string{"output_field2"},
			},
		},
	}
	assert.True(t, IsBm25FunctionInputField(multipleSchema, multipleSchema.Fields[0]))
	assert.False(t, IsBm25FunctionInputField(multipleSchema, multipleSchema.Fields[1]))
	assert.False(t, IsBm25FunctionInputField(multipleSchema, multipleSchema.Fields[2]))
	assert.False(t, IsBm25FunctionInputField(multipleSchema, multipleSchema.Fields[3]))

	// Test with no BM25 functions
	noBM25Schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", DataType: schemapb.DataType_VarChar},
			{Name: "output_field", DataType: schemapb.DataType_FloatVector, IsFunctionOutput: true},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:             "other_func",
				Type:             schemapb.FunctionType_Unknown,
				InputFieldNames:  []string{"input_field"},
				OutputFieldNames: []string{"output_field"},
			},
		},
	}
	assert.False(t, IsBm25FunctionInputField(noBM25Schema, noBM25Schema.Fields[0]))

	// Test with empty functions
	emptySchema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", DataType: schemapb.DataType_VarChar},
		},
		Functions: []*schemapb.FunctionSchema{},
	}
	assert.False(t, IsBm25FunctionInputField(emptySchema, emptySchema.Fields[0]))

	// Test with nil functions
	nilSchema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", DataType: schemapb.DataType_VarChar},
		},
		Functions: nil,
	}
	assert.False(t, IsBm25FunctionInputField(nilSchema, nilSchema.Fields[0]))
}

func TestIsMinHashFunctionInputField(t *testing.T) {
	schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{FieldID: 100, Name: "text", DataType: schemapb.DataType_VarChar},
			{FieldID: 101, Name: "sig", DataType: schemapb.DataType_BinaryVector, IsFunctionOutput: true},
			{FieldID: 102, Name: "bm25_in", DataType: schemapb.DataType_VarChar},
			{FieldID: 103, Name: "sparse", DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name: "minhash_func", Type: schemapb.FunctionType_MinHash,
				InputFieldIds: []int64{100}, InputFieldNames: []string{"text"},
				OutputFieldIds: []int64{101}, OutputFieldNames: []string{"sig"},
			},
			{
				Name: "bm25_func", Type: schemapb.FunctionType_BM25,
				InputFieldIds: []int64{102}, InputFieldNames: []string{"bm25_in"},
				OutputFieldIds: []int64{103}, OutputFieldNames: []string{"sparse"},
			},
		},
	}
	// each helper matches only its own function type's input field
	assert.True(t, IsMinHashFunctionInputField(schema, schema.Fields[0])) // text -> minhash input
	assert.False(t, IsBm25FunctionInputField(schema, schema.Fields[0]))
	assert.True(t, IsBm25FunctionInputField(schema, schema.Fields[2])) // bm25_in -> bm25 input
	assert.False(t, IsMinHashFunctionInputField(schema, schema.Fields[2]))
	assert.False(t, IsMinHashFunctionInputField(schema, schema.Fields[1])) // output field

	// name fallback (no field ids assigned yet) + nil safety
	schemaNoIDs := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{{Name: "text", DataType: schemapb.DataType_VarChar}},
		Functions: []*schemapb.FunctionSchema{
			{Name: "mh", Type: schemapb.FunctionType_MinHash, InputFieldNames: []string{"text"}},
		},
	}
	assert.True(t, IsMinHashFunctionInputField(schemaNoIDs, schemaNoIDs.Fields[0]))
	assert.False(t, IsMinHashFunctionInputField(nil, schemaNoIDs.Fields[0]))
}

func TestIsMinHashFunctionOutputField(t *testing.T) {
	schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", DataType: schemapb.DataType_VarChar, TypeParams: []*commonpb.KeyValuePair{{Key: "enable_analyzer", Value: "true"}}},
			{Name: "output_field", DataType: schemapb.DataType_BinaryVector, IsFunctionOutput: true},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:             "minhash_func",
				Type:             schemapb.FunctionType_MinHash,
				InputFieldNames:  []string{"input_field"},
				OutputFieldNames: []string{"output_field"},
			},
		},
	}
	assert.False(t, IsMinHashFunctionOutputField(schema.Fields[0], schema))
	assert.True(t, IsMinHashFunctionOutputField(schema.Fields[1], schema))

	schemaWithIds := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", FieldID: 100, DataType: schemapb.DataType_VarChar, TypeParams: []*commonpb.KeyValuePair{{Key: "enable_analyzer", Value: "true"}}},
			{Name: "output_field", FieldID: 101, DataType: schemapb.DataType_BinaryVector, IsFunctionOutput: true},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:           "minhash_func",
				Type:           schemapb.FunctionType_MinHash,
				InputFieldIds:  []int64{100},
				OutputFieldIds: []int64{101},
			},
		},
	}
	assert.False(t, IsMinHashFunctionOutputField(schemaWithIds.Fields[0], schemaWithIds))
	assert.True(t, IsMinHashFunctionOutputField(schemaWithIds.Fields[1], schemaWithIds))

	nonMinHashSchema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", DataType: schemapb.DataType_VarChar},
			{Name: "output_field", DataType: schemapb.DataType_BinaryVector, IsFunctionOutput: true},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:             "other_func",
				Type:             schemapb.FunctionType_BM25,
				InputFieldNames:  []string{"input_field"},
				OutputFieldNames: []string{"output_field"},
			},
		},
	}
	assert.False(t, IsMinHashFunctionOutputField(nonMinHashSchema.Fields[1], nonMinHashSchema))

	nonBinarySchema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", DataType: schemapb.DataType_VarChar},
			{Name: "output_field", DataType: schemapb.DataType_FloatVector, IsFunctionOutput: true},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:             "minhash_func",
				Type:             schemapb.FunctionType_MinHash,
				InputFieldNames:  []string{"input_field"},
				OutputFieldNames: []string{"output_field"},
			},
		},
	}
	assert.False(t, IsMinHashFunctionOutputField(nonBinarySchema.Fields[1], nonBinarySchema))

	// Function output ownership depends on the normalized field marker.
	unmarkedOutputSchema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", DataType: schemapb.DataType_VarChar},
			{Name: "output_field", DataType: schemapb.DataType_BinaryVector, IsFunctionOutput: false},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:             "minhash_func",
				Type:             schemapb.FunctionType_MinHash,
				InputFieldNames:  []string{"input_field"},
				OutputFieldNames: []string{"output_field"},
			},
		},
	}
	assert.False(t, IsMinHashFunctionOutputField(unmarkedOutputSchema.Fields[1], unmarkedOutputSchema))
}

func TestSchemaHelper_GetFunctionByOutputField(t *testing.T) {
	schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", FieldID: 100, DataType: schemapb.DataType_VarChar},
			{Name: "output_field", FieldID: 101, DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true},
			{Name: "regular_field", FieldID: 102, DataType: schemapb.DataType_Int64},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:           "bm25_func",
				Type:           schemapb.FunctionType_BM25,
				InputFieldIds:  []int64{100},
				OutputFieldIds: []int64{101},
			},
		},
	}

	helper, err := CreateSchemaHelper(schema)
	assert.NoError(t, err)

	// Test getting function by output field
	function, err := helper.GetFunctionByOutputField(schema.Fields[1])
	assert.NoError(t, err)
	assert.Equal(t, "bm25_func", function.Name)
	assert.Equal(t, schemapb.FunctionType_BM25, function.Type)

	// Test with non-function output field
	_, err = helper.GetFunctionByOutputField(schema.Fields[0])
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "function not exist")

	// Test with regular field
	_, err = helper.GetFunctionByOutputField(schema.Fields[2])
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "function not exist")
}

func TestSchemaHelper_CanRetrieveRawFieldData(t *testing.T) {
	schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "input_field", FieldID: 100, DataType: schemapb.DataType_VarChar},
			{Name: "bm25_output", FieldID: 101, DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true},
			{Name: "other_output", FieldID: 102, DataType: schemapb.DataType_FloatVector, IsFunctionOutput: true},
			{Name: "regular_field", FieldID: 103, DataType: schemapb.DataType_Int64},
		},
		Functions: []*schemapb.FunctionSchema{
			{
				Name:           "bm25_func",
				Type:           schemapb.FunctionType_BM25,
				InputFieldIds:  []int64{100},
				OutputFieldIds: []int64{101},
			},
			{
				Name:           "other_func",
				Type:           schemapb.FunctionType_Unknown,
				InputFieldIds:  []int64{100},
				OutputFieldIds: []int64{102},
			},
		},
	}

	helper, err := CreateSchemaHelper(schema)
	assert.NoError(t, err)

	// Regular field should be retrievable
	assert.True(t, helper.CanRetrieveRawFieldData(schema.Fields[0]))
	assert.True(t, helper.CanRetrieveRawFieldData(schema.Fields[3]))

	// BM25 function output field should NOT be retrievable
	assert.False(t, helper.CanRetrieveRawFieldData(schema.Fields[1]))

	// Other function output field should be retrievable
	assert.True(t, helper.CanRetrieveRawFieldData(schema.Fields[2]))

	// Test with field that has IsFunctionOutput=true but no corresponding function
	orphanField := &schemapb.FieldSchema{
		Name:             "orphan_field",
		FieldID:          104,
		DataType:         schemapb.DataType_FloatVector,
		IsFunctionOutput: true,
	}
	assert.False(t, helper.CanRetrieveRawFieldData(orphanField))
}

func TestIsExternalCollection(t *testing.T) {
	// nil schema
	assert.False(t, IsExternalCollection(nil))

	// empty schema with no fields
	schema := &schemapb.CollectionSchema{}
	assert.False(t, IsExternalCollection(schema))

	// schema with fields but no ExternalField set
	schema.Fields = []*schemapb.FieldSchema{
		{Name: "field1", ExternalField: ""},
	}
	assert.False(t, IsExternalCollection(schema))

	// schema with ExternalSource but no ExternalField set
	schema.ExternalSource = "s3://bucket/path"
	assert.False(t, IsExternalCollection(schema))

	// schema with ExternalSpec but no ExternalField set
	schema.ExternalSource = ""
	schema.ExternalSpec = `{"format":"milvus-table"}`
	assert.False(t, IsExternalCollection(schema))

	// schema with ExternalSource/ExternalSpec but no ExternalField set
	schema.ExternalSource = "s3://bucket/path"
	assert.False(t, IsExternalCollection(schema))

	// schema with ExternalField set (empty ExternalSource is allowed)
	schema.ExternalSource = ""
	schema.ExternalSpec = ""
	schema.Fields = []*schemapb.FieldSchema{
		{Name: "field1", ExternalField: "ext_field1"},
	}
	assert.True(t, IsExternalCollection(schema))

	// schema with both ExternalSource and ExternalField set
	schema.ExternalSource = "s3://bucket/path"
	assert.True(t, IsExternalCollection(schema))
}

func TestNormalizeAndValidateExternalCollectionSchema(t *testing.T) {
	buildSchema := func() *schemapb.CollectionSchema {
		return &schemapb.CollectionSchema{
			Name:           "external",
			ExternalSource: "s3://bucket/path",
			ExternalSpec:   `{"format":"parquet"}`,
			Fields: []*schemapb.FieldSchema{
				{
					Name:          "text",
					DataType:      schemapb.DataType_VarChar,
					ExternalField: "text_col",
					TypeParams: []*commonpb.KeyValuePair{
						{Key: common.MaxLengthKey, Value: "32"},
					},
				},
				{
					Name:          "vec",
					DataType:      schemapb.DataType_FloatVector,
					ExternalField: "vec_col",
					TypeParams: []*commonpb.KeyValuePair{
						{Key: common.DimKey, Value: "16"},
					},
				},
			},
		}
	}

	t.Run("non external schema skipped", func(t *testing.T) {
		// A non-external schema has no ExternalField set on any field
		schema := &schemapb.CollectionSchema{
			Name: "regular_collection",
			Fields: []*schemapb.FieldSchema{
				{
					Name:     "text",
					DataType: schemapb.DataType_VarChar,
					TypeParams: []*commonpb.KeyValuePair{
						{Key: common.MaxLengthKey, Value: "32"},
					},
				},
			},
		}
		assert.NoError(t, NormalizeAndValidateExternalCollectionSchema(schema))
		assert.NoError(t, ValidateExternalCollectionResolvedSchema(schema))
	})

	t.Run("resolved schema skips system and unmapped fields", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields = append(schema.Fields,
			&schemapb.FieldSchema{
				Name:     common.RowIDFieldName,
				DataType: schemapb.DataType_Int64,
			},
			&schemapb.FieldSchema{
				Name:     "late_field",
				DataType: schemapb.DataType_Int64,
			},
		)
		assert.NoError(t, ValidateExternalCollectionResolvedSchema(schema))
	})

	t.Run("resolved schema skips generated output fields", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:             "sparse",
			FieldID:          102,
			DataType:         schemapb.DataType_SparseFloatVector,
			IsFunctionOutput: true,
		})
		assert.NoError(t, ValidateExternalCollectionResolvedSchema(schema))
	})

	t.Run("resolved schema rejects generated output external mapping", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:             "sparse",
			FieldID:          102,
			DataType:         schemapb.DataType_SparseFloatVector,
			IsFunctionOutput: true,
			ExternalField:    "sparse_col",
		})
		err := ValidateExternalCollectionResolvedSchema(schema)
		if assert.Error(t, err) {
			assert.Contains(t, err.Error(), "must not have external_field mapping")
		}
	})

	t.Run("functions allowed", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:             "sparse",
			DataType:         schemapb.DataType_SparseFloatVector,
			IsFunctionOutput: true,
		})
		schema.Functions = []*schemapb.FunctionSchema{{
			Name:             "bm25_fn",
			Type:             schemapb.FunctionType_BM25,
			InputFieldNames:  []string{"text"},
			OutputFieldNames: []string{"sparse"},
		}}
		assert.NoError(t, NormalizeAndValidateExternalCollectionSchema(schema))
	})

	t.Run("unmarked function output is validated as source field", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:     "sparse",
			DataType: schemapb.DataType_SparseFloatVector,
		})
		schema.Functions = []*schemapb.FunctionSchema{{
			Name:             "bm25_fn",
			Type:             schemapb.FunctionType_BM25,
			InputFieldNames:  []string{"text"},
			OutputFieldNames: []string{"sparse"},
		}}
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "must have external_field mapping")
	})

	t.Run("function output external_field rejected", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:             "sparse",
			DataType:         schemapb.DataType_SparseFloatVector,
			ExternalField:    "sparse_col",
			IsFunctionOutput: true,
		})
		schema.Functions = []*schemapb.FunctionSchema{{
			Name:             "bm25_fn",
			Type:             schemapb.FunctionType_BM25,
			InputFieldNames:  []string{"text"},
			OutputFieldNames: []string{"sparse"},
		}}
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "must not have external_field mapping")
	})

	t.Run("unmarked function output with external_field is validated as source field", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:          "sparse",
			DataType:      schemapb.DataType_SparseFloatVector,
			ExternalField: "sparse_col",
		})
		schema.Functions = []*schemapb.FunctionSchema{{
			Name:             "bm25_fn",
			Type:             schemapb.FunctionType_BM25,
			InputFieldNames:  []string{"text"},
			OutputFieldNames: []string{"sparse"},
		}}
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "does not support field type")
	})

	t.Run("dynamic field disabled", func(t *testing.T) {
		schema := buildSchema()
		schema.EnableDynamicField = true
		assert.Error(t, NormalizeAndValidateExternalCollectionSchema(schema))
	})

	t.Run("struct fields disabled", func(t *testing.T) {
		schema := buildSchema()
		schema.StructArrayFields = []*schemapb.StructArrayFieldSchema{
			{Name: "struct_field", Fields: []*schemapb.FieldSchema{{Name: "nested", DataType: schemapb.DataType_Array}}},
		}
		assert.Error(t, NormalizeAndValidateExternalCollectionSchema(schema))
	})

	t.Run("primary key disabled for non milvus table", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields[0].IsPrimaryKey = true
		assert.Error(t, NormalizeAndValidateExternalCollectionSchema(schema))
	})

	t.Run("real primary key allowed for milvus table", func(t *testing.T) {
		schema := buildSchema()
		schema.ExternalSpec = `{"format":"milvus-table"}`
		schema.Fields[0].IsPrimaryKey = true
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.NoError(t, err)
		assert.True(t, schema.Fields[0].GetIsPrimaryKey())
		assert.False(t, schema.Fields[0].GetNullable(), "real PK should remain non-nullable")
		assert.False(t, schema.Fields[1].GetNullable(), "vector field should remain non-nullable")
	})

	t.Run("milvus table requires external field mappings", func(t *testing.T) {
		schema := buildSchema()
		schema.ExternalSpec = `{"format":"milvus-table"}`
		schema.Fields[0].ExternalField = ""
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "must have external_field mapping")
	})

	t.Run("invalid external spec rejected", func(t *testing.T) {
		schema := buildSchema()
		schema.ExternalSpec = `{`
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "invalid external spec JSON")
	})

	t.Run("partition key disabled", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields[0].IsPartitionKey = true
		assert.Error(t, NormalizeAndValidateExternalCollectionSchema(schema))
	})

	t.Run("clustering key disabled", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields[0].IsClusteringKey = true
		assert.Error(t, NormalizeAndValidateExternalCollectionSchema(schema))
	})

	t.Run("auto id disabled", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields[0].AutoID = true
		assert.Error(t, NormalizeAndValidateExternalCollectionSchema(schema))
	})

	t.Run("text match allowed", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields[0].TypeParams = append(schema.Fields[0].TypeParams, &commonpb.KeyValuePair{
			Key:   "enable_match",
			Value: "true",
		}, &commonpb.KeyValuePair{
			Key:   "enable_analyzer",
			Value: "true",
		})
		assert.NoError(t, NormalizeAndValidateExternalCollectionSchema(schema))
	})

	t.Run("external_field mapping required", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields[0].ExternalField = ""
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "must have external_field mapping")
	})

	t.Run("valid schema passes", func(t *testing.T) {
		err := NormalizeAndValidateExternalCollectionSchema(buildSchema())
		assert.NoError(t, err)
	})

	t.Run("source set without spec rejected", func(t *testing.T) {
		schema := buildSchema()
		schema.ExternalSource = "s3://bucket/path"
		schema.ExternalSpec = ""
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "both set or both empty")
	})

	t.Run("spec set without source rejected", func(t *testing.T) {
		schema := buildSchema()
		schema.ExternalSource = ""
		schema.ExternalSpec = `{"format":"parquet"}`
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "both set or both empty")
	})

	t.Run("source/spec pair validated without external fields", func(t *testing.T) {
		schema := &schemapb.CollectionSchema{
			Name:           "regular",
			ExternalSource: "s3://bucket/path",
			Fields: []*schemapb.FieldSchema{
				{
					Name:     "text",
					DataType: schemapb.DataType_VarChar,
					TypeParams: []*commonpb.KeyValuePair{
						{Key: common.MaxLengthKey, Value: "32"},
					},
				},
			},
		}
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "both set or both empty")
	})

	t.Run("both empty allowed deferred refresh", func(t *testing.T) {
		schema := buildSchema()
		schema.ExternalSource = ""
		schema.ExternalSpec = ""
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.NoError(t, err)
	})

	t.Run("both set allowed", func(t *testing.T) {
		schema := buildSchema()
		schema.ExternalSource = "s3://bucket/path"
		schema.ExternalSpec = `{"format":"parquet"}`
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.NoError(t, err)
	})

	t.Run("duplicate external_field rejected different types", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:          "dup_int",
			DataType:      schemapb.DataType_Int64,
			ExternalField: "text_col",
		})
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "text_col")
		assert.Contains(t, err.Error(), "mapped by multiple fields")
	})

	t.Run("duplicate external_field rejected same type", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:          "text_alias",
			DataType:      schemapb.DataType_VarChar,
			ExternalField: "text_col",
			TypeParams: []*commonpb.KeyValuePair{
				{Key: common.MaxLengthKey, Value: "32"},
			},
		})
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "mapped by multiple fields")
	})

	t.Run("external_field matching function output name allowed", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields[1].ExternalField = "sparse"
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:             "sparse",
			DataType:         schemapb.DataType_SparseFloatVector,
			IsFunctionOutput: true,
		})
		schema.Functions = []*schemapb.FunctionSchema{{
			Name:             "bm25_fn",
			Type:             schemapb.FunctionType_BM25,
			InputFieldNames:  []string{"text"},
			OutputFieldNames: []string{"sparse"},
		}}
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.NoError(t, err)
	})

	t.Run("external_field matching function output field id rejected", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields[0].FieldID = 100
		schema.Fields[1].FieldID = 101
		schema.Fields[1].ExternalField = "102"
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:             "sparse",
			FieldID:          102,
			DataType:         schemapb.DataType_SparseFloatVector,
			IsFunctionOutput: true,
		})
		schema.Functions = []*schemapb.FunctionSchema{{
			Name:             "bm25_fn",
			Type:             schemapb.FunctionType_BM25,
			InputFieldNames:  []string{"text"},
			InputFieldIds:    []int64{100},
			OutputFieldNames: []string{"sparse"},
			OutputFieldIds:   []int64{102},
		}}
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "conflicts with generated function output field")
	})

	t.Run("function output ids define generated columns after rootcoord assigns ids", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields[0].FieldID = 100
		schema.Fields[1].FieldID = 101
		schema.Fields[1].ExternalField = "102"
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:             "sparse",
			FieldID:          102,
			DataType:         schemapb.DataType_SparseFloatVector,
			IsFunctionOutput: true,
		})
		schema.Functions = []*schemapb.FunctionSchema{{
			Name:             "bm25_fn",
			Type:             schemapb.FunctionType_BM25,
			InputFieldNames:  []string{"text"},
			InputFieldIds:    []int64{100},
			OutputFieldNames: []string{"sparse"},
			OutputFieldIds:   []int64{102},
		}}
		err := ValidateExternalCollectionResolvedSchema(schema)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "conflicts with generated function output field")
	})

	t.Run("duplicate external field mapping rejected", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:          "score",
			DataType:      schemapb.DataType_Double,
			ExternalField: "text_col",
		})
		err := ValidateExternalCollectionResolvedSchema(schema)
		if assert.Error(t, err) {
			assert.Contains(t, err.Error(), "external_field \"text_col\" is mapped by multiple fields")
		}
	})

	t.Run("resolved schema validation rejects unsupported user field type", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:          "bad_field",
			DataType:      schemapb.DataType_SparseFloatVector,
			ExternalField: "bad_col",
		})
		err := ValidateExternalCollectionResolvedSchema(schema)
		if assert.Error(t, err) {
			assert.Contains(t, err.Error(), "does not support field type")
		}
	})

	t.Run("unsupported field types rejected", func(t *testing.T) {
		unsupportedTypes := []schemapb.DataType{
			schemapb.DataType_SparseFloatVector,
		}
		for _, dt := range unsupportedTypes {
			schema := buildSchema()
			schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
				Name:          "bad_field",
				DataType:      dt,
				ExternalField: "bad_col",
			})
			err := NormalizeAndValidateExternalCollectionSchema(schema)
			assert.Error(t, err, "expected error for type %s", dt.String())
			assert.Contains(t, err.Error(), "does not support field type")
		}
	})

	t.Run("supported field types accepted", func(t *testing.T) {
		supportedTypes := []struct {
			dt     schemapb.DataType
			params []*commonpb.KeyValuePair
		}{
			{schemapb.DataType_Bool, nil},
			{schemapb.DataType_Int8, nil},
			{schemapb.DataType_Int16, nil},
			{schemapb.DataType_Int32, nil},
			{schemapb.DataType_Int64, nil},
			{schemapb.DataType_Float, nil},
			{schemapb.DataType_Double, nil},
			{schemapb.DataType_VarChar, []*commonpb.KeyValuePair{{Key: common.MaxLengthKey, Value: "64"}}},
			{schemapb.DataType_Text, []*commonpb.KeyValuePair{{Key: common.MaxLengthKey, Value: "64"}}},
			{schemapb.DataType_JSON, nil},
			{schemapb.DataType_Array, nil},
			{schemapb.DataType_Timestamptz, nil},
			{schemapb.DataType_Geometry, nil},
			{schemapb.DataType_FloatVector, []*commonpb.KeyValuePair{{Key: common.DimKey, Value: "8"}}},
			{schemapb.DataType_Float16Vector, []*commonpb.KeyValuePair{{Key: common.DimKey, Value: "8"}}},
			{schemapb.DataType_BFloat16Vector, []*commonpb.KeyValuePair{{Key: common.DimKey, Value: "8"}}},
			{schemapb.DataType_BinaryVector, []*commonpb.KeyValuePair{{Key: common.DimKey, Value: "8"}}},
			{schemapb.DataType_Int8Vector, []*commonpb.KeyValuePair{{Key: common.DimKey, Value: "8"}}},
			{schemapb.DataType_ArrayOfVector, []*commonpb.KeyValuePair{{Key: common.DimKey, Value: "8"}}},
		}
		for _, tc := range supportedTypes {
			schema := &schemapb.CollectionSchema{
				Name:           "external",
				ExternalSource: "s3://bucket/path",
				ExternalSpec:   `{"format":"parquet"}`,
				Fields: []*schemapb.FieldSchema{
					{
						Name:          "vec",
						DataType:      schemapb.DataType_FloatVector,
						ExternalField: "vec_col",
						TypeParams:    []*commonpb.KeyValuePair{{Key: common.DimKey, Value: "16"}},
					},
					{
						Name:          "test_field",
						DataType:      tc.dt,
						ExternalField: "test_col",
						TypeParams:    tc.params,
					},
				},
			}
			err := NormalizeAndValidateExternalCollectionSchema(schema)
			assert.NoError(t, err, "expected no error for type %s", tc.dt.String())
			assert.True(t, schema.GetFields()[0].GetNullable(), "base vector field should be nullable")
			assert.True(t, schema.GetFields()[1].GetNullable(), "field type %s should be nullable", tc.dt.String())
		}
	})

	t.Run("user fields forced nullable", func(t *testing.T) {
		schema := buildSchema()
		for _, f := range schema.GetFields() {
			assert.False(t, f.GetNullable())
		}
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.NoError(t, err)
		assert.True(t, schema.GetFields()[0].GetNullable(), "scalar field should be nullable")
		assert.True(t, schema.GetFields()[1].GetNullable(), "vector field should be nullable")
	})

	t.Run("virtual pk stays non-nullable", func(t *testing.T) {
		schema := buildSchema()
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:         common.VirtualPKFieldName,
			DataType:     schemapb.DataType_Int64,
			IsPrimaryKey: true,
			AutoID:       true,
			Nullable:     false,
		})
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.NoError(t, err)
		for _, f := range schema.GetFields() {
			switch f.GetName() {
			case common.VirtualPKFieldName:
				assert.False(t, f.GetNullable(), "virtual PK should remain non-nullable")
			case "vec":
				assert.True(t, f.GetNullable(), "vector field should be nullable")
			default:
				assert.True(t, f.GetNullable())
			}
		}
	})

	t.Run("already nullable fields unchanged", func(t *testing.T) {
		schema := buildSchema()
		for _, f := range schema.GetFields() {
			f.Nullable = true
		}
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.NoError(t, err)
		for _, f := range schema.GetFields() {
			assert.True(t, f.GetNullable())
		}
	})

	t.Run("validation failure leaves schema unmutated", func(t *testing.T) {
		// First field is valid, second field has an unsupported type.
		// Without the two-pass split, the first field would already have
		// Nullable=true by the time the second field's check fails — the
		// caller would receive an error but the schema would be partially
		// normalized, breaking the "atomic" invariant.
		schema := buildSchema()
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:          "bad_field",
			DataType:      schemapb.DataType_SparseFloatVector,
			ExternalField: "bad_col",
		})
		for _, f := range schema.GetFields() {
			f.Nullable = false
		}
		err := NormalizeAndValidateExternalCollectionSchema(schema)
		assert.Error(t, err)
		for _, f := range schema.GetFields() {
			assert.False(t, f.GetNullable(),
				"field %s nullable was flipped despite validation failure", f.GetName())
		}
	})
}

func TestValidateMilvusTableSchemaIdentity(t *testing.T) {
	source := &schemapb.CollectionSchema{
		Name: "source",
		Fields: []*schemapb.FieldSchema{
			{
				FieldID:      100,
				Name:         "pk",
				DataType:     schemapb.DataType_Int64,
				IsPrimaryKey: true,
			},
			{
				FieldID:  101,
				Name:     "vec",
				DataType: schemapb.DataType_FloatVector,
				TypeParams: []*commonpb.KeyValuePair{
					{Key: common.DimKey, Value: "16"},
				},
			},
		},
	}
	validTarget := func() *schemapb.CollectionSchema {
		target := proto.Clone(source).(*schemapb.CollectionSchema)
		target.Fields[0].ExternalField = "pk"
		target.Fields[1].ExternalField = "vec"
		return target
	}

	t.Run("create time ignores field id", func(t *testing.T) {
		target := &schemapb.CollectionSchema{
			Name: "target",
			Fields: []*schemapb.FieldSchema{
				{
					FieldID:       0,
					Name:          "target_pk",
					DataType:      schemapb.DataType_Int64,
					IsPrimaryKey:  true,
					ExternalField: "pk",
				},
				{
					FieldID:       0,
					Name:          "target_vec",
					DataType:      schemapb.DataType_FloatVector,
					ExternalField: "vec",
					TypeParams: []*commonpb.KeyValuePair{
						{Key: common.DimKey, Value: "16"},
					},
				},
			},
		}
		assert.NoError(t, ValidateMilvusTableSchemaIdentity(target, source, false))
	})

	t.Run("nil schemas rejected", func(t *testing.T) {
		err := ValidateMilvusTableSchemaIdentity(nil, source, false)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "target schema is nil")

		err = ValidateMilvusTableSchemaIdentity(validTarget(), nil, false)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "source snapshot schema is nil")
	})

	t.Run("dynamic field mismatch rejected", func(t *testing.T) {
		target := validTarget()
		target.EnableDynamicField = true
		err := ValidateMilvusTableSchemaIdentity(target, source, false)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "dynamic field setting mismatch")
	})

	t.Run("user field count mismatch rejected", func(t *testing.T) {
		target := validTarget()
		target.Fields = target.Fields[:1]
		err := ValidateMilvusTableSchemaIdentity(target, source, false)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "user field count mismatch")
	})

	t.Run("target field without mapping rejected", func(t *testing.T) {
		target := validTarget()
		target.Fields[1].ExternalField = ""
		err := ValidateMilvusTableSchemaIdentity(target, source, false)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "must set external_field mapping")
	})

	t.Run("duplicate source mapping rejected", func(t *testing.T) {
		target := validTarget()
		target.Fields[0] = proto.Clone(source.Fields[1]).(*schemapb.FieldSchema)
		target.Fields[0].Name = "vec_alias"
		target.Fields[0].ExternalField = "vec"
		target.Fields[1].ExternalField = "vec"
		err := ValidateMilvusTableSchemaIdentity(target, source, false)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "mapped by multiple target fields")
	})

	t.Run("unmapped source field rejected", func(t *testing.T) {
		target := validTarget()
		target.Fields[0].Name = "same_name"
		target.Fields[1].Name = "same_name"
		err := ValidateMilvusTableSchemaIdentity(target, source, false)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "is not mapped by target schema")
	})

	t.Run("source function output field is not required to be mapped", func(t *testing.T) {
		sourceWithFunctionOutput := proto.Clone(source).(*schemapb.CollectionSchema)
		sourceWithFunctionOutput.Fields = append(sourceWithFunctionOutput.Fields, &schemapb.FieldSchema{
			FieldID:          102,
			Name:             "sparse",
			DataType:         schemapb.DataType_SparseFloatVector,
			IsFunctionOutput: true,
		})
		sourceWithFunctionOutput.Functions = []*schemapb.FunctionSchema{
			{
				Name:             "bm25",
				InputFieldNames:  []string{"pk"},
				OutputFieldNames: []string{"sparse"},
				OutputFieldIds:   []int64{102},
			},
		}

		assert.NoError(t, ValidateMilvusTableSchemaIdentity(validTarget(), sourceWithFunctionOutput, false))
	})

	t.Run("ordinary target field can map source function output field", func(t *testing.T) {
		sourceWithFunctionOutput := proto.Clone(source).(*schemapb.CollectionSchema)
		sourceWithFunctionOutput.Fields = append(sourceWithFunctionOutput.Fields, &schemapb.FieldSchema{
			FieldID:          102,
			Name:             "sparse",
			DataType:         schemapb.DataType_SparseFloatVector,
			IsFunctionOutput: true,
		})
		sourceWithFunctionOutput.Functions = []*schemapb.FunctionSchema{
			{
				Name:             "bm25",
				InputFieldNames:  []string{"pk"},
				OutputFieldNames: []string{"sparse"},
				OutputFieldIds:   []int64{102},
			},
		}
		target := validTarget()
		target.Fields = append(target.Fields, &schemapb.FieldSchema{
			FieldID:       102,
			Name:          "target_sparse",
			DataType:      schemapb.DataType_SparseFloatVector,
			ExternalField: "sparse",
		})

		assert.NoError(t, ValidateMilvusTableSchemaIdentity(target, sourceWithFunctionOutput, false))
		assert.NoError(t, ValidateMilvusTableSchemaIdentity(target, sourceWithFunctionOutput, true))
	})

	t.Run("target function output field is not external mapping", func(t *testing.T) {
		target := validTarget()
		target.Fields = append(target.Fields, &schemapb.FieldSchema{
			FieldID:          102,
			Name:             "sparse",
			DataType:         schemapb.DataType_SparseFloatVector,
			IsFunctionOutput: true,
		})
		target.Functions = []*schemapb.FunctionSchema{
			{
				Name:             "bm25",
				InputFieldNames:  []string{"pk"},
				OutputFieldNames: []string{"sparse"},
				OutputFieldIds:   []int64{102},
			},
		}

		assert.NoError(t, ValidateMilvusTableSchemaIdentity(target, source, true))
	})

	t.Run("virtual pk maps source pk as data field", func(t *testing.T) {
		virtualSource := proto.Clone(source).(*schemapb.CollectionSchema)
		virtualSource.Fields[0].AutoID = true
		target := &schemapb.CollectionSchema{
			Name: "target",
			Fields: []*schemapb.FieldSchema{
				{
					FieldID:      0,
					Name:         common.VirtualPKFieldName,
					DataType:     schemapb.DataType_Int64,
					IsPrimaryKey: true,
					AutoID:       true,
				},
				{
					FieldID:       0,
					Name:          "source_pk",
					DataType:      schemapb.DataType_Int64,
					ExternalField: "pk",
				},
				{
					FieldID:       0,
					Name:          "target_vec",
					DataType:      schemapb.DataType_FloatVector,
					ExternalField: "vec",
					TypeParams: []*commonpb.KeyValuePair{
						{Key: common.DimKey, Value: "16"},
					},
				},
			},
		}
		assert.NoError(t, ValidateMilvusTableSchemaIdentity(target, virtualSource, false))

		target.Fields[1].FieldID = 100
		target.Fields[2].FieldID = 101
		assert.NoError(t, ValidateMilvusTableSchemaIdentity(target, virtualSource, true))
	})

	t.Run("source pk cannot map to non pk without virtual pk", func(t *testing.T) {
		target := validTarget()
		target.Fields[0].IsPrimaryKey = false
		err := ValidateMilvusTableSchemaIdentity(target, source, false)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "definition mismatch")
	})

	t.Run("refresh time requires field id", func(t *testing.T) {
		target := validTarget()
		target.Fields[1].FieldID = 102
		err := ValidateMilvusTableSchemaIdentity(target, source, true)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "field_id mismatch")
	})

	t.Run("type mismatch rejected", func(t *testing.T) {
		target := validTarget()
		target.Fields[1].TypeParams[0].Value = "32"
		err := ValidateMilvusTableSchemaIdentity(target, source, false)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "definition mismatch")
	})

	t.Run("missing mapped source field rejected", func(t *testing.T) {
		target := validTarget()
		target.Fields[1].ExternalField = "missing_vec"
		err := ValidateMilvusTableSchemaIdentity(target, source, false)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "source snapshot schema has no such field")
	})

	t.Run("type params are compared independent of order", func(t *testing.T) {
		sourceWithParams := proto.Clone(source).(*schemapb.CollectionSchema)
		sourceWithParams.Fields[1].TypeParams = []*commonpb.KeyValuePair{
			{Key: common.MaxLengthKey, Value: "16"},
			{Key: common.DimKey, Value: "16"},
			{Key: common.DimKey, Value: "16"},
		}
		target := proto.Clone(sourceWithParams).(*schemapb.CollectionSchema)
		target.Fields[0].ExternalField = "pk"
		target.Fields[1].ExternalField = "vec"
		target.Fields[1].TypeParams = []*commonpb.KeyValuePair{
			{Key: common.DimKey, Value: "16"},
			{Key: common.MaxLengthKey, Value: "16"},
			{Key: common.DimKey, Value: "16"},
		}
		assert.NoError(t, ValidateMilvusTableSchemaIdentity(target, sourceWithParams, true))
	})
}

func TestStorageColumnResolver(t *testing.T) {
	t.Run("external parquet source columns", func(t *testing.T) {
		schema := &schemapb.CollectionSchema{
			ExternalSource: "s3://bucket/data",
			ExternalSpec:   `{"format":"parquet"}`,
			Fields: []*schemapb.FieldSchema{
				{FieldID: 100, Name: "id", ExternalField: "id_col"},
				{FieldID: 101, Name: "raw"},
				{FieldID: 102, Name: "sparse", IsFunctionOutput: true},
				{FieldID: 103, Name: common.VirtualPKFieldName},
			},
		}
		resolver := NewStorageColumnResolver(schema)

		assert.False(t, resolver.IsMilvusTable())
		assert.Equal(t, []string{"id_col"}, resolver.SourceDataColumnNames())
		assert.True(t, resolver.IsSourceDataField(schema.Fields[0]))
		assert.False(t, resolver.IsSourceDataField(schema.Fields[1]))
		assert.False(t, resolver.IsSourceDataField(schema.Fields[2]))

		columnName, ok := resolver.ManifestStoredColumnName(schema.Fields[0])
		assert.True(t, ok)
		assert.Equal(t, "id_col", columnName)
		columnName, ok = resolver.ManifestStoredColumnName(schema.Fields[1])
		assert.True(t, ok)
		assert.Equal(t, "101", columnName)
		columnName, ok = resolver.ManifestStoredColumnName(schema.Fields[2])
		assert.True(t, ok)
		assert.Equal(t, "102", columnName)
	})

	t.Run("milvus table source columns", func(t *testing.T) {
		schema := &schemapb.CollectionSchema{
			ExternalSource: "s3://bucket/snapshot/metadata.json",
			ExternalSpec:   `{"format":"milvus-table"}`,
			Fields: []*schemapb.FieldSchema{
				{FieldID: 0, Name: common.VirtualPKFieldName},
				{FieldID: 100, Name: "target_pk", ExternalField: "pk"},
				{FieldID: 101, Name: "target_vec", ExternalField: "vec"},
				{FieldID: 102, Name: "sparse", IsFunctionOutput: true},
			},
			Functions: []*schemapb.FunctionSchema{
				{OutputFieldNames: []string{"sparse"}},
			},
		}
		resolver := NewStorageColumnResolver(schema)

		assert.True(t, resolver.IsMilvusTable())
		assert.Equal(t, []string{"100", "101"}, resolver.SourceDataColumnNames())
		assert.True(t, resolver.IsSourceDataField(schema.Fields[1]))
		assert.False(t, resolver.IsSourceDataField(schema.Fields[0]))
		assert.False(t, resolver.IsSourceDataField(schema.Fields[3]))

		columnName, ok := resolver.ManifestStoredColumnName(schema.Fields[0])
		assert.False(t, ok)
		assert.Empty(t, columnName)
		columnName, ok = resolver.ManifestStoredColumnName(schema.Fields[3])
		assert.True(t, ok)
		assert.Equal(t, "102", columnName)
	})

	t.Run("milvus table real primary key source columns include timestamp", func(t *testing.T) {
		schema := &schemapb.CollectionSchema{
			ExternalSource: "s3://bucket/snapshot/metadata.json",
			ExternalSpec:   `{"format":"milvus-table"}`,
			Fields: []*schemapb.FieldSchema{
				{FieldID: 100, Name: "pk", IsPrimaryKey: true},
				{FieldID: 101, Name: "vec"},
			},
		}
		resolver := NewStorageColumnResolver(schema)

		assert.Equal(t, []string{"100", "101", "1"}, resolver.SourceDataColumnNames())
	})

	t.Run("external spec override only enables milvus table rules", func(t *testing.T) {
		schema := &schemapb.CollectionSchema{
			ExternalSpec: `{"format":"parquet"}`,
			Fields: []*schemapb.FieldSchema{
				{FieldID: 100, Name: "id", ExternalField: "id_col"},
			},
		}
		resolver := NewStorageColumnResolver(schema, WithStorageColumnExternalSpec(`{"format":"milvus-table"}`))

		assert.True(t, resolver.IsMilvusTable())
		columnName, ok := resolver.SourceDataColumnName(schema.Fields[0])
		assert.True(t, ok)
		assert.Equal(t, "100", columnName)
	})

	t.Run("ordinary schemas keep legacy source column listing", func(t *testing.T) {
		schema := &schemapb.CollectionSchema{
			Fields: []*schemapb.FieldSchema{
				{FieldID: 100, Name: "id"},
				{FieldID: 101, Name: "vec"},
			},
		}
		resolver := NewStorageColumnResolver(schema)

		assert.Equal(t, []string{"id", "vec"}, resolver.SourceDataColumnNames())
		assert.False(t, resolver.IsSourceDataField(schema.Fields[0]))
	})
}

func TestAppendFieldDataByColumn(t *testing.T) {
	t.Run("nil dst or src", func(t *testing.T) {
		src := &schemapb.FieldData{
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{Data: []int32{1, 2, 3}},
					},
				},
			},
		}
		dst := &schemapb.FieldData{}
		// nil dst
		AppendFieldDataByColumn(nil, src, []int64{0, 1})
		// nil src
		AppendFieldDataByColumn(dst, nil, []int64{0, 1})
		// empty dataIndices
		AppendFieldDataByColumn(dst, src, []int64{})
		assert.Nil(t, dst.GetScalars())
	})

	t.Run("scalar int32", func(t *testing.T) {
		src := &schemapb.FieldData{
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{Data: []int32{10, 20, 30, 40, 50}},
					},
				},
			},
		}
		dst := &schemapb.FieldData{}
		AppendFieldDataByColumn(dst, src, []int64{1, 3})
		assert.Equal(t, []int32{20, 40}, dst.GetScalars().GetIntData().Data)
	})

	t.Run("scalar with ValidData", func(t *testing.T) {
		src := &schemapb.FieldData{
			ValidData: []bool{true, false, true, false, true},
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_LongData{
						LongData: &schemapb.LongArray{Data: []int64{100, 200, 300, 400, 500}},
					},
				},
			},
		}
		dst := &schemapb.FieldData{}
		AppendFieldDataByColumn(dst, src, []int64{0, 2, 4})
		assert.Equal(t, []int64{100, 300, 500}, dst.GetScalars().GetLongData().Data)
		assert.Equal(t, []bool{true, true, true}, dst.ValidData)
	})

	t.Run("string data", func(t *testing.T) {
		src := &schemapb.FieldData{
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_StringData{
						StringData: &schemapb.StringArray{Data: []string{"a", "b", "c", "d"}},
					},
				},
			},
		}
		dst := &schemapb.FieldData{}
		AppendFieldDataByColumn(dst, src, []int64{1, 2})
		assert.Equal(t, []string{"b", "c"}, dst.GetScalars().GetStringData().Data)
	})

	t.Run("json data", func(t *testing.T) {
		src := &schemapb.FieldData{
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_JsonData{
						JsonData: &schemapb.JSONArray{Data: [][]byte{[]byte(`{"a":1}`), []byte(`{"b":2}`)}},
					},
				},
			},
		}
		dst := &schemapb.FieldData{}
		AppendFieldDataByColumn(dst, src, []int64{0})
		assert.Equal(t, [][]byte{[]byte(`{"a":1}`)}, dst.GetScalars().GetJsonData().Data)
	})

	t.Run("float vector", func(t *testing.T) {
		dim := int64(4)
		src := &schemapb.FieldData{
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_FloatVector{
						FloatVector: &schemapb.FloatArray{Data: []float32{
							1, 2, 3, 4,
							5, 6, 7, 8,
							9, 10, 11, 12,
						}},
					},
				},
			},
		}
		dst := &schemapb.FieldData{}
		AppendFieldDataByColumn(dst, src, []int64{0, 2})
		assert.Equal(t, []float32{1, 2, 3, 4, 9, 10, 11, 12}, dst.GetVectors().GetFloatVector().Data)
		assert.Equal(t, dim, dst.GetVectors().GetDim())
	})

	t.Run("float vector with ValidData - all null case", func(t *testing.T) {
		dim := int64(4)
		src := &schemapb.FieldData{
			ValidData: []bool{false, false, false}, // all null
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_FloatVector{
						FloatVector: &schemapb.FloatArray{Data: []float32{}}, // COMPRESSED: no data
					},
				},
			},
		}
		dst := &schemapb.FieldData{}
		// dataIndices is empty (no non-null data), but rowIndices has all rows
		dataIndices := []int64{}
		rowIndices := []int64{0, 1, 2}
		AppendFieldDataByColumn(dst, src, dataIndices, rowIndices)
		// ValidData should be copied even though data is empty
		assert.Equal(t, []bool{false, false, false}, dst.ValidData)
		// No vector data copied
		assert.Nil(t, dst.GetVectors())
	})

	t.Run("sparse float vector", func(t *testing.T) {
		src := &schemapb.FieldData{
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: 100,
					Data: &schemapb.VectorField_SparseFloatVector{
						SparseFloatVector: &schemapb.SparseFloatArray{
							Dim:      100,
							Contents: [][]byte{{1, 2}, {3, 4}, {5, 6}},
						},
					},
				},
			},
		}
		dst := &schemapb.FieldData{}
		AppendFieldDataByColumn(dst, src, []int64{0, 2})
		assert.Equal(t, [][]byte{{1, 2}, {5, 6}}, dst.GetVectors().GetSparseFloatVector().Contents)
	})

	t.Run("nullable array of vector copies dense row data", func(t *testing.T) {
		validData := []bool{true, false, true, false}
		src := newArrayOfVectorFieldData(200, "arrvec", 1, schemapb.DataType_FloatVector, validData)
		dst := &schemapb.FieldData{Type: schemapb.DataType_ArrayOfVector}

		AppendFieldDataByColumn(dst, src, []int64{0, 1, 2, 3}, []int64{0, 1, 2, 3})

		got := dst.GetVectors().GetVectorArray()
		require.NotNil(t, got)
		require.Len(t, got.GetData(), 4)
		assert.Equal(t, validData, dst.GetValidData())
		assert.Equal(t, []float32{1}, got.GetData()[0].GetFloatVector().GetData())
		assert.Empty(t, got.GetData()[1].GetFloatVector().GetData())
		assert.Equal(t, []float32{3}, got.GetData()[2].GetFloatVector().GetData())
		assert.Empty(t, got.GetData()[3].GetFloatVector().GetData())
		assert.EqualValues(t, 1, got.GetDim())
		assert.Equal(t, schemapb.DataType_FloatVector, got.GetElementType())
	})

	t.Run("struct array appends dense nullable sub-field rows", func(t *testing.T) {
		src := &schemapb.FieldData{
			FieldName: "profile",
			FieldId:   200,
			Type:      schemapb.DataType_ArrayOfStruct,
			Field: &schemapb.FieldData_StructArrays{
				StructArrays: &schemapb.StructArrayField{
					Fields: []*schemapb.FieldData{
						{
							FieldName: "age",
							FieldId:   201,
							Type:      schemapb.DataType_Array,
							ValidData: []bool{true, false, true},
							Field: &schemapb.FieldData_Scalars{
								Scalars: &schemapb.ScalarField{
									Data: &schemapb.ScalarField_ArrayData{
										ArrayData: &schemapb.ArrayArray{
											ElementType: schemapb.DataType_Int32,
											Data: []*schemapb.ScalarField{
												{Data: &schemapb.ScalarField_IntData{IntData: &schemapb.IntArray{Data: []int32{10}}}},
												{Data: &schemapb.ScalarField_IntData{IntData: &schemapb.IntArray{}}},
												{Data: &schemapb.ScalarField_IntData{IntData: &schemapb.IntArray{Data: []int32{30}}}},
											},
										},
									},
								},
							},
						},
					},
				},
			},
		}
		dst := PrepareResultFieldData([]*schemapb.FieldData{src}, 3)[0]

		AppendFieldDataByColumn(dst, src, []int64{0, 1, 2})

		require.NotNil(t, dst.GetStructArrays())
		require.Len(t, dst.GetStructArrays().GetFields(), 1)
		subField := dst.GetStructArrays().GetFields()[0]
		assert.Equal(t, []bool{true, false, true}, subField.GetValidData())
		got := subField.GetScalars().GetArrayData().GetData()
		require.Len(t, got, 3)
		assert.Equal(t, []int32{10}, got[0].GetIntData().GetData())
		assert.Empty(t, got[1].GetIntData().GetData())
		assert.Equal(t, []int32{30}, got[2].GetIntData().GetData())
	})

	t.Run("struct array matches sub-fields by id", func(t *testing.T) {
		dst := &schemapb.FieldData{
			FieldName: "profile",
			FieldId:   200,
			Type:      schemapb.DataType_ArrayOfStruct,
			Field: &schemapb.FieldData_StructArrays{
				StructArrays: &schemapb.StructArrayField{
					Fields: []*schemapb.FieldData{
						{
							FieldName: "profile[age]",
							FieldId:   201,
							Type:      schemapb.DataType_Array,
							Field: &schemapb.FieldData_Scalars{Scalars: &schemapb.ScalarField{Data: &schemapb.ScalarField_ArrayData{ArrayData: &schemapb.ArrayArray{
								ElementType: schemapb.DataType_Int32,
								Data:        []*schemapb.ScalarField{},
							}}}},
						},
					},
				},
			},
		}
		src := &schemapb.FieldData{
			FieldName: "profile",
			FieldId:   200,
			Type:      schemapb.DataType_ArrayOfStruct,
			Field: &schemapb.FieldData_StructArrays{
				StructArrays: &schemapb.StructArrayField{
					Fields: []*schemapb.FieldData{
						{
							FieldName: "age",
							FieldId:   201,
							Type:      schemapb.DataType_Array,
							Field: &schemapb.FieldData_Scalars{Scalars: &schemapb.ScalarField{Data: &schemapb.ScalarField_ArrayData{ArrayData: &schemapb.ArrayArray{
								ElementType: schemapb.DataType_Int32,
								Data: []*schemapb.ScalarField{
									{Data: &schemapb.ScalarField_IntData{IntData: &schemapb.IntArray{Data: []int32{42}}}},
								},
							}}}},
						},
					},
				},
			},
		}

		AppendFieldDataByColumn(dst, src, []int64{0})

		require.NotNil(t, dst.GetStructArrays())
		require.Len(t, dst.GetStructArrays().GetFields(), 1)
		subField := dst.GetStructArrays().GetFields()[0]
		assert.Equal(t, "profile[age]", subField.GetFieldName())
		got := subField.GetScalars().GetArrayData().GetData()
		require.Len(t, got, 1)
		assert.Equal(t, []int32{42}, got[0].GetIntData().GetData())
	})
}

func TestUpdateFieldDataByColumn(t *testing.T) {
	t.Run("nil base or update", func(t *testing.T) {
		base := &schemapb.FieldData{
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{Data: []int32{1, 2, 3}},
					},
				},
			},
		}
		update := &schemapb.FieldData{
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{Data: []int32{10, 20}},
					},
				},
			},
		}
		// nil base
		err := UpdateFieldDataByColumn(nil, update, []int64{0}, []int64{0})
		assert.NoError(t, err)
		// nil update
		err = UpdateFieldDataByColumn(base, nil, []int64{0}, []int64{0})
		assert.NoError(t, err)
		// empty indices
		err = UpdateFieldDataByColumn(base, update, []int64{}, []int64{})
		assert.NoError(t, err)
		assert.Equal(t, []int32{1, 2, 3}, base.GetScalars().GetIntData().Data)
	})

	t.Run("indices length mismatch", func(t *testing.T) {
		base := &schemapb.FieldData{
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{Data: []int32{1, 2, 3}},
					},
				},
			},
		}
		update := &schemapb.FieldData{
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{Data: []int32{10, 20}},
					},
				},
			},
		}
		err := UpdateFieldDataByColumn(base, update, []int64{0, 1}, []int64{0})
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "length mismatch")
	})

	t.Run("scalar int32", func(t *testing.T) {
		base := &schemapb.FieldData{
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{Data: []int32{1, 2, 3, 4, 5}},
					},
				},
			},
		}
		update := &schemapb.FieldData{
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{Data: []int32{100, 200, 300}},
					},
				},
			},
		}
		err := UpdateFieldDataByColumn(base, update, []int64{1, 3}, []int64{0, 2})
		assert.NoError(t, err)
		assert.Equal(t, []int32{1, 100, 3, 300, 5}, base.GetScalars().GetIntData().Data)
	})

	t.Run("scalar with ValidData", func(t *testing.T) {
		base := &schemapb.FieldData{
			ValidData: []bool{true, true, true, true},
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_LongData{
						LongData: &schemapb.LongArray{Data: []int64{10, 20, 30, 40}},
					},
				},
			},
		}
		update := &schemapb.FieldData{
			ValidData: []bool{false, true},
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_LongData{
						LongData: &schemapb.LongArray{Data: []int64{100, 200}},
					},
				},
			},
		}
		err := UpdateFieldDataByColumn(base, update, []int64{1, 2}, []int64{0, 1})
		assert.NoError(t, err)
		assert.Equal(t, []int64{10, 100, 200, 40}, base.GetScalars().GetLongData().Data)
		assert.Equal(t, []bool{true, false, true, true}, base.ValidData)
	})

	t.Run("string data", func(t *testing.T) {
		base := &schemapb.FieldData{
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_StringData{
						StringData: &schemapb.StringArray{Data: []string{"a", "b", "c"}},
					},
				},
			},
		}
		update := &schemapb.FieldData{
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_StringData{
						StringData: &schemapb.StringArray{Data: []string{"X", "Y"}},
					},
				},
			},
		}
		err := UpdateFieldDataByColumn(base, update, []int64{0, 2}, []int64{0, 1})
		assert.NoError(t, err)
		assert.Equal(t, []string{"X", "b", "Y"}, base.GetScalars().GetStringData().Data)
	})

	t.Run("float vector", func(t *testing.T) {
		dim := int64(4)
		base := &schemapb.FieldData{
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_FloatVector{
						FloatVector: &schemapb.FloatArray{Data: []float32{
							1, 2, 3, 4,
							5, 6, 7, 8,
							9, 10, 11, 12,
						}},
					},
				},
			},
		}
		update := &schemapb.FieldData{
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_FloatVector{
						FloatVector: &schemapb.FloatArray{Data: []float32{
							100, 200, 300, 400,
							500, 600, 700, 800,
						}},
					},
				},
			},
		}
		err := UpdateFieldDataByColumn(base, update, []int64{0, 2}, []int64{0, 1})
		assert.NoError(t, err)
		expected := []float32{100, 200, 300, 400, 5, 6, 7, 8, 500, 600, 700, 800}
		assert.Equal(t, expected, base.GetVectors().GetFloatVector().Data)
	})

	t.Run("sparse float vector", func(t *testing.T) {
		base := &schemapb.FieldData{
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Data: &schemapb.VectorField_SparseFloatVector{
						SparseFloatVector: &schemapb.SparseFloatArray{
							Dim:      100,
							Contents: [][]byte{{1}, {2}, {3}},
						},
					},
				},
			},
		}
		update := &schemapb.FieldData{
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Data: &schemapb.VectorField_SparseFloatVector{
						SparseFloatVector: &schemapb.SparseFloatArray{
							Dim:      100,
							Contents: [][]byte{{10}, {20}},
						},
					},
				},
			},
		}
		err := UpdateFieldDataByColumn(base, update, []int64{0, 2}, []int64{0, 1})
		assert.NoError(t, err)
		assert.Equal(t, [][]byte{{10}, {2}, {20}}, base.GetVectors().GetSparseFloatVector().Contents)
	})

	t.Run("nullable array of vector stays row-dense", func(t *testing.T) {
		base := newArrayOfVectorFieldData(200, "arrvec", 1, schemapb.DataType_FloatVector, []bool{true, true, true})
		update := newArrayOfVectorFieldData(200, "arrvec", 1, schemapb.DataType_FloatVector, []bool{false, true})

		err := UpdateFieldDataByColumn(base, update, []int64{0, 2}, []int64{0, 1})

		require.NoError(t, err)
		assert.Equal(t, []bool{false, true, true}, base.GetValidData())

		got := base.GetVectors().GetVectorArray()
		require.NotNil(t, got)
		require.Len(t, got.GetData(), 3)
		assert.Empty(t, got.GetData()[0].GetFloatVector().GetData())
		assert.Equal(t, []float32{2}, got.GetData()[2].GetFloatVector().GetData())
		assert.EqualValues(t, 1, got.GetDim())
		assert.Equal(t, schemapb.DataType_FloatVector, got.GetElementType())
	})

	t.Run("nullable compact float vector changes validity", func(t *testing.T) {
		dim := int64(2)
		base := &schemapb.FieldData{
			Type:      schemapb.DataType_FloatVector,
			ValidData: []bool{true, false, true},
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_FloatVector{
						FloatVector: &schemapb.FloatArray{Data: []float32{
							1, 1,
							3, 3,
						}},
					},
				},
			},
		}
		update := &schemapb.FieldData{
			Type:      schemapb.DataType_FloatVector,
			ValidData: []bool{false, true},
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_FloatVector{
						FloatVector: &schemapb.FloatArray{Data: []float32{20, 20}},
					},
				},
			},
		}

		err := UpdateFieldDataByColumn(base, update, []int64{0, 1}, []int64{0, 1})

		require.NoError(t, err)
		assert.Equal(t, []bool{false, true, true}, base.GetValidData())
		assert.Equal(t, []float32{20, 20, 3, 3}, base.GetVectors().GetFloatVector().GetData())
	})

	t.Run("nullable compact sparse vector changes validity", func(t *testing.T) {
		base := &schemapb.FieldData{
			Type:      schemapb.DataType_SparseFloatVector,
			ValidData: []bool{true, false, true},
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Data: &schemapb.VectorField_SparseFloatVector{
						SparseFloatVector: &schemapb.SparseFloatArray{
							Dim:      100,
							Contents: [][]byte{CreateSparseFloatRow([]uint32{1}, []float32{1}), CreateSparseFloatRow([]uint32{3}, []float32{3})},
						},
					},
				},
			},
		}
		update := &schemapb.FieldData{
			Type:      schemapb.DataType_SparseFloatVector,
			ValidData: []bool{false, true},
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Data: &schemapb.VectorField_SparseFloatVector{
						SparseFloatVector: &schemapb.SparseFloatArray{
							Dim:      100,
							Contents: [][]byte{CreateSparseFloatRow([]uint32{20}, []float32{20})},
						},
					},
				},
			},
		}

		err := UpdateFieldDataByColumn(base, update, []int64{0, 1}, []int64{0, 1})

		require.NoError(t, err)
		assert.Equal(t, []bool{false, true, true}, base.GetValidData())
		assert.Equal(t, [][]byte{
			CreateSparseFloatRow([]uint32{20}, []float32{20}),
			CreateSparseFloatRow([]uint32{3}, []float32{3}),
		}, base.GetVectors().GetSparseFloatVector().GetContents())
	})

	t.Run("nullable compact sparse vector all null base accepts valid update", func(t *testing.T) {
		base := &schemapb.FieldData{
			Type:      schemapb.DataType_SparseFloatVector,
			ValidData: []bool{false, false},
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{},
			},
		}
		updateRow := CreateSparseFloatRow([]uint32{20}, []float32{20})
		update := &schemapb.FieldData{
			Type:      schemapb.DataType_SparseFloatVector,
			ValidData: []bool{true},
			Field: &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Data: &schemapb.VectorField_SparseFloatVector{
						SparseFloatVector: &schemapb.SparseFloatArray{
							Dim:      100,
							Contents: [][]byte{updateRow},
						},
					},
				},
			},
		}

		err := UpdateFieldDataByColumn(base, update, []int64{1}, []int64{0})

		require.NoError(t, err)
		assert.Equal(t, []bool{false, true}, base.GetValidData())
		assert.Equal(t, [][]byte{updateRow}, base.GetVectors().GetSparseFloatVector().GetContents())
	})
}

func TestIsClusteringKeyType(t *testing.T) {
	// Supported scalar types
	assert.True(t, IsClusteringKeyType(schemapb.DataType_Int8))
	assert.True(t, IsClusteringKeyType(schemapb.DataType_Int16))
	assert.True(t, IsClusteringKeyType(schemapb.DataType_Int32))
	assert.True(t, IsClusteringKeyType(schemapb.DataType_Int64))
	assert.True(t, IsClusteringKeyType(schemapb.DataType_Float))
	assert.True(t, IsClusteringKeyType(schemapb.DataType_Double))
	assert.True(t, IsClusteringKeyType(schemapb.DataType_VarChar))
	assert.True(t, IsClusteringKeyType(schemapb.DataType_String))

	// Supported vector types
	assert.True(t, IsClusteringKeyType(schemapb.DataType_FloatVector))

	// Unsupported types
	assert.False(t, IsClusteringKeyType(schemapb.DataType_JSON))
	assert.False(t, IsClusteringKeyType(schemapb.DataType_Bool))
	assert.False(t, IsClusteringKeyType(schemapb.DataType_Array))
	assert.False(t, IsClusteringKeyType(schemapb.DataType_Geometry))
	assert.False(t, IsClusteringKeyType(schemapb.DataType_Text))
	assert.False(t, IsClusteringKeyType(schemapb.DataType_Timestamptz))
	assert.False(t, IsClusteringKeyType(schemapb.DataType_BinaryVector))
	assert.False(t, IsClusteringKeyType(schemapb.DataType_Float16Vector))
	assert.False(t, IsClusteringKeyType(schemapb.DataType_BFloat16Vector))
}

func TestAppendFieldDataNullableVectorWithSchemaWithoutValidData(t *testing.T) {
	const fieldID = int64(101)
	schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{
				FieldID:  fieldID,
				Name:     "nullable_sparse",
				DataType: schemapb.DataType_SparseFloatVector,
				Nullable: true,
			},
		},
	}
	src := []*schemapb.FieldData{
		{
			Type:      schemapb.DataType_SparseFloatVector,
			FieldName: "nullable_sparse",
			FieldId:   fieldID,
			Field: &schemapb.FieldData_Vectors{Vectors: &schemapb.VectorField{
				Data: &schemapb.VectorField_SparseFloatVector{
					SparseFloatVector: &schemapb.SparseFloatArray{},
				},
			}},
		},
	}
	dst := PrepareResultFieldData(src, 1)
	idxComputer := NewFieldDataIdxComputerWithSchema(src, schema)
	fieldIdxs := idxComputer.Compute(0)

	require.NotPanics(t, func() {
		AppendFieldData(dst, src, 0, fieldIdxs...)
	})
	require.Len(t, dst, 1)
	assert.Equal(t, []bool{false}, dst[0].GetValidData())
	assert.Empty(t, dst[0].GetVectors().GetSparseFloatVector().GetContents())
}

func TestAppendFieldDataNullableVectorWithSchemaAndValidData(t *testing.T) {
	const (
		fieldID = int64(101)
		dim     = int64(2)
	)
	schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{
				FieldID:  fieldID,
				Name:     "nullable_float",
				DataType: schemapb.DataType_FloatVector,
				Nullable: true,
				TypeParams: []*commonpb.KeyValuePair{
					{Key: common.DimKey, Value: "2"},
				},
			},
		},
	}
	src := []*schemapb.FieldData{
		{
			Type:      schemapb.DataType_FloatVector,
			FieldName: "nullable_float",
			FieldId:   fieldID,
			ValidData: []bool{true, false, true},
			Field: &schemapb.FieldData_Vectors{Vectors: &schemapb.VectorField{
				Dim: dim,
				Data: &schemapb.VectorField_FloatVector{
					FloatVector: &schemapb.FloatArray{Data: []float32{1, 2, 3, 4}},
				},
			}},
		},
	}
	dst := PrepareResultFieldData(src, 3)
	idxComputer := NewFieldDataIdxComputerWithSchema(src, schema)
	for _, rowIdx := range []int64{0, 1, 2} {
		fieldIdxs := idxComputer.Compute(rowIdx)
		AppendFieldData(dst, src, rowIdx, fieldIdxs...)
	}

	assert.Equal(t, []bool{true, false, true}, dst[0].GetValidData())
	assert.Equal(t, []float32{1, 2, 3, 4}, dst[0].GetVectors().GetFloatVector().GetData())
}

func newArrayOfVectorFieldData(fieldID int64, fieldName string, dim int64, elementType schemapb.DataType, rows any) *schemapb.FieldData {
	var rowCount int
	var validData []bool
	switch v := rows.(type) {
	case int:
		rowCount = v
	case []bool:
		validData = v
		rowCount = len(v)
	default:
		panic("unexpected ArrayOfVector row spec")
	}

	data := make([]*schemapb.VectorField, rowCount)
	for i := 0; i < rowCount; i++ {
		if len(validData) > 0 && !validData[i] {
			data[i] = &schemapb.VectorField{
				Dim: dim,
				Data: &schemapb.VectorField_FloatVector{
					FloatVector: &schemapb.FloatArray{},
				},
			}
			continue
		}
		data[i] = &schemapb.VectorField{
			Dim: dim,
			Data: &schemapb.VectorField_FloatVector{
				FloatVector: &schemapb.FloatArray{Data: []float32{float32(i + 1)}},
			},
		}
	}
	fd := &schemapb.FieldData{
		Type:      schemapb.DataType_ArrayOfVector,
		FieldName: fieldName,
		FieldId:   fieldID,
		Field: &schemapb.FieldData_Vectors{
			Vectors: &schemapb.VectorField{
				Dim: dim,
				Data: &schemapb.VectorField_VectorArray{
					VectorArray: &schemapb.VectorArray{
						Data:        data,
						Dim:         dim,
						ElementType: elementType,
					},
				},
			},
		},
	}
	if len(validData) > 0 {
		fd.ValidData = validData
	}
	return fd
}

func TestFieldDataIdxComputer_ArrayOfVectorIsNonVector(t *testing.T) {
	scalar := &schemapb.FieldData{
		Type:    schemapb.DataType_Int64,
		FieldId: 100,
		Field: &schemapb.FieldData_Scalars{
			Scalars: &schemapb.ScalarField{
				Data: &schemapb.ScalarField_LongData{
					LongData: &schemapb.LongArray{Data: []int64{10, 20, 30, 40}},
				},
			},
		},
		ValidData: []bool{true, false, true, false},
	}

	// Compact vector: 2 valid rows out of 4; Data len==2.
	vec := &schemapb.FieldData{
		Type:    schemapb.DataType_FloatVector,
		FieldId: 101,
		Field: &schemapb.FieldData_Vectors{
			Vectors: &schemapb.VectorField{
				Dim: 1,
				Data: &schemapb.VectorField_FloatVector{
					FloatVector: &schemapb.FloatArray{Data: []float32{1.0, 2.0}},
				},
			},
		},
		ValidData: []bool{true, false, true, false},
	}

	arrVec := newArrayOfVectorFieldData(102, "arrvec", 1, schemapb.DataType_FloatVector, []bool{true, false, true, false})

	c := NewFieldDataIdxComputer([]*schemapb.FieldData{scalar, vec, arrVec})

	// The key invariant: for every row, ArrayOfVector's index must equal rowIdx
	// (dense/scalar semantics), never the compact index like a regular vector.
	// Regular vector uses compact indexing, which diverges from rowIdx at row 2:
	// rowIdx=2 but there's only one valid row before it, so compact index == 1.
	for rowIdx := int64(0); rowIdx < 4; rowIdx++ {
		got := c.Compute(rowIdx)
		assert.Equal(t, rowIdx, got[0], "scalar row %d", rowIdx)
		assert.Equal(t, rowIdx, got[2], "ArrayOfVector row %d must follow rowIdx", rowIdx)
	}

	// Reset by calling Compute on a smaller rowIdx.
	c.Compute(0)
	// At row 2 (the second valid row), compact vector index is 1, which differs
	// from rowIdx=2. This confirms ArrayOfVector is NOT using the compact path.
	got := c.Compute(2)
	assert.Equal(t, int64(1), got[1], "compact vector at row 2 -> compact index 1")
	assert.Equal(t, int64(2), got[2], "ArrayOfVector at row 2 -> rowIdx 2")
}

func TestNewEmptyArrayOfVectorRow(t *testing.T) {
	cases := []struct {
		name string
		elem schemapb.DataType
		data any
	}{
		{name: "FloatVector", elem: schemapb.DataType_FloatVector, data: &schemapb.VectorField_FloatVector{}},
		{name: "BinaryVector", elem: schemapb.DataType_BinaryVector, data: &schemapb.VectorField_BinaryVector{}},
		{name: "Float16Vector", elem: schemapb.DataType_Float16Vector, data: &schemapb.VectorField_Float16Vector{}},
		{name: "BFloat16Vector", elem: schemapb.DataType_BFloat16Vector, data: &schemapb.VectorField_Bfloat16Vector{}},
		{name: "Int8Vector", elem: schemapb.DataType_Int8Vector, data: &schemapb.VectorField_Int8Vector{}},
	}

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			vf, err := NewEmptyArrayOfVectorRow(8, c.elem)
			require.NoError(t, err)
			require.NotNil(t, vf)
			assert.EqualValues(t, 8, vf.GetDim())
			assert.IsType(t, c.data, vf.GetData())
		})
	}

	t.Run("unsupported element type returns error", func(t *testing.T) {
		vf, err := NewEmptyArrayOfVectorRow(8, schemapb.DataType_None)
		require.Error(t, err)
		assert.Nil(t, vf)
		assert.Contains(t, err.Error(), "unsupported ArrayOfVector element type")
	})
}

func TestAppendFieldData_ArrayOfVectorNullRowAppendsPlaceholder(t *testing.T) {
	validData := []bool{true, false, true, false}
	src := newArrayOfVectorFieldData(200, "arrvec", 1, schemapb.DataType_FloatVector, validData)

	// dst is a slice of same length as src, elements will be filled on first append.
	dst := make([]*schemapb.FieldData, 1)

	// Walk rows 0..3, fieldIdx == rowIdx because IdxComputer treats ArrayOfVector
	// as non-vector.
	for i := int64(0); i < 4; i++ {
		AppendFieldData(dst, []*schemapb.FieldData{src}, i, i)
	}

	require.NotNil(t, dst[0])
	got := dst[0].GetVectors().GetVectorArray()
	require.NotNil(t, got)
	// All 4 rows must be present — null rows are placeholders, not skipped.
	require.Len(t, got.GetData(), 4)
	assert.Equal(t, []float32{1}, got.GetData()[0].GetFloatVector().GetData())
	assert.Equal(t, []float32{3}, got.GetData()[2].GetFloatVector().GetData())
	assert.Empty(t, got.GetData()[1].GetFloatVector().GetData())
	assert.Empty(t, got.GetData()[3].GetFloatVector().GetData())
	assert.Equal(t, validData, dst[0].GetValidData())
	assert.EqualValues(t, 1, got.GetDim())
	assert.Equal(t, schemapb.DataType_FloatVector, got.GetElementType())
}

func TestGetTotalFieldsNumIncludesStructParent(t *testing.T) {
	schema := &schemapb.CollectionSchema{
		Fields: []*schemapb.FieldSchema{
			{Name: "pk", DataType: schemapb.DataType_Int64},
			{Name: "scalar", DataType: schemapb.DataType_Bool},
		},
		StructArrayFields: []*schemapb.StructArrayFieldSchema{
			{
				Name: "profile",
				Fields: []*schemapb.FieldSchema{
					{Name: "profile[ints]", DataType: schemapb.DataType_Array},
					{Name: "profile[vectors]", DataType: schemapb.DataType_ArrayOfVector},
				},
			},
		},
	}

	assert.Equal(t, 5, GetTotalFieldsNum(schema))
	assert.Len(t, GetAllFieldSchemas(schema), 4)
}
