milvus-io--milvus/internal/storage/utils.go

// 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 storage

import (
	"bytes"
	"context"
	"encoding/binary"
	"fmt"
	"io"
	"io/fs"
	"math"
	"os"
	"sort"
	"strconv"

	"github.com/apache/arrow/go/v17/arrow"
	"github.com/samber/lo"
	"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/mlog"
	"github.com/milvus-io/milvus/pkg/v3/mq/msgstream"
	"github.com/milvus-io/milvus/pkg/v3/objectstorage"
	"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
	"github.com/milvus-io/milvus/pkg/v3/proto/segcorepb"
	"github.com/milvus-io/milvus/pkg/v3/util/funcutil"
	"github.com/milvus-io/milvus/pkg/v3/util/merr"
	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
)

//////////////////////////////////////////////////////////////////////////////////////////////////

// Open opens file as os.Open works,
// also converts the os errors to Milvus errors
func Open(filepath string) (*os.File, error) {
	// NOLINT
	reader, err := os.Open(filepath)
	if os.IsNotExist(err) {
		return nil, merr.WrapErrIoKeyNotFound(filepath)
	} else if err != nil {
		return nil, merr.WrapErrIoFailed(filepath, err)
	}

	return reader, nil
}

// ReadFile reads file as os.ReadFile works,
// also converts the os errors to Milvus errors
func ReadFile(filepath string) ([]byte, error) {
	// NOLINT
	data, err := os.ReadFile(filepath)
	if os.IsNotExist(err) {
		return nil, merr.WrapErrIoKeyNotFound(filepath)
	} else if err != nil {
		return nil, merr.WrapErrIoFailed(filepath, err)
	}

	return data, nil
}

// WriteFile writes file as os.WriteFile works，
// also converts the os errors to Milvus errors
func WriteFile(filepath string, data []byte, perm fs.FileMode) error {
	// NOLINT
	err := os.WriteFile(filepath, data, perm)
	if err != nil {
		return merr.WrapErrIoFailed(filepath, err)
	}
	return nil
}

// ValidateStorageV1InsertWritableSchema validates schema constraints required by V1 insert binlogs.
func ValidateStorageV1InsertWritableSchema(schema *schemapb.CollectionSchema) error {
	for _, field := range schema.GetFields() {
		if isNullableArrayOfVectorField(field) {
			return merr.WrapErrParameterInvalidMsg("nullable ArrayOfVector is not supported in V1 storage format, fieldName=%s", field.GetName())
		}
	}

	for _, structField := range schema.GetStructArrayFields() {
		for _, field := range structField.GetFields() {
			if isNullableArrayOfVectorField(field) {
				return merr.WrapErrParameterInvalidMsg("nullable ArrayOfVector is not supported in V1 storage format, structName=%s, fieldName=%s",
					structField.GetName(), field.GetName())
			}
		}
	}

	return nil
}

func isNullableArrayOfVectorField(field *schemapb.FieldSchema) bool {
	return field.GetDataType() == schemapb.DataType_ArrayOfVector && field.GetNullable()
}

func checkTsField(data *InsertData) bool {
	tsData, ok := data.Data[common.TimeStampField]
	if !ok {
		return false
	}

	_, ok = tsData.(*Int64FieldData)
	return ok
}

func checkRowIDField(data *InsertData) bool {
	rowIDData, ok := data.Data[common.RowIDField]
	if !ok {
		return false
	}

	_, ok = rowIDData.(*Int64FieldData)
	return ok
}

func checkNumRows(fieldDatas ...FieldData) bool {
	if len(fieldDatas) <= 0 {
		return true
	}

	numRows := fieldDatas[0].RowNum()
	for i := 1; i < len(fieldDatas); i++ {
		if numRows != fieldDatas[i].RowNum() {
			return false
		}
	}

	return true
}

type fieldDataList struct {
	IDs   []FieldID
	datas []FieldData
}

func (ls fieldDataList) Len() int {
	return len(ls.IDs)
}

func (ls fieldDataList) Less(i, j int) bool {
	return ls.IDs[i] < ls.IDs[j]
}

func (ls fieldDataList) Swap(i, j int) {
	ls.IDs[i], ls.IDs[j] = ls.IDs[j], ls.IDs[i]
	ls.datas[i], ls.datas[j] = ls.datas[j], ls.datas[i]
}

func sortFieldDataList(ls fieldDataList) {
	sort.Sort(ls)
}

// TransferColumnBasedInsertDataToRowBased transfer column-based insert data to row-based rows.
// Note:
//   - ts column must exist in insert data;
//   - row id column must exist in insert data;
//   - the row num of all column must be equal;
//   - num_rows = len(RowData), a row will be assembled into the value of blob with field id order;
func TransferColumnBasedInsertDataToRowBased(data *InsertData) (
	Timestamps []uint64,
	RowIDs []int64,
	RowData []*commonpb.Blob,
	err error,
) {
	if !checkTsField(data) {
		return nil, nil, nil,
			merr.WrapErrServiceInternalMsg("cannot get timestamps from insert data")
	}

	if !checkRowIDField(data) {
		return nil, nil, nil,
			merr.WrapErrServiceInternalMsg("cannot get row ids from insert data")
	}

	tss := data.Data[common.TimeStampField].(*Int64FieldData)
	rowIDs := data.Data[common.RowIDField].(*Int64FieldData)

	ls := fieldDataList{}
	for fieldID := range data.Data {
		if fieldID == common.TimeStampField || fieldID == common.RowIDField {
			continue
		}

		ls.IDs = append(ls.IDs, fieldID)
		ls.datas = append(ls.datas, data.Data[fieldID])
	}

	// checkNumRows(tss, rowIDs, ls.datas...) // don't work
	all := []FieldData{tss, rowIDs}
	all = append(all, ls.datas...)
	if !checkNumRows(all...) {
		return nil, nil, nil,
			merr.WrapErrServiceInternalMsg("columns of insert data have different length")
	}

	sortFieldDataList(ls)

	numRows := tss.RowNum()
	rows := make([]*commonpb.Blob, numRows)
	for i := 0; i < numRows; i++ {
		blob := &commonpb.Blob{}
		var buffer bytes.Buffer

		for j := 0; j < ls.Len(); j++ {
			d := ls.datas[j].GetRow(i)
			err := binary.Write(&buffer, common.Endian, d)
			if err != nil {
				return nil, nil, nil,
					merr.WrapErrServiceInternalMsg("failed to get binary row, err: %v", err)
			}
		}

		blob.Value = buffer.Bytes()
		rows[i] = blob
	}

	utss := make([]uint64, tss.RowNum())
	for i := 0; i < tss.RowNum(); i++ {
		utss[i] = uint64(tss.Data[i])
	}

	return utss, rowIDs.Data, rows, nil
}

///////////////////////////////////////////////////////////////////////////////////////////

// TODO: remove these functions to proper file.

// GetDimFromParams get dim from params.
func GetDimFromParams(params []*commonpb.KeyValuePair) (int, error) {
	var dim int
	var err error
	for _, t := range params {
		if t.Key == common.DimKey {
			dim, err = strconv.Atoi(t.Value)
			if err != nil {
				return -1, err
			}
			return dim, nil
		}
	}
	return -1, merr.WrapErrServiceInternalMsg("dim not found in params")
}

// ReadBinary read data in bytes and write it into receiver.
//
//	The receiver can be any type in int8, int16, int32, int64, float32, float64 and bool
//	ReadBinary uses LittleEndian ByteOrder.
func ReadBinary(reader io.Reader, receiver interface{}, dataType schemapb.DataType) {
	err := binary.Read(reader, common.Endian, receiver)
	if err != nil {
		mlog.Error(context.TODO(), "binary.Read failed", mlog.Any("data type", dataType), mlog.Err(err))
	}
}

// It will save my life if golang support generic programming.

// TODO: string type.

func readFloatVectors(blobReaders []io.Reader, dim int) []float32 {
	ret := make([]float32, 0)
	for _, r := range blobReaders {
		v := make([]float32, dim)
		ReadBinary(r, &v, schemapb.DataType_FloatVector)
		ret = append(ret, v...)
	}
	return ret
}

func readBinaryVectors(blobReaders []io.Reader, dim int) []byte {
	ret := make([]byte, 0)
	for _, r := range blobReaders {
		v := make([]byte, dim/8)
		ReadBinary(r, &v, schemapb.DataType_BinaryVector)
		ret = append(ret, v...)
	}
	return ret
}

func readFloat16Vectors(blobReaders []io.Reader, dim int) []byte {
	ret := make([]byte, 0)
	for _, r := range blobReaders {
		v := make([]byte, dim*2)
		ReadBinary(r, &v, schemapb.DataType_Float16Vector)
		ret = append(ret, v...)
	}
	return ret
}

func readBFloat16Vectors(blobReaders []io.Reader, dim int) []byte {
	ret := make([]byte, 0)
	for _, r := range blobReaders {
		v := make([]byte, dim*2)
		ReadBinary(r, &v, schemapb.DataType_BFloat16Vector)
		ret = append(ret, v...)
	}
	return ret
}

func readInt8Vectors(blobReaders []io.Reader, dim int) []int8 {
	ret := make([]int8, 0)
	for _, r := range blobReaders {
		v := make([]int8, dim)
		ReadBinary(r, &v, schemapb.DataType_Int8Vector)
		ret = append(ret, v...)
	}
	return ret
}

func readBoolArray(blobReaders []io.Reader) []bool {
	ret := make([]bool, 0)
	for _, r := range blobReaders {
		var v bool
		ReadBinary(r, &v, schemapb.DataType_Bool)
		ret = append(ret, v)
	}
	return ret
}

func readInt8Array(blobReaders []io.Reader) []int8 {
	ret := make([]int8, 0)
	for _, r := range blobReaders {
		var v int8
		ReadBinary(r, &v, schemapb.DataType_Int8)
		ret = append(ret, v)
	}
	return ret
}

func readInt16Array(blobReaders []io.Reader) []int16 {
	ret := make([]int16, 0)
	for _, r := range blobReaders {
		var v int16
		ReadBinary(r, &v, schemapb.DataType_Int16)
		ret = append(ret, v)
	}
	return ret
}

func readInt32Array(blobReaders []io.Reader) []int32 {
	ret := make([]int32, 0)
	for _, r := range blobReaders {
		var v int32
		ReadBinary(r, &v, schemapb.DataType_Int32)
		ret = append(ret, v)
	}
	return ret
}

func readInt64Array(blobReaders []io.Reader) []int64 {
	ret := make([]int64, 0)
	for _, r := range blobReaders {
		var v int64
		ReadBinary(r, &v, schemapb.DataType_Int64)
		ret = append(ret, v)
	}
	return ret
}

func readFloatArray(blobReaders []io.Reader) []float32 {
	ret := make([]float32, 0)
	for _, r := range blobReaders {
		var v float32
		ReadBinary(r, &v, schemapb.DataType_Float)
		ret = append(ret, v)
	}
	return ret
}

func readDoubleArray(blobReaders []io.Reader) []float64 {
	ret := make([]float64, 0)
	for _, r := range blobReaders {
		var v float64
		ReadBinary(r, &v, schemapb.DataType_Double)
		ret = append(ret, v)
	}
	return ret
}

func readTimestamptzArray(blobReaders []io.Reader) []int64 {
	ret := make([]int64, 0)
	for _, r := range blobReaders {
		var v int64
		ReadBinary(r, &v, schemapb.DataType_Timestamptz) // DataType is only used for logging
		ret = append(ret, v)
	}
	return ret
}

func RowBasedInsertMsgToInsertData(msg *msgstream.InsertMsg, collSchema *schemapb.CollectionSchema, skipFunction bool) (idata *InsertData, err error) {
	blobReaders := make([]io.Reader, 0)
	for _, blob := range msg.RowData {
		blobReaders = append(blobReaders, bytes.NewReader(blob.GetValue()))
	}

	idata = &InsertData{
		Data: make(map[FieldID]FieldData),
		// TODO: handle Infos.
		Infos: nil,
	}

	if len(collSchema.StructArrayFields) > 0 {
		return nil, merr.WrapErrServiceInternalMsg("struct fields are not implemented in row based insert data")
	}

	for _, field := range collSchema.Fields {
		if skipFunction && isEmbeddingFunctionOutputField(field, collSchema) {
			continue
		}

		switch field.DataType {
		case schemapb.DataType_FloatVector:
			dim, err := GetDimFromParams(field.TypeParams)
			if err != nil {
				mlog.Error(context.TODO(), "failed to get dim", mlog.Err(err))
				return nil, err
			}

			vecs := readFloatVectors(blobReaders, dim)
			idata.Data[field.FieldID] = &FloatVectorFieldData{
				Data: vecs,
				Dim:  dim,
			}

		case schemapb.DataType_Float16Vector:
			dim, err := GetDimFromParams(field.TypeParams)
			if err != nil {
				mlog.Error(context.TODO(), "failed to get dim", mlog.Err(err))
				return nil, err
			}

			vecs := readFloat16Vectors(blobReaders, dim)
			idata.Data[field.FieldID] = &Float16VectorFieldData{
				Data: vecs,
				Dim:  dim,
			}

		case schemapb.DataType_BFloat16Vector:
			dim, err := GetDimFromParams(field.TypeParams)
			if err != nil {
				mlog.Error(context.TODO(), "failed to get dim", mlog.Err(err))
				return nil, err
			}

			vecs := readBFloat16Vectors(blobReaders, dim)
			idata.Data[field.FieldID] = &BFloat16VectorFieldData{
				Data: vecs,
				Dim:  dim,
			}

		case schemapb.DataType_BinaryVector:
			var dim int
			dim, err := GetDimFromParams(field.TypeParams)
			if err != nil {
				mlog.Error(context.TODO(), "failed to get dim", mlog.Err(err))
				return nil, err
			}

			vecs := readBinaryVectors(blobReaders, dim)
			idata.Data[field.FieldID] = &BinaryVectorFieldData{
				Data: vecs,
				Dim:  dim,
			}
		case schemapb.DataType_SparseFloatVector:
			return nil, merr.WrapErrServiceInternalMsg("Sparse Float Vector is not supported in row based data")

		case schemapb.DataType_Int8Vector:
			dim, err := GetDimFromParams(field.TypeParams)
			if err != nil {
				mlog.Error(context.TODO(), "failed to get dim", mlog.Err(err))
				return nil, err
			}

			vecs := readInt8Vectors(blobReaders, dim)
			idata.Data[field.FieldID] = &Int8VectorFieldData{
				Data: vecs,
				Dim:  dim,
			}

		case schemapb.DataType_Bool:
			idata.Data[field.FieldID] = &BoolFieldData{
				Data: readBoolArray(blobReaders),
			}

		case schemapb.DataType_Int8:
			idata.Data[field.FieldID] = &Int8FieldData{
				Data: readInt8Array(blobReaders),
			}

		case schemapb.DataType_Int16:
			idata.Data[field.FieldID] = &Int16FieldData{
				Data: readInt16Array(blobReaders),
			}

		case schemapb.DataType_Int32:
			idata.Data[field.FieldID] = &Int32FieldData{
				Data: readInt32Array(blobReaders),
			}

		case schemapb.DataType_Int64:
			idata.Data[field.FieldID] = &Int64FieldData{
				Data: nil,
			}

			fieldData := idata.Data[field.FieldID].(*Int64FieldData)
			switch field.FieldID {
			case 0: // rowIDs
				fieldData.Data = append(fieldData.Data, msg.RowIDs...)
			case 1: // Timestamps
				for _, ts := range msg.Timestamps {
					fieldData.Data = append(fieldData.Data, int64(ts))
				}
			default:
				fieldData.Data = readInt64Array(blobReaders)
			}

		case schemapb.DataType_Float:
			idata.Data[field.FieldID] = &FloatFieldData{
				Data: readFloatArray(blobReaders),
			}

		case schemapb.DataType_Double:
			idata.Data[field.FieldID] = &DoubleFieldData{
				Data: readDoubleArray(blobReaders),
			}
		case schemapb.DataType_Timestamptz:
			idata.Data[field.FieldID] = &TimestamptzFieldData{
				Data: readTimestamptzArray(blobReaders),
			}
		}
	}

	return idata, nil
}

func isEmbeddingFunctionOutputField(field *schemapb.FieldSchema, collSchema *schemapb.CollectionSchema) bool {
	return typeutil.IsBM25FunctionOutputField(field, collSchema) || typeutil.IsMinHashFunctionOutputField(field, collSchema)
}

// ColumnBasedInsertMsgToInsertData converts an InsertMsg msg into InsertData based
// on provided CollectionSchema collSchema.
//
// This function checks whether all fields are provided in the collSchema.Fields and not function output.
// If any field is missing in the msg, an error will be returned.
//
// This funcion also checks the length of each column. All columns shall have the same length.
// Also, the InsertData.Infos shall have BlobInfo with this length returned.
// When the length is not aligned, an error will be returned.
func validateColumnBasedNullableVectorFieldData(field *schemapb.FieldSchema, srcField *schemapb.FieldData, logicalRows int) error {
	if !field.GetNullable() || !typeutil.IsSupportedNullableVectorType(field.GetDataType()) {
		return nil
	}
	dim := int64(0)
	if field.GetDataType() != schemapb.DataType_SparseFloatVector && srcField.GetVectors() != nil && srcField.GetVectors().GetDim() == 0 {
		fieldDim, err := GetDimFromParams(field.GetTypeParams())
		if err != nil {
			return err
		}
		dim = int64(fieldDim)
	}
	requireValidData := logicalRows > 0 || len(srcField.GetValidData()) > 0
	if err := funcutil.ValidateNullableVectorFieldDataCompactWithDim(srcField, uint64(logicalRows), requireValidData, dim); err != nil {
		return merr.WrapErrParameterInvalidMsg(err.Error())
	}
	return nil
}

func validateColumnBasedInsertMsgNullableVectors(schema *schemapb.CollectionSchema, msg *msgstream.InsertMsg) error {
	srcFields := make(map[int64]*schemapb.FieldData, len(msg.GetFieldsData()))
	for _, fieldData := range msg.GetFieldsData() {
		srcFields[fieldData.GetFieldId()] = fieldData
	}

	for _, field := range typeutil.GetAllFieldSchemas(schema) {
		srcField, ok := srcFields[field.GetFieldID()]
		if !ok {
			continue
		}
		if err := validateColumnBasedNullableVectorFieldData(field, srcField, int(msg.GetNumRows())); err != nil {
			return err
		}
	}
	return nil
}

func ColumnBasedInsertMsgToInsertData(msg *msgstream.InsertMsg, collSchema *schemapb.CollectionSchema) (idata *InsertData, err error) {
	srcFields := make(map[FieldID]*schemapb.FieldData)
	for _, field := range msg.FieldsData {
		if _, ok := field.Field.(*schemapb.FieldData_StructArrays); ok {
			// unreachable
			panic("struct is not flattened")
		}
		srcFields[field.FieldId] = field
	}

	idata = &InsertData{
		Data: make(map[FieldID]FieldData),
	}
	length := 0
	hasMissingFields := false
	getFieldData := func(field *schemapb.FieldSchema) (FieldData, error) {
		srcField, ok := srcFields[field.GetFieldID()]
		if !ok && field.GetFieldID() >= common.StartOfUserFieldID {
			// Field not found in incoming message, will be handled by fillMissingFields later
			hasMissingFields = true
			return nil, nil
		}
		var fieldData FieldData
		switch field.DataType {
		case schemapb.DataType_FloatVector:
			dim, err := GetDimFromParams(field.TypeParams)
			if err != nil {
				mlog.Error(context.TODO(), "failed to get dim", mlog.Err(err))
				return nil, err
			}

			srcData := srcField.GetVectors().GetFloatVector().GetData()
			validData := srcField.GetValidData()
			if err := validateColumnBasedNullableVectorFieldData(field, srcField, int(msg.GetNumRows())); err != nil {
				return nil, err
			}
			fd := &FloatVectorFieldData{
				Data:      srcData,
				Dim:       dim,
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}
			if len(validData) > 0 {
				fd.L2PMapping.Build(validData, 0, len(validData))
			}
			fieldData = fd

		case schemapb.DataType_BinaryVector:
			dim, err := GetDimFromParams(field.TypeParams)
			if err != nil {
				mlog.Error(context.TODO(), "failed to get dim", mlog.Err(err))
				return nil, err
			}

			srcData := srcField.GetVectors().GetBinaryVector()
			validData := srcField.GetValidData()
			if err := validateColumnBasedNullableVectorFieldData(field, srcField, int(msg.GetNumRows())); err != nil {
				return nil, err
			}
			fd := &BinaryVectorFieldData{
				Data:      srcData,
				Dim:       dim,
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}
			if len(validData) > 0 {
				fd.L2PMapping.Build(validData, 0, len(validData))
			}
			fieldData = fd

		case schemapb.DataType_Float16Vector:
			dim, err := GetDimFromParams(field.TypeParams)
			if err != nil {
				mlog.Error(context.TODO(), "failed to get dim", mlog.Err(err))
				return nil, err
			}

			srcData := srcField.GetVectors().GetFloat16Vector()
			validData := srcField.GetValidData()
			if err := validateColumnBasedNullableVectorFieldData(field, srcField, int(msg.GetNumRows())); err != nil {
				return nil, err
			}
			fd := &Float16VectorFieldData{
				Data:      srcData,
				Dim:       dim,
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}
			if len(validData) > 0 {
				fd.L2PMapping.Build(validData, 0, len(validData))
			}
			fieldData = fd

		case schemapb.DataType_BFloat16Vector:
			dim, err := GetDimFromParams(field.TypeParams)
			if err != nil {
				mlog.Error(context.TODO(), "failed to get dim", mlog.Err(err))
				return nil, err
			}

			srcData := srcField.GetVectors().GetBfloat16Vector()
			validData := srcField.GetValidData()
			if err := validateColumnBasedNullableVectorFieldData(field, srcField, int(msg.GetNumRows())); err != nil {
				return nil, err
			}
			fd := &BFloat16VectorFieldData{
				Data:      srcData,
				Dim:       dim,
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}
			if len(validData) > 0 {
				fd.L2PMapping.Build(validData, 0, len(validData))
			}
			fieldData = fd

		case schemapb.DataType_SparseFloatVector:
			sparseArray := srcFields[field.FieldID].GetVectors().GetSparseFloatVector()
			validData := srcField.GetValidData()
			var contents [][]byte
			var dim int64
			if sparseArray != nil {
				contents = sparseArray.GetContents()
				dim = sparseArray.GetDim()
			}
			if err := validateColumnBasedNullableVectorFieldData(field, srcField, int(msg.GetNumRows())); err != nil {
				return nil, err
			}
			fd := &SparseFloatVectorFieldData{
				SparseFloatArray: schemapb.SparseFloatArray{
					Contents: contents,
					Dim:      dim,
				},
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}
			if len(validData) > 0 {
				fd.L2PMapping.Build(validData, 0, len(validData))
			}
			fieldData = fd

		case schemapb.DataType_Int8Vector:
			dim, err := GetDimFromParams(field.TypeParams)
			if err != nil {
				mlog.Error(context.TODO(), "failed to get dim", mlog.Err(err))
				return nil, err
			}

			srcData := srcField.GetVectors().GetInt8Vector()
			validData := srcField.GetValidData()
			if err := validateColumnBasedNullableVectorFieldData(field, srcField, int(msg.GetNumRows())); err != nil {
				return nil, err
			}
			fd := &Int8VectorFieldData{
				Data:      lo.Map(srcData, func(v byte, _ int) int8 { return int8(v) }),
				Dim:       dim,
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}
			if len(validData) > 0 {
				fd.L2PMapping.Build(validData, 0, len(validData))
			}
			fieldData = fd

		case schemapb.DataType_Bool:
			srcData := srcField.GetScalars().GetBoolData().GetData()
			validData := srcField.GetValidData()

			fieldData = &BoolFieldData{
				Data:      srcData,
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}

		case schemapb.DataType_Int8:
			srcData := srcField.GetScalars().GetIntData().GetData()
			validData := srcField.GetValidData()

			fieldData = &Int8FieldData{
				Data:      lo.Map(srcData, func(v int32, _ int) int8 { return int8(v) }),
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}

		case schemapb.DataType_Int16:
			srcData := srcField.GetScalars().GetIntData().GetData()
			validData := srcField.GetValidData()

			fieldData = &Int16FieldData{
				Data:      lo.Map(srcData, func(v int32, _ int) int16 { return int16(v) }),
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}

		case schemapb.DataType_Int32:
			srcData := srcField.GetScalars().GetIntData().GetData()
			validData := srcField.GetValidData()

			fieldData = &Int32FieldData{
				Data:      srcData,
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}

		case schemapb.DataType_Int64:
			switch field.FieldID {
			case common.RowIDField: // rowIDs
				fieldData = &Int64FieldData{
					Data: msg.GetRowIDs(),
				}
			case common.TimeStampField: // Timestamps
				fieldData = &Int64FieldData{
					Data: lo.Map(msg.GetTimestamps(), func(v uint64, _ int) int64 { return int64(v) }),
				}
			default:
				srcData := srcField.GetScalars().GetLongData().GetData()
				validData := srcField.GetValidData()
				fieldData = &Int64FieldData{
					Data:      srcData,
					ValidData: validData,
					Nullable:  field.GetNullable(),
				}
			}

		case schemapb.DataType_Float:
			srcData := srcField.GetScalars().GetFloatData().GetData()
			validData := srcField.GetValidData()

			fieldData = &FloatFieldData{
				Data:      srcData,
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}

		case schemapb.DataType_Double:
			srcData := srcField.GetScalars().GetDoubleData().GetData()
			validData := srcField.GetValidData()

			fieldData = &DoubleFieldData{
				Data:      srcData,
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}

		case schemapb.DataType_Timestamptz:
			srcData := srcField.GetScalars().GetTimestamptzData().GetData()
			validData := srcField.GetValidData()

			fieldData = &TimestamptzFieldData{
				Data:      srcData,
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}

		case schemapb.DataType_String, schemapb.DataType_VarChar, schemapb.DataType_Text:
			srcData := srcField.GetScalars().GetStringData().GetData()
			validData := srcField.GetValidData()

			fieldData = &StringFieldData{
				Data:      srcData,
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}

		case schemapb.DataType_Array:
			srcData := srcField.GetScalars().GetArrayData().GetData()
			validData := srcField.GetValidData()

			fieldData = &ArrayFieldData{
				ElementType: field.GetElementType(),
				Data:        srcData,
				ValidData:   validData,
				Nullable:    field.GetNullable(),
			}

		case schemapb.DataType_JSON:
			srcData := srcField.GetScalars().GetJsonData().GetData()
			validData := srcField.GetValidData()

			fieldData = &JSONFieldData{
				Data:      srcData,
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}

		case schemapb.DataType_ArrayOfVector:
			vectorArray := srcField.GetVectors().GetVectorArray()
			validData := srcField.GetValidData()

			fieldData = &VectorArrayFieldData{
				ElementType: field.GetElementType(),
				Data:        vectorArray.GetData(),
				Dim:         vectorArray.GetDim(),
				ValidData:   validData,
				Nullable:    field.GetNullable(),
			}
		case schemapb.DataType_Geometry:
			srcData := srcField.GetScalars().GetGeometryData().GetData()
			validData := srcField.GetValidData()
			fieldData = &GeometryFieldData{
				Data:      srcData,
				ValidData: validData,
				Nullable:  field.GetNullable(),
			}

		default:
			return nil, merr.WrapErrServiceInternal("data type not handled", field.GetDataType().String())
		}

		return fieldData, nil
	}

	handleFieldData := func(field *schemapb.FieldSchema) (FieldData, error) {
		if isEmbeddingFunctionOutputField(field, collSchema) {
			if _, ok := srcFields[field.GetFieldID()]; !ok {
				return nil, nil
			}
		}

		fieldData, err := getFieldData(field)
		if err != nil || fieldData == nil {
			return nil, err
		}

		if length == 0 {
			length = fieldData.RowNum()
		}

		if fieldData.RowNum() != length {
			return nil, merr.WrapErrServiceInternal("row num not match", fmt.Sprintf("field %s row num not match %d, other column %d", field.GetName(), fieldData.RowNum(), length))
		}

		return fieldData, nil
	}

	for _, field := range collSchema.Fields {
		fieldData, err := handleFieldData(field)
		if err != nil {
			return nil, err
		}
		if fieldData != nil {
			idata.Data[field.FieldID] = fieldData
		}
	}

	for _, structField := range collSchema.GetStructArrayFields() {
		for _, field := range structField.GetFields() {
			fieldData, err := handleFieldData(field)
			if err != nil {
				return nil, err
			}
			if fieldData != nil {
				idata.Data[field.FieldID] = fieldData
			}
		}
	}
	if hasMissingFields {
		// Fill missing fields after all fields are processed
		if err := fillMissingFields(collSchema, idata); err != nil {
			return nil, err
		}
	}

	idata.Infos = []BlobInfo{
		{Length: length},
	}
	return idata, nil
}

func InsertMsgToInsertData(msg *msgstream.InsertMsg, schema *schemapb.CollectionSchema) (idata *InsertData, err error) {
	if msg.IsRowBased() {
		return RowBasedInsertMsgToInsertData(msg, schema, true)
	}
	return ColumnBasedInsertMsgToInsertData(msg, schema)
}

func mergeBoolField(data *InsertData, fid FieldID, field *BoolFieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &BoolFieldData{
			Data:      nil,
			ValidData: nil,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*BoolFieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeInt8Field(data *InsertData, fid FieldID, field *Int8FieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &Int8FieldData{
			Data:      nil,
			ValidData: nil,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*Int8FieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeInt16Field(data *InsertData, fid FieldID, field *Int16FieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &Int16FieldData{
			Data:      nil,
			ValidData: nil,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*Int16FieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeInt32Field(data *InsertData, fid FieldID, field *Int32FieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &Int32FieldData{
			Data:      nil,
			ValidData: nil,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*Int32FieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeInt64Field(data *InsertData, fid FieldID, field *Int64FieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &Int64FieldData{
			Data:      nil,
			ValidData: nil,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*Int64FieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeFloatField(data *InsertData, fid FieldID, field *FloatFieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &FloatFieldData{
			Data:      nil,
			ValidData: nil,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*FloatFieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeDoubleField(data *InsertData, fid FieldID, field *DoubleFieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &DoubleFieldData{
			Data:      nil,
			ValidData: nil,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*DoubleFieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeTimestamptzField(data *InsertData, fid FieldID, field *TimestamptzFieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &TimestamptzFieldData{
			Data:      nil,
			ValidData: nil,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*TimestamptzFieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeStringField(data *InsertData, fid FieldID, field *StringFieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &StringFieldData{
			Data:      nil,
			ValidData: nil,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*StringFieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeArrayField(data *InsertData, fid FieldID, field *ArrayFieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &ArrayFieldData{
			ElementType: field.ElementType,
			Data:        nil,
			ValidData:   nil,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*ArrayFieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeJSONField(data *InsertData, fid FieldID, field *JSONFieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &JSONFieldData{
			Data:      nil,
			ValidData: nil,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*JSONFieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeBinaryVectorField(data *InsertData, fid FieldID, field *BinaryVectorFieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &BinaryVectorFieldData{
			Data:      nil,
			Dim:       field.Dim,
			ValidData: nil,
			Nullable:  field.Nullable,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*BinaryVectorFieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	if len(field.ValidData) > 0 {
		fieldData.L2PMapping.Build(field.ValidData, len(fieldData.ValidData), len(field.ValidData))
	}
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeFloatVectorField(data *InsertData, fid FieldID, field *FloatVectorFieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &FloatVectorFieldData{
			Data:      nil,
			Dim:       field.Dim,
			ValidData: nil,
			Nullable:  field.Nullable,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*FloatVectorFieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	if len(field.ValidData) > 0 {
		fieldData.L2PMapping.Build(field.ValidData, len(fieldData.ValidData), len(field.ValidData))
	}
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeFloat16VectorField(data *InsertData, fid FieldID, field *Float16VectorFieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &Float16VectorFieldData{
			Data:      nil,
			Dim:       field.Dim,
			ValidData: nil,
			Nullable:  field.Nullable,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*Float16VectorFieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	if len(field.ValidData) > 0 {
		fieldData.L2PMapping.Build(field.ValidData, len(fieldData.ValidData), len(field.ValidData))
	}
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeBFloat16VectorField(data *InsertData, fid FieldID, field *BFloat16VectorFieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &BFloat16VectorFieldData{
			Data:      nil,
			Dim:       field.Dim,
			ValidData: nil,
			Nullable:  field.Nullable,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*BFloat16VectorFieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	if len(field.ValidData) > 0 {
		fieldData.L2PMapping.Build(field.ValidData, len(fieldData.ValidData), len(field.ValidData))
	}
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeSparseFloatVectorField(data *InsertData, fid FieldID, field *SparseFloatVectorFieldData) {
	if _, ok := data.Data[fid]; !ok {
		data.Data[fid] = &SparseFloatVectorFieldData{
			Nullable: field.Nullable,
		}
	}
	fieldData := data.Data[fid].(*SparseFloatVectorFieldData)
	fieldData.AppendAllRows(field)
}

func mergeVectorArrayField(data *InsertData, fid FieldID, field *VectorArrayFieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &VectorArrayFieldData{
			Data:        nil,
			Dim:         field.Dim,
			ElementType: field.ElementType,
			ValidData:   nil,
			Nullable:    field.Nullable,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*VectorArrayFieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

func mergeInt8VectorField(data *InsertData, fid FieldID, field *Int8VectorFieldData) {
	if _, ok := data.Data[fid]; !ok {
		fieldData := &Int8VectorFieldData{
			Data:      nil,
			Dim:       field.Dim,
			ValidData: nil,
			Nullable:  field.Nullable,
		}
		data.Data[fid] = fieldData
	}
	fieldData := data.Data[fid].(*Int8VectorFieldData)
	fieldData.Data = append(fieldData.Data, field.Data...)
	if len(field.ValidData) > 0 {
		fieldData.L2PMapping.Build(field.ValidData, len(fieldData.ValidData), len(field.ValidData))
	}
	fieldData.ValidData = append(fieldData.ValidData, field.ValidData...)
}

// MergeFieldData merge field into data.
func MergeFieldData(data *InsertData, fid FieldID, field FieldData) {
	if field == nil {
		return
	}
	switch field := field.(type) {
	case *BoolFieldData:
		mergeBoolField(data, fid, field)
	case *Int8FieldData:
		mergeInt8Field(data, fid, field)
	case *Int16FieldData:
		mergeInt16Field(data, fid, field)
	case *Int32FieldData:
		mergeInt32Field(data, fid, field)
	case *Int64FieldData:
		mergeInt64Field(data, fid, field)
	case *FloatFieldData:
		mergeFloatField(data, fid, field)
	case *DoubleFieldData:
		mergeDoubleField(data, fid, field)
	case *TimestamptzFieldData:
		mergeTimestamptzField(data, fid, field)
	case *StringFieldData:
		mergeStringField(data, fid, field)
	case *ArrayFieldData:
		mergeArrayField(data, fid, field)
	case *JSONFieldData:
		mergeJSONField(data, fid, field)
	case *BinaryVectorFieldData:
		mergeBinaryVectorField(data, fid, field)
	case *FloatVectorFieldData:
		mergeFloatVectorField(data, fid, field)
	case *Float16VectorFieldData:
		mergeFloat16VectorField(data, fid, field)
	case *BFloat16VectorFieldData:
		mergeBFloat16VectorField(data, fid, field)
	case *SparseFloatVectorFieldData:
		mergeSparseFloatVectorField(data, fid, field)
	case *Int8VectorFieldData:
		mergeInt8VectorField(data, fid, field)
	case *VectorArrayFieldData:
		mergeVectorArrayField(data, fid, field)
	}
}

// MergeInsertData append the insert datas to the original buffer.
func MergeInsertData(buffer *InsertData, datas ...*InsertData) {
	if buffer == nil {
		mlog.Warn(context.TODO(), "Attempt to merge data into a nil buffer, skip the data merge.")
		return
	}

	for _, data := range datas {
		if data != nil {
			for fid, field := range data.Data {
				MergeFieldData(buffer, fid, field)
			}

			// TODO: handle storage.InsertData.Infos
			buffer.Infos = append(buffer.Infos, data.Infos...)
		}
	}
}

// TODO: string type.
func GetPkFromInsertData(collSchema *schemapb.CollectionSchema, data *InsertData) (FieldData, error) {
	helper, err := typeutil.CreateSchemaHelper(collSchema)
	if err != nil {
		mlog.Error(context.TODO(), "failed to create schema helper", mlog.Err(err))
		return nil, err
	}

	pf, err := helper.GetPrimaryKeyField()
	if err != nil {
		mlog.Warn(context.TODO(), "no primary field found", mlog.Err(err))
		return nil, err
	}

	pfData, ok := data.Data[pf.FieldID]
	if !ok {
		mlog.Warn(context.TODO(), "no primary field found in insert msg", mlog.Int64("fieldID", pf.FieldID))
		return nil, merr.WrapErrServiceInternalMsg("no primary field found in insert msg")
	}

	var realPfData FieldData
	switch pf.DataType {
	case schemapb.DataType_Int64:
		realPfData, ok = pfData.(*Int64FieldData)
	case schemapb.DataType_VarChar:
		realPfData, ok = pfData.(*StringFieldData)
	default:
		// TODO
	}
	if !ok {
		mlog.Warn(context.TODO(), "primary field not in Int64 or VarChar format", mlog.Int64("fieldID", pf.FieldID))
		return nil, merr.WrapErrServiceInternalMsg("primary field not in Int64 or VarChar format")
	}

	return realPfData, nil
}

// GetTimestampFromInsertData returns the Int64FieldData for timestamp field.
func GetTimestampFromInsertData(data *InsertData) (*Int64FieldData, error) {
	if data == nil {
		return nil, merr.WrapErrServiceInternalMsg("try to get timestamp from nil insert data")
	}
	fieldData, ok := data.Data[common.TimeStampField]
	if !ok {
		return nil, merr.WrapErrServiceInternalMsg("no timestamp field in insert data")
	}

	ifd, ok := fieldData.(*Int64FieldData)
	if !ok {
		return nil, merr.WrapErrServiceInternalMsg("timestamp field is not Int64")
	}

	return ifd, nil
}

func boolFieldDataToPbBytes(field *BoolFieldData) ([]byte, error) {
	arr := &schemapb.BoolArray{Data: field.Data}
	return proto.Marshal(arr)
}

func stringFieldDataToPbBytes(field *StringFieldData) ([]byte, error) {
	arr := &schemapb.StringArray{Data: field.Data}
	return proto.Marshal(arr)
}

func arrayFieldDataToPbBytes(field *ArrayFieldData) ([]byte, error) {
	arr := &schemapb.ArrayArray{Data: field.Data}
	return proto.Marshal(arr)
}

func jsonFieldDataToPbBytes(field *JSONFieldData) ([]byte, error) {
	arr := &schemapb.JSONArray{Data: field.Data}
	return proto.Marshal(arr)
}

func binaryWrite(endian binary.ByteOrder, data interface{}) ([]byte, error) {
	buf := new(bytes.Buffer)
	err := binary.Write(buf, endian, data)
	if err != nil {
		return nil, err
	}
	return buf.Bytes(), nil
}

func TransferInsertDataToInsertRecord(insertData *InsertData) (*segcorepb.InsertRecord, error) {
	insertRecord := &segcorepb.InsertRecord{}
	for fieldID, rawData := range insertData.Data {
		var fieldData *schemapb.FieldData
		switch rawData := rawData.(type) {
		case *BoolFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_Bool,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_BoolData{
							BoolData: &schemapb.BoolArray{
								Data: rawData.Data,
							},
						},
					},
				},
				ValidData: rawData.ValidData,
			}
		case *Int8FieldData:
			int32Data := make([]int32, len(rawData.Data))
			for index, v := range rawData.Data {
				int32Data[index] = int32(v)
			}
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_Int8,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_IntData{
							IntData: &schemapb.IntArray{
								Data: int32Data,
							},
						},
					},
				},
				ValidData: rawData.ValidData,
			}
		case *Int16FieldData:
			int32Data := make([]int32, len(rawData.Data))
			for index, v := range rawData.Data {
				int32Data[index] = int32(v)
			}
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_Int16,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_IntData{
							IntData: &schemapb.IntArray{
								Data: int32Data,
							},
						},
					},
				},
				ValidData: rawData.ValidData,
			}
		case *Int32FieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_Int32,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_IntData{
							IntData: &schemapb.IntArray{
								Data: rawData.Data,
							},
						},
					},
				},
				ValidData: rawData.ValidData,
			}
		case *Int64FieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_Int64,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_LongData{
							LongData: &schemapb.LongArray{
								Data: rawData.Data,
							},
						},
					},
				},
				ValidData: rawData.ValidData,
			}
		case *FloatFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_Float,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_FloatData{
							FloatData: &schemapb.FloatArray{
								Data: rawData.Data,
							},
						},
					},
				},
				ValidData: rawData.ValidData,
			}
		case *DoubleFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_Double,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_DoubleData{
							DoubleData: &schemapb.DoubleArray{
								Data: rawData.Data,
							},
						},
					},
				},
				ValidData: rawData.ValidData,
			}
		case *TimestamptzFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_Timestamptz,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_TimestamptzData{
							TimestamptzData: &schemapb.TimestamptzArray{
								Data: rawData.Data,
							},
						},
					},
				},
				ValidData: rawData.ValidData,
			}
		case *StringFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_VarChar,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_StringData{
							StringData: &schemapb.StringArray{
								Data: rawData.Data,
							},
						},
					},
				},
				ValidData: rawData.ValidData,
			}
		case *ArrayFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_Array,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_ArrayData{
							ArrayData: &schemapb.ArrayArray{
								Data: rawData.Data,
							},
						},
					},
				},
				ValidData: rawData.ValidData,
			}
		case *JSONFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_JSON,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_JsonData{
							JsonData: &schemapb.JSONArray{
								Data: rawData.Data,
							},
						},
					},
				},
				ValidData: rawData.ValidData,
			}
		case *GeometryFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_Geometry,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_GeometryData{
							GeometryData: &schemapb.GeometryArray{
								Data: rawData.Data,
							},
						},
					},
				},
				ValidData: rawData.ValidData,
			}
		case *FloatVectorFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_FloatVector,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Data: &schemapb.VectorField_FloatVector{
							FloatVector: &schemapb.FloatArray{
								Data: rawData.Data,
							},
						},
						Dim: int64(rawData.Dim),
					},
				},
				ValidData: rawData.ValidData,
			}
		case *BinaryVectorFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_BinaryVector,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Data: &schemapb.VectorField_BinaryVector{
							BinaryVector: rawData.Data,
						},
						Dim: int64(rawData.Dim),
					},
				},
				ValidData: rawData.ValidData,
			}
		case *Float16VectorFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_Float16Vector,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Data: &schemapb.VectorField_Float16Vector{
							Float16Vector: rawData.Data,
						},
						Dim: int64(rawData.Dim),
					},
				},
				ValidData: rawData.ValidData,
			}
		case *BFloat16VectorFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_BFloat16Vector,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Data: &schemapb.VectorField_Bfloat16Vector{
							Bfloat16Vector: rawData.Data,
						},
						Dim: int64(rawData.Dim),
					},
				},
				ValidData: rawData.ValidData,
			}
		case *SparseFloatVectorFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_SparseFloatVector,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Data: &schemapb.VectorField_SparseFloatVector{
							SparseFloatVector: &rawData.SparseFloatArray,
						},
					},
				},
				ValidData: rawData.ValidData,
			}
		case *Int8VectorFieldData:
			dataBytes := arrow.Int8Traits.CastToBytes(rawData.Data)
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_Int8Vector,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Data: &schemapb.VectorField_Int8Vector{
							Int8Vector: dataBytes,
						},
						Dim: int64(rawData.Dim),
					},
				},
				ValidData: rawData.ValidData,
			}
		case *VectorArrayFieldData:
			fieldData = &schemapb.FieldData{
				Type:    schemapb.DataType_ArrayOfVector,
				FieldId: fieldID,
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Data: &schemapb.VectorField_VectorArray{
							VectorArray: &schemapb.VectorArray{
								Data:        rawData.Data,
								ElementType: rawData.ElementType,
								Dim:         rawData.Dim,
							},
						},
						Dim: rawData.Dim,
					},
				},
				ValidData: rawData.ValidData,
			}
		default:
			return insertRecord, merr.WrapErrServiceInternalMsg("unsupported data type when transter storage.InsertData to internalpb.InsertRecord")
		}

		insertRecord.FieldsData = append(insertRecord.FieldsData, fieldData)
		insertRecord.NumRows = int64(rawData.RowNum())
	}

	return insertRecord, nil
}

// TransferInsertMsgToInsertRecord converts an insert message into a segcore
// insert record under the given schema. Payload columns of fields absent from
// the schema are skipped and their field IDs returned, so the caller decides
// how to surface the drop (log/metric) with its own context. Only the
// column-based branch applies this filtering: the legacy row-based format
// predates schema change and cannot carry since-dropped fields.
func TransferInsertMsgToInsertRecord(schema *schemapb.CollectionSchema, msg *msgstream.InsertMsg) (*segcorepb.InsertRecord, []int64, error) {
	if msg.IsRowBased() {
		insertData, err := RowBasedInsertMsgToInsertData(msg, schema, false)
		if err != nil {
			return nil, nil, err
		}
		insertRecord, err := TransferInsertDataToInsertRecord(insertData)
		return insertRecord, nil, err
	}

	// column base insert msg
	if err := validateColumnBasedInsertMsgNullableVectors(schema, msg); err != nil {
		return nil, nil, err
	}

	insertRecord := &segcorepb.InsertRecord{
		NumRows:    int64(msg.NumRows),
		FieldsData: make([]*schemapb.FieldData, 0, len(msg.FieldsData)),
	}

	// The payload was written under the schema version at WAL-append time. On WAL
	// replay after schema changes, it may still carry columns of since-dropped
	// fields, while the growing segment is rebuilt with the current schema and can
	// never own such columns (segcore hard-asserts on unknown fields). Dropped-field
	// data is unqueryable by definition, so skip those columns instead of passing
	// them through.
	knownFields := typeutil.NewSet[int64]()
	for _, field := range typeutil.GetAllFieldSchemas(schema) {
		knownFields.Insert(field.GetFieldID())
	}
	var skippedFields []int64
	for _, fieldData := range msg.FieldsData {
		if !knownFields.Contain(fieldData.GetFieldId()) {
			skippedFields = append(skippedFields, fieldData.GetFieldId())
			continue
		}
		insertRecord.FieldsData = append(insertRecord.FieldsData, fieldData)
	}

	return insertRecord, skippedFields, nil
}

func Min(a, b int64) int64 {
	if a < b {
		return a
	}
	return b
}

func NewTestChunkManagerFactory(params *paramtable.ComponentParam, rootPath string) *ChunkManagerFactory {
	return NewChunkManagerFactory("minio",
		objectstorage.RootPath(rootPath),
		objectstorage.Address(params.MinioCfg.Address.GetValue()),
		objectstorage.AccessKeyID(params.MinioCfg.AccessKeyID.GetValue()),
		objectstorage.SecretAccessKeyID(params.MinioCfg.SecretAccessKey.GetValue()),
		objectstorage.UseSSL(params.MinioCfg.UseSSL.GetAsBool()),
		objectstorage.BucketName(params.MinioCfg.BucketName.GetValue()),
		objectstorage.UseIAM(params.MinioCfg.UseIAM.GetAsBool()),
		objectstorage.CloudProvider(params.MinioCfg.CloudProvider.GetValue()),
		objectstorage.IAMEndpoint(params.MinioCfg.IAMEndpoint.GetValue()),
		objectstorage.CreateBucket(true))
}

func GetFilesSize(ctx context.Context, paths []string, cm ChunkManager) (int64, error) {
	totalSize := int64(0)
	for _, filePath := range paths {
		size, err := cm.Size(ctx, filePath)
		if err != nil {
			return 0, err
		}
		totalSize += size
	}
	return totalSize, nil
}

type NullableInt struct {
	Value *int
}

// NewNullableInt creates a new NullableInt instance
func NewNullableInt(value int) *NullableInt {
	return &NullableInt{Value: &value}
}

func (ni NullableInt) GetValue() int {
	return *ni.Value
}

// IsNull checks if the NullableInt is null
func (ni NullableInt) IsNull() bool {
	return ni.Value == nil
}

func GetDefaultValue(fieldSchema *schemapb.FieldSchema) interface{} {
	switch fieldSchema.DataType {
	case schemapb.DataType_Bool:
		return fieldSchema.GetDefaultValue().GetBoolData()
	case schemapb.DataType_Int8:
		return int8(fieldSchema.GetDefaultValue().GetIntData())
	case schemapb.DataType_Int16:
		return int16(fieldSchema.GetDefaultValue().GetIntData())
	case schemapb.DataType_Int32:
		return fieldSchema.GetDefaultValue().GetIntData()
	case schemapb.DataType_Int64:
		return fieldSchema.GetDefaultValue().GetLongData()
	case schemapb.DataType_Float:
		return fieldSchema.GetDefaultValue().GetFloatData()
	case schemapb.DataType_Double:
		return fieldSchema.GetDefaultValue().GetDoubleData()
	case schemapb.DataType_VarChar, schemapb.DataType_String:
		return fieldSchema.GetDefaultValue().GetStringData()
	case schemapb.DataType_Timestamptz:
		return fieldSchema.GetDefaultValue().GetTimestamptzData()
	case schemapb.DataType_JSON:
		return fieldSchema.GetDefaultValue().GetBytesData()
	case schemapb.DataType_Geometry:
		// ignore err because the default value has been checked when create collection.
		wkbValue, _ := common.ConvertWKTToWKB(fieldSchema.GetDefaultValue().GetStringData())
		return wkbValue

	default:
		// won't happen
		panic(fmt.Sprintf("undefined data type:%s", fieldSchema.DataType.String()))
	}
}

// fillMissingFields fills default values or null values for missing fields in insertData
func fillMissingFields(schema *schemapb.CollectionSchema, insertData *InsertData) error {
	batchRows := int64(insertData.GetRowNum())

	allFields := typeutil.GetAllFieldSchemas(schema)
	for _, field := range allFields {
		// Skip embedding function output fields and system fields.
		if isEmbeddingFunctionOutputField(field, schema) || field.GetFieldID() < 100 {
			continue
		}

		_, exists := insertData.Data[field.GetFieldID()]

		if !exists {
			// Create default field data if not found
			fieldData, err := NewFieldData(field.DataType, field, int(batchRows))
			if err != nil {
				return merr.WrapErrServiceInternalMsg("failed to create default field data for field %s: %v", field.Name, err)
			}

			if field.GetDefaultValue() != nil { // Fill with default value
				defaultValue := GetDefaultValue(field)

				for j := 0; j < int(batchRows); j++ {
					if err := fieldData.AppendRow(defaultValue); err != nil {
						return merr.WrapErrServiceInternalMsg("failed to append default value for field %s: %v", field.Name, err)
					}
				}
			} else if field.GetNullable() { // Fill with null values
				for j := 0; j < int(batchRows); j++ {
					if err := fieldData.AppendRow(nil); err != nil {
						return merr.WrapErrServiceInternalMsg("failed to append null value for field %s: %v", field.Name, err)
					}
				}
			} else {
				return merr.WrapErrServiceInternalMsg("field %s is not nullable and has no default value", field.Name)
			}
			insertData.Data[field.GetFieldID()] = fieldData
		}
	}
	return nil
}

// sort by field binlogs key
func SortFieldBinlogs(fieldBinlogs map[int64]*datapb.FieldBinlog) []*datapb.FieldBinlog {
	fieldIDs := lo.Keys(fieldBinlogs)
	sort.Slice(fieldIDs, func(i, j int) bool {
		return fieldIDs[i] < fieldIDs[j]
	})
	binlogs := make([]*datapb.FieldBinlog, 0, len(fieldIDs))
	for _, fieldID := range fieldIDs {
		binlogs = append(binlogs, fieldBinlogs[fieldID])
	}
	return binlogs
}

// VectorArrayToArrowType converts VectorArray element type to the corresponding Arrow type
// Note: This returns the element type (e.g., FixedSizeBinary), not a list type
// The caller is responsible for wrapping it in a list if needed
func VectorArrayToArrowType(elementType schemapb.DataType, dim int) (arrow.DataType, error) {
	switch elementType {
	case schemapb.DataType_FloatVector:
		// Each vector is stored as a fixed-size binary chunk
		return &arrow.FixedSizeBinaryType{ByteWidth: dim * 4}, nil
	case schemapb.DataType_BinaryVector:
		return &arrow.FixedSizeBinaryType{ByteWidth: (dim + 7) / 8}, nil
	case schemapb.DataType_Float16Vector:
		return &arrow.FixedSizeBinaryType{ByteWidth: dim * 2}, nil
	case schemapb.DataType_BFloat16Vector:
		return &arrow.FixedSizeBinaryType{ByteWidth: dim * 2}, nil
	case schemapb.DataType_Int8Vector:
		return &arrow.FixedSizeBinaryType{ByteWidth: dim}, nil
	default:
		return nil, merr.WrapErrParameterInvalidMsg("unsupported element type in VectorArray: %s", elementType.String())
	}
}

func getTTLFieldID(schema *schemapb.CollectionSchema) int64 {
	ttlFieldName := ""
	for _, pair := range schema.GetProperties() {
		if pair.GetKey() == common.CollectionTTLFieldKey {
			ttlFieldName = pair.GetValue()
			break
		}
	}
	if ttlFieldName == "" {
		return -1
	}
	for _, field := range schema.GetFields() {
		if field.GetName() == ttlFieldName && field.GetDataType() == schemapb.DataType_Timestamptz {
			return field.GetFieldID()
		}
	}
	return -1
}

// calculateExpirQuantiles computes TTL values at 20%, 40%, 60%, 80%, 100% percentiles.
// Returns nil if ttlFieldID is not enabled or no rows exist.
// Precondition: ttlFieldValues must contain only positive values (>0); the caller is responsible
// for filtering out null/zero TTL values which represent "never expire" rows.
func calculateExpirQuantiles(ttlFieldID int64, rowNum int64, ttlFieldValues []int64) []int64 {
	// If ttl field is not enabled for this writer, do not emit percentile info.
	if ttlFieldID <= 1 {
		return nil
	}

	// If segment is empty, do not emit percentile info.
	if rowNum <= 0 {
		return nil
	}

	sort.Slice(ttlFieldValues, func(i, j int) bool {
		return ttlFieldValues[i] < ttlFieldValues[j]
	})

	// Calculate percentile indices for 20%, 40%, 60%, 80%, 100%
	percentiles := []float64{0.2, 0.4, 0.6, 0.8, 1.0}
	result := make([]int64, len(percentiles))

	// Treat rows with null/<=0 ttl as "never expire".
	const neverExpire = int64(^uint64(0) >> 1)

	for i, p := range percentiles {
		// Calculate index: for n elements, p percentile means ceil(n * p) elements
		// e.g., for n=5: 20%->idx 0, 40%->idx 1, 60%->idx 2, 80%->idx 3, 100%->idx 4
		idx := int(math.Ceil(p*float64(rowNum))) - 1
		if idx < 0 {
			idx = 0
		}
		// If idx exceeds collected ttl values, it falls into "never expire" region.
		if idx >= len(ttlFieldValues) {
			result[i] = neverExpire
			continue
		}
		result[i] = ttlFieldValues[idx]
	}

	return result
}