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

import (
	"context"
	"encoding/base64"
	"fmt"
	"math"
	"net/http"
	"reflect"
	"strconv"
	"strings"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/spf13/cast"
	"github.com/tidwall/gjson"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/propagation"
	oteltrace "go.opentelemetry.io/otel/trace"
	"google.golang.org/protobuf/proto"

	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
	"github.com/milvus-io/milvus-proto/go-api/v3/milvuspb"
	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
	mhttp "github.com/milvus-io/milvus/internal/http"
	"github.com/milvus-io/milvus/internal/json"
	"github.com/milvus-io/milvus/internal/proxy"
	"github.com/milvus-io/milvus/internal/proxy/accesslog"
	"github.com/milvus-io/milvus/internal/types"
	"github.com/milvus-io/milvus/internal/util/function/chain"
	"github.com/milvus-io/milvus/pkg/v3/common"
	"github.com/milvus-io/milvus/pkg/v3/metrics"
	"github.com/milvus-io/milvus/pkg/v3/mlog"
	"github.com/milvus-io/milvus/pkg/v3/util"
	"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/parameterutil"
	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
)

func HTTPReturn(c *gin.Context, code int, result gin.H) {
	c.Set(HTTPReturnCode, result[HTTPReturnCode])
	if errorMsg, ok := result[HTTPReturnMessage]; ok {
		c.Set(HTTPReturnMessage, errorMsg)
	}
	setTraceIDHeader(c)
	c.JSON(code, result)
}

// HTTPReturnStream uses custom jsonRender that encodes JSON data directly into the response writer.
// Timeout-wrapped REST routes still buffer encoded bytes before committing them to the client.
func HTTPReturnStream(c *gin.Context, code int, result gin.H) {
	c.Set(HTTPReturnCode, result[HTTPReturnCode])
	if errorMsg, ok := result[HTTPReturnMessage]; ok {
		c.Set(HTTPReturnMessage, errorMsg)
	}
	setTraceIDHeader(c)
	c.Render(code, jsonRender{Data: result})
}

func HTTPAbortReturn(c *gin.Context, code int, result gin.H) {
	c.Set(HTTPReturnCode, result[HTTPReturnCode])
	if errorMsg, ok := result[HTTPReturnMessage]; ok {
		c.Set(HTTPReturnMessage, errorMsg)
	}
	setTraceIDHeader(c)
	c.AbortWithStatusJSON(code, result)
}

func TraceIDHandlerFunc(c *gin.Context) {
	ctx := otel.GetTextMapPropagator().Extract(c.Request.Context(), propagation.HeaderCarrier(c.Request.Header))
	ctx, span := otel.Tracer(typeutil.ProxyRole).Start(ctx, c.Request.URL.Path)
	defer span.End()

	traceID := span.SpanContext().TraceID()
	if traceID.IsValid() {
		traceIDStr := traceID.String()
		c.Set("traceID", traceIDStr)
		c.Request = c.Request.WithContext(ctx)
		setTraceIDHeader(c)
	}

	c.Next()
}

func getTraceID(c *gin.Context) (string, bool) {
	traceID, ok := c.Get("traceID")
	if ok {
		traceIDStr, ok := traceID.(string)
		if ok && traceIDStr != "" {
			return traceIDStr, true
		}
	}

	if c.Request == nil {
		return "", false
	}
	spanTraceID := oteltrace.SpanFromContext(c.Request.Context()).SpanContext().TraceID()
	if !spanTraceID.IsValid() {
		return "", false
	}
	return spanTraceID.String(), true
}

func setTraceIDHeader(c *gin.Context) {
	traceID, ok := getTraceID(c)
	if !ok {
		return
	}
	setTraceIDHeaderTo(c.Writer.Header(), traceID)
}

func setTraceIDHeaderTo(header http.Header, traceID string) {
	header.Set(HTTPHeaderMilvusTraceID, traceID)
}

func ParseUsernamePassword(c *gin.Context) (string, string, bool) {
	username, password, ok := c.Request.BasicAuth()
	if !ok {
		token := GetAuthorization(c)
		i := strings.IndexAny(token, util.CredentialSeparator)
		if i != -1 {
			username = token[:i]
			password = token[i+1:]
		}
	} else {
		c.Header("WWW-Authenticate", `Basic realm="restricted", charset="UTF-8"`)
	}
	return username, password, username != "" && password != ""
}

func GetAuthorization(c *gin.Context) string {
	auth := c.Request.Header.Get("Authorization")
	return strings.TrimPrefix(auth, "Bearer ")
}

// find the primary field of collection
func getPrimaryField(schema *schemapb.CollectionSchema) (*schemapb.FieldSchema, bool) {
	for _, field := range schema.Fields {
		if field.IsPrimaryKey {
			return field, true
		}
	}
	return nil, false
}

func joinArray(data interface{}) string {
	if data == nil {
		return ""
	}
	var builder strings.Builder
	switch arr := data.(type) {
	case []int64:
		for i, v := range arr {
			if i > 0 {
				builder.WriteString(",")
			}
			builder.WriteString(strconv.FormatInt(v, 10))
		}
	case []string:
		for i, v := range arr {
			if i > 0 {
				builder.WriteString(",")
			}
			builder.WriteString(v)
		}
	default:
		rv := reflect.ValueOf(data)
		if rv.Kind() != reflect.Slice && rv.Kind() != reflect.Array {
			return ""
		}
		for i := 0; i < rv.Len(); i++ {
			if i > 0 {
				builder.WriteString(",")
			}
			fmt.Fprintf(&builder, "%v", rv.Index(i))
		}
	}
	return builder.String()
}

func convertRange(field *schemapb.FieldSchema, result gjson.Result) (string, error) {
	var resultStr string
	fieldType := field.DataType

	switch fieldType {
	case schemapb.DataType_Int64:
		dataArray := make([]int64, 0, len(result.Array()))
		for _, data := range result.Array() {
			if data.Type == gjson.String {
				value, err := cast.ToInt64E(data.Str)
				if err != nil {
					return "", err
				}
				dataArray = append(dataArray, value)
			} else {
				value, err := cast.ToInt64E(data.Raw)
				if err != nil {
					return "", err
				}
				dataArray = append(dataArray, value)
			}
		}
		resultStr = joinArray(dataArray)
	case schemapb.DataType_VarChar:
		dataArray := make([]string, 0, len(result.Array()))
		for _, data := range result.Array() {
			value, err := cast.ToStringE(data.Str)
			if err != nil {
				return "", err
			}
			dataArray = append(dataArray, fmt.Sprintf(`"%s"`, value))
		}
		resultStr = joinArray(dataArray)
	}
	return resultStr, nil
}

// generate the expression: $primaryFieldName in [1,2,3]
func checkGetPrimaryKey(coll *schemapb.CollectionSchema, idResult gjson.Result) (string, error) {
	primaryField, ok := getPrimaryField(coll)
	if !ok {
		return "", merr.WrapErrParameterInvalidMsg("collection: %s has no primary field", coll.Name)
	}
	resultStr, err := convertRange(primaryField, idResult)
	if err != nil {
		return "", err
	}
	filter := primaryField.Name + " in [" + resultStr + "]"
	return filter, nil
}

// convertIDsToSchemapbIDs converts a slice of interface{} (JSON ids) to schemapb.IDs
// based on the primary key field type
func convertIDsToSchemapbIDs(ids []interface{}, pkField *schemapb.FieldSchema) (*schemapb.IDs, error) {
	if len(ids) == 0 {
		return nil, merr.WrapErrParameterMissingMsg("ids array cannot be empty")
	}

	switch pkField.DataType {
	case schemapb.DataType_Int64:
		int64IDs := make([]int64, 0, len(ids))
		for i, id := range ids {
			var int64ID int64
			switch v := id.(type) {
			case int64:
				int64ID = v
			case int:
				int64ID = int64(v)
			case float64:
				// JSON numbers are decoded as float64
				// Check if the float has a fractional part
				if v != math.Trunc(v) {
					return nil, merr.WrapErrParameterInvalidMsg("invalid int64 id at index %d: %v has fractional part", i, v)
				}
				int64ID = int64(v)
			case string:
				// Try to parse string as int64
				parsed, err := strconv.ParseInt(v, 10, 64)
				if err != nil {
					return nil, merr.WrapErrParameterInvalidErr(err, "invalid int64 id at index %d: %v", i, id)
				}
				int64ID = parsed
			default:
				return nil, merr.WrapErrParameterInvalidMsg("invalid id type at index %d: expected int64, got %T", i, id)
			}
			int64IDs = append(int64IDs, int64ID)
		}
		return &schemapb.IDs{
			IdField: &schemapb.IDs_IntId{
				IntId: &schemapb.LongArray{
					Data: int64IDs,
				},
			},
		}, nil

	case schemapb.DataType_VarChar:
		stringIDs := make([]string, 0, len(ids))
		for i, id := range ids {
			var stringID string
			switch v := id.(type) {
			case string:
				stringID = v
			case int64, int, float64:
				// Convert number to string
				stringID = fmt.Sprintf("%v", v)
			default:
				return nil, merr.WrapErrParameterInvalidMsg("invalid id type at index %d: expected string, got %T", i, id)
			}
			if stringID == "" {
				return nil, merr.WrapErrParameterInvalidMsg("empty string id at index %d", i)
			}
			stringIDs = append(stringIDs, stringID)
		}
		return &schemapb.IDs{
			IdField: &schemapb.IDs_StrId{
				StrId: &schemapb.StringArray{
					Data: stringIDs,
				},
			},
		}, nil

	default:
		return nil, merr.WrapErrParameterInvalidMsg("unsupported primary key type: %s", pkField.DataType.String())
	}
}

// --------------------- collection details --------------------- //

func printFields(fields []*schemapb.FieldSchema) []gin.H {
	res := make([]gin.H, 0, len(fields))
	for _, field := range fields {
		if field.Name == common.MetaFieldName || field.Name == common.NamespaceFieldName {
			continue
		}
		fieldDetail := printFieldDetail(field, true)
		res = append(res, fieldDetail)
	}
	return res
}

func printFieldsV2(fields []*schemapb.FieldSchema) []gin.H {
	res := make([]gin.H, 0, len(fields))
	for _, field := range fields {
		if field.Name == common.MetaFieldName || field.Name == common.NamespaceFieldName {
			continue
		}
		fieldDetail := printFieldDetail(field, false)
		res = append(res, fieldDetail)
	}
	return res
}

func printFieldDetail(field *schemapb.FieldSchema, oldVersion bool) gin.H {
	fieldDetail := gin.H{
		HTTPReturnFieldName:          field.Name,
		HTTPReturnFieldPrimaryKey:    field.IsPrimaryKey,
		HTTPReturnFieldPartitionKey:  field.IsPartitionKey,
		HTTPReturnFieldClusteringKey: field.IsClusteringKey,
		HTTPReturnFieldAutoID:        field.AutoID,
		HTTPReturnDescription:        field.Description,
		HTTPReturnFieldNullable:      field.Nullable,
	}
	if field.DefaultValue != nil {
		fieldDetail[HTTPRequestDefaultValue] = field.DefaultValue
	}
	if field.GetIsFunctionOutput() {
		fieldDetail[HTTPReturnFieldIsFunctionOutput] = true
	}
	if field.GetExternalField() != "" {
		fieldDetail["externalField"] = field.GetExternalField()
	}
	if typeutil.IsVectorType(field.DataType) {
		fieldDetail[HTTPReturnFieldType] = field.DataType.String()
		if oldVersion {
			dim, _ := getDim(field)
			fieldDetail[HTTPReturnFieldType] = field.DataType.String() + "(" + strconv.FormatInt(dim, 10) + ")"
		}
	} else if field.DataType == schemapb.DataType_VarChar {
		fieldDetail[HTTPReturnFieldType] = field.DataType.String()
		if oldVersion {
			maxLength, _ := parameterutil.GetMaxLength(field)
			fieldDetail[HTTPReturnFieldType] = field.DataType.String() + "(" + strconv.FormatInt(maxLength, 10) + ")"
		}
	} else {
		fieldDetail[HTTPReturnFieldType] = field.DataType.String()
	}
	if !oldVersion {
		fieldDetail[HTTPReturnFieldID] = field.FieldID
		if field.TypeParams != nil {
			fieldDetail[Params] = field.TypeParams
		}
		if field.DataType == schemapb.DataType_Array || field.DataType == schemapb.DataType_ArrayOfVector {
			fieldDetail[HTTPReturnFieldElementType] = field.GetElementType().String()
		}
	}
	return fieldDetail
}

func printStructArrayFieldsV2(structFields []*schemapb.StructArrayFieldSchema) []gin.H {
	res := make([]gin.H, 0, len(structFields))
	for _, sf := range structFields {
		subs := make([]gin.H, 0, len(sf.GetFields()))
		for _, sub := range sf.GetFields() {
			detail := printFieldDetail(sub, false)
			if short, err := typeutil.ExtractStructFieldName(sub.GetName()); err == nil && short != "" {
				detail[HTTPReturnFieldName] = short
			}
			subs = append(subs, detail)
		}
		entry := gin.H{
			HTTPReturnFieldName:     sf.GetName(),
			HTTPReturnFieldID:       sf.GetFieldID(),
			HTTPReturnDescription:   sf.GetDescription(),
			HTTPReturnFieldNullable: sf.GetNullable(),
			HTTPReturnFieldType:     schemapb.DataType_ArrayOfStruct.String(),
			"fields":                subs,
		}
		if len(sf.GetTypeParams()) > 0 {
			entry[Params] = sf.GetTypeParams()
		}
		res = append(res, entry)
	}
	return res
}

func printFunctionDetails(functions []*schemapb.FunctionSchema) []gin.H {
	res := make([]gin.H, 0, len(functions))
	for _, function := range functions {
		res = append(res, gin.H{
			HTTPReturnFunctionName:             function.Name,
			HTTPReturnDescription:              function.Description,
			HTTPReturnFunctionType:             function.Type,
			HTTPReturnFunctionID:               function.Id,
			HTTPReturnFunctionInputFieldNames:  function.InputFieldNames,
			HTTPReturnFunctionOutputFieldNames: function.OutputFieldNames,
			HTTPReturnFunctionParams:           function.Params,
		})
	}
	return res
}

func getMetricType(pairs []*commonpb.KeyValuePair) string {
	metricType := DefaultMetricType
	for _, pair := range pairs {
		if pair.Key == common.MetricTypeKey {
			metricType = pair.Value
			break
		}
	}
	return metricType
}

func printIndexes(indexes []*milvuspb.IndexDescription) []gin.H {
	res := make([]gin.H, 0, len(indexes))
	for _, index := range indexes {
		res = append(res, gin.H{
			HTTPIndexName:             index.IndexName,
			HTTPIndexField:            index.FieldName,
			HTTPReturnIndexMetricType: getMetricType(index.Params),
		})
	}
	return res
}

// --------------------- insert param --------------------- //

func checkAndSetData(body []byte, collSchema *schemapb.CollectionSchema, partialUpdate bool) ([]map[string]interface{}, map[string][]bool, error) {
	var reallyDataArray []map[string]interface{}
	validDataMap := make(map[string][]bool)
	dataResult := gjson.GetBytes(body, HTTPRequestData)
	dataResultArray := dataResult.Array()
	if len(dataResultArray) == 0 {
		return reallyDataArray, validDataMap, merr.ErrMissingRequiredParameters
	}

	fieldNames := make([]string, 0, len(collSchema.Fields)+len(collSchema.StructArrayFields))
	for _, field := range collSchema.Fields {
		if field.IsDynamic {
			continue
		}
		fieldNames = append(fieldNames, field.Name)
	}
	for _, structField := range collSchema.StructArrayFields {
		fieldNames = append(fieldNames, structField.GetName())
	}

	for _, data := range dataResultArray {
		reallyData := map[string]interface{}{}
		if data.Type == gjson.JSON {
			for _, structField := range collSchema.StructArrayFields {
				rawValue := gjson.Get(data.Raw, structField.GetName())
				if partialUpdate && !rawValue.Exists() {
					continue
				}
				structRow, err := parseStructArrayRow(rawValue.Raw, structField)
				if err != nil {
					return reallyDataArray, validDataMap, err
				}
				reallyData[structField.GetName()] = structRow
			}
			for _, field := range collSchema.Fields {
				if field.IsDynamic {
					continue
				}
				fieldType := field.DataType
				fieldName := field.Name
				fieldValue := data.Get(fieldName)

				// For partial update, missing fields mean "do not update this field".
				// Explicit JSON null is handled below as an update to null for nullable fields.
				if partialUpdate && !fieldValue.Exists() {
					continue
				}

				if field.Nullable || field.DefaultValue != nil {
					if fieldValue.Type == gjson.Null {
						validDataMap[fieldName] = append(validDataMap[fieldName], false)
						continue
					} else {
						validDataMap[fieldName] = append(validDataMap[fieldName], true)
					}
				}

				dataString := fieldValue.String()
				// if has pass pk than just to try to set it
				if field.IsPrimaryKey && field.AutoID && len(dataString) == 0 {
					continue
				}

				// skip function output field if user didn't provide data,
				// let proxy validate when data is provided
				if field.GetIsFunctionOutput() && dataString == "" {
					continue
				}

				switch fieldType {
				case schemapb.DataType_FloatVector:
					if dataString == "" {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], "", "missing vector field: "+fieldName)
					}
					var vectorArray []float32
					err := json.Unmarshal([]byte(dataString), &vectorArray)
					if err != nil {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, err.Error())
					}
					reallyData[fieldName] = vectorArray
				case schemapb.DataType_BinaryVector:
					if dataString == "" {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], "", "missing vector field: "+fieldName)
					}
					vectorStr := fieldValue.Raw
					var vectorArray []byte
					err := json.Unmarshal([]byte(vectorStr), &vectorArray)
					if err != nil {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, err.Error())
					}
					reallyData[fieldName] = vectorArray
				case schemapb.DataType_SparseFloatVector:
					if dataString == "" {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], "", "missing vector field: "+fieldName)
					}
					sparseVec, err := typeutil.CreateSparseFloatRowFromJSON([]byte(dataString))
					if err != nil {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, err.Error())
					}
					reallyData[fieldName] = sparseVec
				case schemapb.DataType_Float16Vector:
					fallthrough
				case schemapb.DataType_BFloat16Vector:
					if dataString == "" {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], "", "missing vector field: "+fieldName)
					}
					vectorJSON := fieldValue
					// Clients may send float32 vector because they are inconvenient of processing float16 or bfloat16.
					// Float32 vector is an array in JSON format, like `[1.0, 2.0, 3.0]`, `[1, 2, 3]`, etc,
					// while float16 or bfloat16 vector is a string in JSON format, like `"4z1jPgAAgL8="`, `"gD+AP4A/gD8="`, etc.
					if vectorJSON.IsArray() {
						// `data` is a float32 vector
						// same as `case schemapb.DataType_FloatVector`
						var vectorArray []float32
						err := json.Unmarshal([]byte(dataString), &vectorArray)
						if err != nil {
							return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, err.Error())
						}
						reallyData[fieldName] = vectorArray
					} else if vectorJSON.Type == gjson.String {
						// `data` is a float16 or bfloat16 vector
						// same as `case schemapb.DataType_BinaryVector`
						vectorStr := fieldValue.Raw
						var vectorArray []byte
						err := json.Unmarshal([]byte(vectorStr), &vectorArray)
						if err != nil {
							return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, err.Error())
						}
						reallyData[fieldName] = vectorArray
					} else {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, "invalid vector field: "+fieldName)
					}
				case schemapb.DataType_Int8Vector:
					if dataString == "" {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], "", "missing vector field: "+fieldName)
					}
					var vectorArray []int8
					err := json.Unmarshal([]byte(dataString), &vectorArray)
					if err != nil {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, err.Error())
					}
					reallyData[fieldName] = vectorArray
				case schemapb.DataType_Bool:
					result, err := cast.ToBoolE(dataString)
					if err != nil {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, err.Error())
					}
					reallyData[fieldName] = result
				case schemapb.DataType_Int8:
					result, err := cast.ToInt8E(dataString)
					if err != nil {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, err.Error())
					}
					reallyData[fieldName] = result
				case schemapb.DataType_Int16:
					result, err := cast.ToInt16E(dataString)
					if err != nil {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, err.Error())
					}
					reallyData[fieldName] = result
				case schemapb.DataType_Int32:
					result, err := cast.ToInt32E(dataString)
					if err != nil {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, err.Error())
					}
					reallyData[fieldName] = result
				case schemapb.DataType_Int64:
					result, err := json.Number(dataString).Int64()
					if err != nil {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, err.Error())
					}
					reallyData[fieldName] = result
				case schemapb.DataType_Array:
					switch field.ElementType {
					case schemapb.DataType_Bool:
						arr := make([]bool, 0)
						err := json.Unmarshal([]byte(dataString), &arr)
						if err != nil {
							return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)]+
								" of "+schemapb.DataType_name[int32(field.ElementType)], dataString, err.Error())
						}
						reallyData[fieldName] = &schemapb.ScalarField{
							Data: &schemapb.ScalarField_BoolData{
								BoolData: &schemapb.BoolArray{
									Data: arr,
								},
							},
						}
					case schemapb.DataType_Int8:
						arr := make([]int32, 0)
						err := json.Unmarshal([]byte(dataString), &arr)
						if err != nil {
							return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)]+
								" of "+schemapb.DataType_name[int32(field.ElementType)], dataString, err.Error())
						}
						reallyData[fieldName] = &schemapb.ScalarField{
							Data: &schemapb.ScalarField_IntData{
								IntData: &schemapb.IntArray{
									Data: arr,
								},
							},
						}
					case schemapb.DataType_Int16:
						arr := make([]int32, 0)
						err := json.Unmarshal([]byte(dataString), &arr)
						if err != nil {
							return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)]+
								" of "+schemapb.DataType_name[int32(field.ElementType)], dataString, err.Error())
						}
						reallyData[fieldName] = &schemapb.ScalarField{
							Data: &schemapb.ScalarField_IntData{
								IntData: &schemapb.IntArray{
									Data: arr,
								},
							},
						}
					case schemapb.DataType_Int32:
						arr := make([]int32, 0)
						err := json.Unmarshal([]byte(dataString), &arr)
						if err != nil {
							return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)]+
								" of "+schemapb.DataType_name[int32(field.ElementType)], dataString, err.Error())
						}
						reallyData[fieldName] = &schemapb.ScalarField{
							Data: &schemapb.ScalarField_IntData{
								IntData: &schemapb.IntArray{
									Data: arr,
								},
							},
						}
					case schemapb.DataType_Int64:
						arr := make([]int64, 0)
						err := json.Unmarshal([]byte(dataString), &arr)
						if err != nil {
							return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)]+
								" of "+schemapb.DataType_name[int32(field.ElementType)], dataString, err.Error())
						}
						reallyData[fieldName] = &schemapb.ScalarField{
							Data: &schemapb.ScalarField_LongData{
								LongData: &schemapb.LongArray{
									Data: arr,
								},
							},
						}
					case schemapb.DataType_Float:
						arr := make([]float32, 0)
						err := json.Unmarshal([]byte(dataString), &arr)
						if err != nil {
							return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)]+
								" of "+schemapb.DataType_name[int32(field.ElementType)], dataString, err.Error())
						}
						reallyData[fieldName] = &schemapb.ScalarField{
							Data: &schemapb.ScalarField_FloatData{
								FloatData: &schemapb.FloatArray{
									Data: arr,
								},
							},
						}
					case schemapb.DataType_Double:
						arr := make([]float64, 0)
						err := json.Unmarshal([]byte(dataString), &arr)
						if err != nil {
							return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)]+
								" of "+schemapb.DataType_name[int32(field.ElementType)], dataString, err.Error())
						}
						reallyData[fieldName] = &schemapb.ScalarField{
							Data: &schemapb.ScalarField_DoubleData{
								DoubleData: &schemapb.DoubleArray{
									Data: arr,
								},
							},
						}
					case schemapb.DataType_VarChar:
						arr := make([]string, 0)
						err := json.Unmarshal([]byte(dataString), &arr)
						if err != nil {
							return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)]+
								" of "+schemapb.DataType_name[int32(field.ElementType)], dataString, err.Error())
						}
						reallyData[fieldName] = &schemapb.ScalarField{
							Data: &schemapb.ScalarField_StringData{
								StringData: &schemapb.StringArray{
									Data: arr,
								},
							},
						}
					}
				case schemapb.DataType_JSON:
					reallyData[fieldName] = []byte(dataString)
				case schemapb.DataType_Geometry:
					reallyData[fieldName] = dataString
				case schemapb.DataType_Float:
					result, err := cast.ToFloat32E(dataString)
					if err != nil {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, err.Error())
					}
					reallyData[fieldName] = result
				case schemapb.DataType_Double:
					result, err := cast.ToFloat64E(dataString)
					if err != nil {
						return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(fieldType)], dataString, err.Error())
					}
					reallyData[fieldName] = result
				case schemapb.DataType_Timestamptz:
					reallyData[fieldName] = dataString
				case schemapb.DataType_VarChar:
					reallyData[fieldName] = dataString
				case schemapb.DataType_String:
					reallyData[fieldName] = dataString
				default:
					return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid("", schemapb.DataType_name[int32(fieldType)], "fieldName: "+fieldName)
				}
			}

			// fill dynamic schema

			for mapKey, mapValue := range data.Map() {
				if !containsString(fieldNames, mapKey) {
					if collSchema.EnableDynamicField {
						if mapKey == common.MetaFieldName {
							return nil, nil, merr.WrapErrParameterInvalidMsg("use the invalid field name(%s) when enable dynamicField", mapKey)
						}
						mapValueStr := mapValue.String()
						switch mapValue.Type {
						case gjson.True, gjson.False:
							reallyData[mapKey] = cast.ToBool(mapValueStr)
						case gjson.String:
							reallyData[mapKey] = mapValueStr
						case gjson.Number:
							if strings.Contains(mapValue.Raw, ".") {
								reallyData[mapKey] = cast.ToFloat64(mapValue.Raw)
							} else {
								reallyData[mapKey] = cast.ToInt64(mapValueStr)
							}
						case gjson.JSON:
							reallyData[mapKey] = mapValue.Value()
						case gjson.Null:
							// skip null
						default:
							mlog.Warn(context.TODO(), "unknown json type found", mlog.Int("mapValue.Type", int(mapValue.Type)))
						}
					} else {
						return nil, nil, merr.WrapErrParameterInvalidMsg("has pass more field without dynamic schema, please check it")
					}
				}
			}

			reallyDataArray = append(reallyDataArray, reallyData)
		} else {
			return reallyDataArray, validDataMap, merr.WrapErrParameterInvalid(gjson.JSON, data.Type, "NULL:0, FALSE:1, NUMBER:2, STRING:3, TRUE:4, JSON:5")
		}
	}
	return reallyDataArray, validDataMap, nil
}

func containsString(arr []string, s string) bool {
	for _, str := range arr {
		if str == s {
			return true
		}
	}
	return false
}

func getDim(field *schemapb.FieldSchema) (int64, error) {
	dimensionInSchema, err := funcutil.GetAttrByKeyFromRepeatedKV(common.DimKey, field.TypeParams)
	if err != nil {
		return 0, err
	}
	dim, err := strconv.Atoi(dimensionInSchema)
	if err != nil {
		return 0, err
	}
	return int64(dim), nil
}

type structArrayRow map[string]interface{}

func structFieldShortName(name string) string {
	short, err := typeutil.ExtractStructFieldName(name)
	if err != nil || short == "" {
		return name
	}
	return short
}

func subShortName(sub *schemapb.FieldSchema) string {
	return structFieldShortName(sub.GetName())
}

func parseStructArrayRow(rawJSON string, structSchema *schemapb.StructArrayFieldSchema) (structArrayRow, error) {
	if rawJSON == "" {
		return nil, merr.WrapErrParameterInvalidMsg("missing struct array field: %s", structSchema.GetName())
	}
	parsed := gjson.Parse(rawJSON)
	if !parsed.IsArray() {
		return nil, merr.WrapErrParameterInvalidMsg(
			"struct array field %s expects a JSON array of objects", structSchema.GetName())
	}
	elems := parsed.Array()
	collected := make(map[string][]gjson.Result, len(structSchema.GetFields()))
	expected := make(map[string]struct{}, len(structSchema.GetFields()))
	for _, sub := range structSchema.GetFields() {
		key := subShortName(sub)
		collected[key] = make([]gjson.Result, 0, len(elems))
		expected[key] = struct{}{}
	}
	for idx, elem := range elems {
		if elem.Type != gjson.JSON || !elem.IsObject() {
			return nil, merr.WrapErrParameterInvalidMsg(
				"struct array field %s element #%d must be a JSON object", structSchema.GetName(), idx)
		}
		seen := make(map[string]struct{}, len(structSchema.GetFields()))
		elem.ForEach(func(key, value gjson.Result) bool {
			name := key.String()
			if _, ok := expected[name]; !ok {
				return true
			}
			collected[name] = append(collected[name], value)
			seen[name] = struct{}{}
			return true
		})
		if len(seen) != len(expected) {
			for name := range expected {
				if _, ok := seen[name]; !ok {
					return nil, merr.WrapErrParameterInvalidMsg(
						"struct array field %s element #%d missing sub-field %s",
						structSchema.GetName(), idx, name)
				}
			}
		}
	}

	row := structArrayRow{}
	for _, sub := range structSchema.GetFields() {
		key := subShortName(sub)
		vals := collected[key]
		switch sub.GetDataType() {
		case schemapb.DataType_Array:
			scalar, err := buildStructSubArrayScalar(sub, vals)
			if err != nil {
				return nil, err
			}
			row[key] = scalar
		case schemapb.DataType_ArrayOfVector:
			vec, err := buildStructSubVectorField(sub, vals)
			if err != nil {
				return nil, err
			}
			row[key] = vec
		default:
			return nil, merr.WrapErrParameterInvalidMsg(
				"sub-field %s of struct %s has unsupported data type %s",
				key, structSchema.GetName(), sub.GetDataType())
		}
	}
	return row, nil
}

func buildStructSubArrayScalar(sub *schemapb.FieldSchema, vals []gjson.Result) (*schemapb.ScalarField, error) {
	switch sub.GetElementType() {
	case schemapb.DataType_Bool:
		arr := make([]bool, 0, len(vals))
		for _, v := range vals {
			if v.Type != gjson.True && v.Type != gjson.False {
				return nil, wrapStructSubParseError(sub, v, "expect bool")
			}
			arr = append(arr, v.Bool())
		}
		return &schemapb.ScalarField{
			Data: &schemapb.ScalarField_BoolData{BoolData: &schemapb.BoolArray{Data: arr}},
		}, nil
	case schemapb.DataType_Int8, schemapb.DataType_Int16, schemapb.DataType_Int32:
		arr := make([]int32, 0, len(vals))
		for _, v := range vals {
			if v.Type != gjson.Number {
				return nil, wrapStructSubParseError(sub, v, "expect integer")
			}
			arr = append(arr, int32(v.Int()))
		}
		return &schemapb.ScalarField{
			Data: &schemapb.ScalarField_IntData{IntData: &schemapb.IntArray{Data: arr}},
		}, nil
	case schemapb.DataType_Int64:
		arr := make([]int64, 0, len(vals))
		for _, v := range vals {
			if v.Type != gjson.Number {
				return nil, wrapStructSubParseError(sub, v, "expect integer")
			}
			arr = append(arr, v.Int())
		}
		return &schemapb.ScalarField{
			Data: &schemapb.ScalarField_LongData{LongData: &schemapb.LongArray{Data: arr}},
		}, nil
	case schemapb.DataType_Float:
		arr := make([]float32, 0, len(vals))
		for _, v := range vals {
			if v.Type != gjson.Number {
				return nil, wrapStructSubParseError(sub, v, "expect float")
			}
			arr = append(arr, float32(v.Float()))
		}
		return &schemapb.ScalarField{
			Data: &schemapb.ScalarField_FloatData{FloatData: &schemapb.FloatArray{Data: arr}},
		}, nil
	case schemapb.DataType_Double:
		arr := make([]float64, 0, len(vals))
		for _, v := range vals {
			if v.Type != gjson.Number {
				return nil, wrapStructSubParseError(sub, v, "expect double")
			}
			arr = append(arr, v.Float())
		}
		return &schemapb.ScalarField{
			Data: &schemapb.ScalarField_DoubleData{DoubleData: &schemapb.DoubleArray{Data: arr}},
		}, nil
	case schemapb.DataType_VarChar, schemapb.DataType_String:
		arr := make([]string, 0, len(vals))
		for _, v := range vals {
			if v.Type != gjson.String {
				return nil, wrapStructSubParseError(sub, v, "expect string")
			}
			arr = append(arr, v.String())
		}
		return &schemapb.ScalarField{
			Data: &schemapb.ScalarField_StringData{StringData: &schemapb.StringArray{Data: arr}},
		}, nil
	default:
		return nil, merr.WrapErrParameterInvalidMsg(
			"sub-field %s has unsupported array element type %s",
			sub.GetName(), sub.GetElementType())
	}
}

func buildStructSubVectorField(sub *schemapb.FieldSchema, vals []gjson.Result) (*schemapb.VectorField, error) {
	dim, err := getDim(sub)
	if err != nil {
		return nil, merr.WrapErrParameterInvalidMsg(
			"sub-field %s: %s", sub.GetName(), err.Error())
	}
	switch sub.GetElementType() {
	case schemapb.DataType_FloatVector:
		packed := &schemapb.FloatArray{}
		for _, v := range vals {
			if !v.IsArray() {
				return nil, wrapStructSubParseError(sub, v, "expect float vector array")
			}
			var row []float32
			if err := json.Unmarshal([]byte(v.Raw), &row); err != nil {
				return nil, wrapStructSubParseError(sub, v, err.Error())
			}
			if int64(len(row)) != dim {
				return nil, merr.WrapErrParameterInvalidMsg(
					"sub-field %s vector dim mismatch: expect %d, got %d",
					sub.GetName(), dim, len(row))
			}
			packed.Data = append(packed.Data, row...)
		}
		return &schemapb.VectorField{
			Dim:  dim,
			Data: &schemapb.VectorField_FloatVector{FloatVector: packed},
		}, nil
	case schemapb.DataType_Float16Vector, schemapb.DataType_BFloat16Vector:
		isFloat16 := sub.GetElementType() == schemapb.DataType_Float16Vector
		bytesPerVec := dim * 2
		buf := make([]byte, 0, int(bytesPerVec)*len(vals))
		for _, v := range vals {
			b, err := decodeByteVectorElement(v, dim, bytesPerVec, isFloat16)
			if err != nil {
				return nil, wrapStructSubParseError(sub, v, err.Error())
			}
			buf = append(buf, b...)
		}
		if isFloat16 {
			return &schemapb.VectorField{
				Dim:  dim,
				Data: &schemapb.VectorField_Float16Vector{Float16Vector: buf},
			}, nil
		}
		return &schemapb.VectorField{
			Dim:  dim,
			Data: &schemapb.VectorField_Bfloat16Vector{Bfloat16Vector: buf},
		}, nil
	case schemapb.DataType_BinaryVector:
		bytesPerVec := dim / 8
		buf := make([]byte, 0, int(bytesPerVec)*len(vals))
		for _, v := range vals {
			if v.Type != gjson.String {
				return nil, wrapStructSubParseError(sub, v, "binary vector must be base64-encoded string")
			}
			var row []byte
			if err := json.Unmarshal([]byte(v.Raw), &row); err != nil {
				return nil, wrapStructSubParseError(sub, v, err.Error())
			}
			if int64(len(row)) != bytesPerVec {
				return nil, merr.WrapErrParameterInvalidMsg(
					"sub-field %s binary vector byte-length mismatch: expect %d, got %d",
					sub.GetName(), bytesPerVec, len(row))
			}
			buf = append(buf, row...)
		}
		return &schemapb.VectorField{
			Dim:  dim,
			Data: &schemapb.VectorField_BinaryVector{BinaryVector: buf},
		}, nil
	case schemapb.DataType_Int8Vector:
		buf := make([]byte, 0, int(dim)*len(vals))
		for _, v := range vals {
			if !v.IsArray() {
				return nil, wrapStructSubParseError(sub, v, "expect int8 vector array")
			}
			var row []int8
			if err := json.Unmarshal([]byte(v.Raw), &row); err != nil {
				return nil, wrapStructSubParseError(sub, v, err.Error())
			}
			if int64(len(row)) != dim {
				return nil, merr.WrapErrParameterInvalidMsg(
					"sub-field %s int8 vector dim mismatch: expect %d, got %d",
					sub.GetName(), dim, len(row))
			}
			for _, x := range row {
				buf = append(buf, byte(x))
			}
		}
		return &schemapb.VectorField{
			Dim:  dim,
			Data: &schemapb.VectorField_Int8Vector{Int8Vector: buf},
		}, nil
	default:
		return nil, merr.WrapErrParameterInvalidMsg(
			"sub-field %s has unsupported vector element type %s",
			sub.GetName(), sub.GetElementType())
	}
}

func decodeByteVectorElement(v gjson.Result, dim, bytesPerVec int64, isFloat16 bool) ([]byte, error) {
	if v.IsArray() {
		var row []float32
		if err := json.Unmarshal([]byte(v.Raw), &row); err != nil {
			return nil, err
		}
		if int64(len(row)) != dim {
			return nil, merr.WrapErrParameterInvalidMsg("vector dim mismatch: expect %d, got %d", dim, len(row))
		}
		if isFloat16 {
			return typeutil.Float32ArrayToFloat16Bytes(row), nil
		}
		return typeutil.Float32ArrayToBFloat16Bytes(row), nil
	}
	if v.Type != gjson.String {
		return nil, merr.WrapErrParameterInvalidMsg("expect float vector array or base64 string")
	}
	var row []byte
	if err := json.Unmarshal([]byte(v.Raw), &row); err != nil {
		return nil, err
	}
	if int64(len(row)) != bytesPerVec {
		return nil, merr.WrapErrParameterInvalidMsg("byte length mismatch: expect %d, got %d", bytesPerVec, len(row))
	}
	return row, nil
}

func wrapStructSubParseError(sub *schemapb.FieldSchema, v gjson.Result, msg string) error {
	return merr.WrapErrParameterInvalidMsg(
		"sub-field %s parse error: %s (value=%s)", sub.GetName(), msg, v.Raw)
}

func isEmbeddingListData(body string) bool {
	raw := gjson.Get(body, HTTPRequestData)
	if !raw.IsArray() {
		return false
	}
	arr := raw.Array()
	if len(arr) == 0 {
		return false
	}
	first := arr[0]
	if first.Type == gjson.String {
		return false
	}
	if !first.IsArray() {
		return false
	}
	inner := first.Array()
	if len(inner) == 0 {
		return false
	}
	firstInner := inner[0]
	return firstInner.IsArray() || firstInner.Type == gjson.String
}

func convertEmbListQueries2Placeholder(body string, elemType schemapb.DataType, dim int64) (*commonpb.PlaceholderValue, error) {
	raw := gjson.Get(body, HTTPRequestData)
	if !raw.IsArray() {
		return nil, merr.WrapErrParameterInvalidMsg("search data must be an array of embedding lists")
	}
	queries := raw.Array()
	if len(queries) == 0 {
		return nil, merr.WrapErrParameterInvalidMsg("search data is empty")
	}

	placeholderType, err := embListPlaceholderType(elemType)
	if err != nil {
		return nil, err
	}

	values := make([][]byte, 0, len(queries))
	for qIdx, q := range queries {
		if !q.IsArray() {
			return nil, merr.WrapErrParameterInvalidMsg(
				"search data[%d] must be an array of vectors", qIdx)
		}
		vecs := q.Array()
		if len(vecs) == 0 {
			return nil, merr.WrapErrParameterInvalidMsg(
				"search data[%d] embedding list is empty", qIdx)
		}
		buf, err := encodeEmbListQuery(vecs, elemType, dim, qIdx)
		if err != nil {
			return nil, err
		}
		values = append(values, buf)
	}
	return &commonpb.PlaceholderValue{
		Tag:    "$0",
		Type:   placeholderType,
		Values: values,
	}, nil
}

func embListPlaceholderType(elemType schemapb.DataType) (commonpb.PlaceholderType, error) {
	switch elemType {
	case schemapb.DataType_FloatVector:
		return commonpb.PlaceholderType_EmbListFloatVector, nil
	case schemapb.DataType_Float16Vector:
		return commonpb.PlaceholderType_EmbListFloat16Vector, nil
	case schemapb.DataType_BFloat16Vector:
		return commonpb.PlaceholderType_EmbListBFloat16Vector, nil
	case schemapb.DataType_BinaryVector:
		return commonpb.PlaceholderType_EmbListBinaryVector, nil
	case schemapb.DataType_Int8Vector:
		return commonpb.PlaceholderType_EmbListInt8Vector, nil
	default:
		return 0, merr.WrapErrParameterInvalidMsg(
			"unsupported embedding list element type %s", elemType)
	}
}

func encodeEmbListQuery(vecs []gjson.Result, elemType schemapb.DataType, dim int64, qIdx int) ([]byte, error) {
	switch elemType {
	case schemapb.DataType_FloatVector:
		buf := make([]byte, 0, int(dim*4)*len(vecs))
		for vIdx, v := range vecs {
			if !v.IsArray() {
				return nil, merr.WrapErrParameterInvalidMsg(
					"search data[%d][%d] must be a float vector array", qIdx, vIdx)
			}
			var row []float32
			if err := json.Unmarshal([]byte(v.Raw), &row); err != nil {
				return nil, merr.WrapErrParameterInvalidMsg(
					"search data[%d][%d] parse fail: %s", qIdx, vIdx, err.Error())
			}
			if int64(len(row)) != dim {
				return nil, merr.WrapErrParameterInvalidMsg(
					"search data[%d][%d] dim mismatch: expect %d got %d", qIdx, vIdx, dim, len(row))
			}
			buf = append(buf, typeutil.Float32ArrayToBytes(row)...)
		}
		return buf, nil
	case schemapb.DataType_Float16Vector, schemapb.DataType_BFloat16Vector:
		isFloat16 := elemType == schemapb.DataType_Float16Vector
		bytesPerVec := dim * 2
		buf := make([]byte, 0, int(bytesPerVec)*len(vecs))
		for vIdx, v := range vecs {
			b, err := decodeByteVectorElement(v, dim, bytesPerVec, isFloat16)
			if err != nil {
				return nil, merr.WrapErrParameterInvalidMsg(
					"search data[%d][%d]: %s", qIdx, vIdx, err.Error())
			}
			buf = append(buf, b...)
		}
		return buf, nil
	case schemapb.DataType_BinaryVector:
		bytesPerVec := dim / 8
		buf := make([]byte, 0, int(bytesPerVec)*len(vecs))
		for vIdx, v := range vecs {
			if v.Type != gjson.String {
				return nil, merr.WrapErrParameterInvalidMsg(
					"search data[%d][%d] binary vector must be base64-encoded string", qIdx, vIdx)
			}
			var row []byte
			if err := json.Unmarshal([]byte(v.Raw), &row); err != nil {
				return nil, merr.WrapErrParameterInvalidMsg(
					"search data[%d][%d] parse fail: %s", qIdx, vIdx, err.Error())
			}
			if int64(len(row)) != bytesPerVec {
				return nil, merr.WrapErrParameterInvalidMsg(
					"search data[%d][%d] byte-length mismatch: expect %d got %d", qIdx, vIdx, bytesPerVec, len(row))
			}
			buf = append(buf, row...)
		}
		return buf, nil
	case schemapb.DataType_Int8Vector:
		buf := make([]byte, 0, int(dim)*len(vecs))
		for vIdx, v := range vecs {
			if !v.IsArray() {
				return nil, merr.WrapErrParameterInvalidMsg(
					"search data[%d][%d] must be an int8 vector array", qIdx, vIdx)
			}
			var row []int8
			if err := json.Unmarshal([]byte(v.Raw), &row); err != nil {
				return nil, merr.WrapErrParameterInvalidMsg(
					"search data[%d][%d] parse fail: %s", qIdx, vIdx, err.Error())
			}
			if int64(len(row)) != dim {
				return nil, merr.WrapErrParameterInvalidMsg(
					"search data[%d][%d] dim mismatch: expect %d got %d", qIdx, vIdx, dim, len(row))
			}
			buf = append(buf, typeutil.Int8ArrayToBytes(row)...)
		}
		return buf, nil
	default:
		return nil, merr.WrapErrParameterInvalidMsg(
			"unsupported embedding list element type %s", elemType)
	}
}

func buildStructArrayFieldData(structSchema *schemapb.StructArrayFieldSchema, perRow []structArrayRow) (*schemapb.FieldData, error) {
	if len(perRow) == 0 {
		return nil, merr.WrapErrParameterInvalidMsg("struct array field %s has no rows", structSchema.GetName())
	}
	subs := structSchema.GetFields()
	subFieldData := make([]*schemapb.FieldData, 0, len(subs))
	for _, sub := range subs {
		short := subShortName(sub)
		switch sub.GetDataType() {
		case schemapb.DataType_Array:
			arrayArray := &schemapb.ArrayArray{
				Data:        make([]*schemapb.ScalarField, 0, len(perRow)),
				ElementType: sub.GetElementType(),
			}
			for rowIdx, row := range perRow {
				val, ok := row[short]
				if !ok {
					return nil, merr.WrapErrParameterInvalidMsg("struct %s row %d missing sub-field %s",
						structSchema.GetName(), rowIdx, short)
				}
				scalar, ok := val.(*schemapb.ScalarField)
				if !ok {
					return nil, merr.WrapErrParameterInvalidMsg("struct %s sub-field %s row %d: unexpected payload type %T",
						structSchema.GetName(), short, rowIdx, val)
				}
				arrayArray.Data = append(arrayArray.Data, scalar)
			}
			subFieldData = append(subFieldData, &schemapb.FieldData{
				Type:      schemapb.DataType_Array,
				FieldName: short,
				FieldId:   sub.GetFieldID(),
				Field: &schemapb.FieldData_Scalars{
					Scalars: &schemapb.ScalarField{
						Data: &schemapb.ScalarField_ArrayData{ArrayData: arrayArray},
					},
				},
			})
		case schemapb.DataType_ArrayOfVector:
			dim, err := getDim(sub)
			if err != nil {
				return nil, err
			}
			vecArray := &schemapb.VectorArray{
				ElementType: sub.GetElementType(),
				Dim:         dim,
				Data:        make([]*schemapb.VectorField, 0, len(perRow)),
			}
			for rowIdx, row := range perRow {
				val, ok := row[short]
				if !ok {
					return nil, merr.WrapErrParameterInvalidMsg("struct %s row %d missing sub-field %s",
						structSchema.GetName(), rowIdx, short)
				}
				vf, ok := val.(*schemapb.VectorField)
				if !ok {
					return nil, merr.WrapErrParameterInvalidMsg("struct %s sub-field %s row %d: unexpected payload type %T",
						structSchema.GetName(), short, rowIdx, val)
				}
				vecArray.Data = append(vecArray.Data, vf)
			}
			subFieldData = append(subFieldData, &schemapb.FieldData{
				Type:      schemapb.DataType_ArrayOfVector,
				FieldName: short,
				FieldId:   sub.GetFieldID(),
				Field: &schemapb.FieldData_Vectors{
					Vectors: &schemapb.VectorField{
						Dim: dim,
						Data: &schemapb.VectorField_VectorArray{
							VectorArray: vecArray,
						},
					},
				},
			})
		default:
			return nil, merr.WrapErrParameterInvalidMsg("unsupported struct sub-field data type: %s", sub.GetDataType())
		}
	}
	return &schemapb.FieldData{
		Type:      schemapb.DataType_ArrayOfStruct,
		FieldName: structSchema.GetName(),
		FieldId:   structSchema.GetFieldID(),
		Field: &schemapb.FieldData_StructArrays{
			StructArrays: &schemapb.StructArrayField{Fields: subFieldData},
		},
	}, nil
}

func extractStructArrayRow(fd *schemapb.FieldData, rowIdx int, schema *schemapb.CollectionSchema) ([]map[string]interface{}, error) {
	subs := fd.GetStructArrays().GetFields()
	if len(subs) == 0 {
		return []map[string]interface{}{}, nil
	}

	subDims, err := structArraySubDims(fd.GetFieldName(), schema)
	if err != nil {
		return nil, err
	}

	elemCount, err := structSubElemCount(subs[0], rowIdx, subDims)
	if err != nil {
		return nil, err
	}
	out := make([]map[string]interface{}, elemCount)
	for i := 0; i < elemCount; i++ {
		out[i] = make(map[string]interface{}, len(subs))
	}
	for _, sub := range subs {
		short := structFieldShortName(sub.GetFieldName())
		switch sub.GetType() {
		case schemapb.DataType_Array:
			rowData := sub.GetScalars().GetArrayData().GetData()
			if rowIdx >= len(rowData) {
				return nil, merr.WrapErrParameterInvalidMsg("struct sub-field %s missing row %d", short, rowIdx)
			}
			values := scalarArrayToInterfaces(rowData[rowIdx])
			if len(values) != elemCount {
				return nil, merr.WrapErrParameterInvalidMsg("struct sub-field %s element count mismatch: expect %d got %d",
					short, elemCount, len(values))
			}
			for i, v := range values {
				out[i][short] = v
			}
		case schemapb.DataType_ArrayOfVector:
			va := sub.GetVectors().GetVectorArray()
			if va == nil {
				return nil, merr.WrapErrParameterInvalidMsg("struct sub-field %s has no vector array", short)
			}
			if rowIdx >= len(va.GetData()) {
				return nil, merr.WrapErrParameterInvalidMsg("struct sub-field %s missing row %d", short, rowIdx)
			}
			dim, ok := subDims[short]
			if !ok || dim <= 0 {
				return nil, merr.WrapErrParameterInvalidMsg("schema missing dim for struct sub-field %s", short)
			}
			values, err := vectorFieldToInterfaces(va.GetData()[rowIdx], va.GetElementType(), dim)
			if err != nil {
				return nil, err
			}
			if len(values) != elemCount {
				return nil, merr.WrapErrParameterInvalidMsg("struct sub-field %s vector element count mismatch: expect %d got %d",
					short, elemCount, len(values))
			}
			for i, v := range values {
				out[i][short] = v
			}
		default:
			return nil, merr.WrapErrParameterInvalidMsg("unsupported struct sub-field type %s", sub.GetType())
		}
	}
	return out, nil
}

func structArraySubDims(fieldName string, schema *schemapb.CollectionSchema) (map[string]int64, error) {
	subDims := make(map[string]int64)
	for _, sf := range schema.GetStructArrayFields() {
		if sf.GetName() != fieldName {
			continue
		}
		for _, sub := range sf.GetFields() {
			if sub.GetDataType() != schemapb.DataType_ArrayOfVector {
				continue
			}
			dim, err := getDim(sub)
			if err != nil {
				return nil, merr.WrapErrParameterInvalidErr(err, "schema sub-field %s has no dim", sub.GetName())
			}
			subDims[subShortName(sub)] = dim
		}
		break
	}
	return subDims, nil
}

func structSubElemCount(sub *schemapb.FieldData, rowIdx int, subDims map[string]int64) (int, error) {
	switch sub.GetType() {
	case schemapb.DataType_Array:
		rowData := sub.GetScalars().GetArrayData().GetData()
		if rowIdx >= len(rowData) {
			return 0, merr.WrapErrParameterInvalidMsg("struct sub-field %s row %d out of range", sub.GetFieldName(), rowIdx)
		}
		return len(scalarArrayToInterfaces(rowData[rowIdx])), nil
	case schemapb.DataType_ArrayOfVector:
		va := sub.GetVectors().GetVectorArray()
		if va == nil || rowIdx >= len(va.GetData()) {
			return 0, merr.WrapErrParameterInvalidMsg("struct sub-field %s row %d out of range", sub.GetFieldName(), rowIdx)
		}
		short := structFieldShortName(sub.GetFieldName())
		dim, ok := subDims[short]
		if !ok || dim <= 0 {
			return 0, merr.WrapErrParameterInvalidMsg("schema missing dim for struct sub-field %s", short)
		}
		return vectorFieldElemCount(va.GetData()[rowIdx], va.GetElementType(), dim)
	default:
		return 0, merr.WrapErrParameterInvalidMsg("unsupported struct sub-field type %s", sub.GetType())
	}
}

func scalarArrayToInterfaces(sf *schemapb.ScalarField) []interface{} {
	switch sf.GetData().(type) {
	case *schemapb.ScalarField_BoolData:
		src := sf.GetBoolData().GetData()
		out := make([]interface{}, len(src))
		for i, v := range src {
			out[i] = v
		}
		return out
	case *schemapb.ScalarField_IntData:
		src := sf.GetIntData().GetData()
		out := make([]interface{}, len(src))
		for i, v := range src {
			out[i] = v
		}
		return out
	case *schemapb.ScalarField_LongData:
		src := sf.GetLongData().GetData()
		out := make([]interface{}, len(src))
		for i, v := range src {
			out[i] = v
		}
		return out
	case *schemapb.ScalarField_FloatData:
		src := sf.GetFloatData().GetData()
		out := make([]interface{}, len(src))
		for i, v := range src {
			out[i] = v
		}
		return out
	case *schemapb.ScalarField_DoubleData:
		src := sf.GetDoubleData().GetData()
		out := make([]interface{}, len(src))
		for i, v := range src {
			out[i] = v
		}
		return out
	case *schemapb.ScalarField_StringData:
		src := sf.GetStringData().GetData()
		out := make([]interface{}, len(src))
		for i, v := range src {
			out[i] = v
		}
		return out
	default:
		return nil
	}
}

func vectorFieldElemCount(vf *schemapb.VectorField, elemType schemapb.DataType, dim int64) (int, error) {
	if dim <= 0 {
		return 0, merr.WrapErrParameterInvalidMsg("invalid dim %d", dim)
	}
	switch elemType {
	case schemapb.DataType_FloatVector:
		return len(vf.GetFloatVector().GetData()) / int(dim), nil
	case schemapb.DataType_Float16Vector:
		return len(vf.GetFloat16Vector()) / int(dim*2), nil
	case schemapb.DataType_BFloat16Vector:
		return len(vf.GetBfloat16Vector()) / int(dim*2), nil
	case schemapb.DataType_BinaryVector:
		return len(vf.GetBinaryVector()) / int(dim/8), nil
	case schemapb.DataType_Int8Vector:
		return len(vf.GetInt8Vector()) / int(dim), nil
	default:
		return 0, merr.WrapErrParameterInvalidMsg("unsupported vector element type %s", elemType)
	}
}

func vectorFieldToInterfaces(vf *schemapb.VectorField, elemType schemapb.DataType, dim int64) ([]interface{}, error) {
	if dim <= 0 {
		return nil, merr.WrapErrParameterInvalidMsg("invalid dim %d", dim)
	}
	switch elemType {
	case schemapb.DataType_FloatVector:
		buf := vf.GetFloatVector().GetData()
		count := len(buf) / int(dim)
		out := make([]interface{}, count)
		for i := 0; i < count; i++ {
			out[i] = buf[i*int(dim) : (i+1)*int(dim)]
		}
		return out, nil
	case schemapb.DataType_Float16Vector:
		buf := vf.GetFloat16Vector()
		step := int(dim * 2)
		count := len(buf) / step
		out := make([]interface{}, count)
		for i := 0; i < count; i++ {
			out[i] = base64.StdEncoding.EncodeToString(buf[i*step : (i+1)*step])
		}
		return out, nil
	case schemapb.DataType_BFloat16Vector:
		buf := vf.GetBfloat16Vector()
		step := int(dim * 2)
		count := len(buf) / step
		out := make([]interface{}, count)
		for i := 0; i < count; i++ {
			out[i] = base64.StdEncoding.EncodeToString(buf[i*step : (i+1)*step])
		}
		return out, nil
	case schemapb.DataType_BinaryVector:
		buf := vf.GetBinaryVector()
		step := int(dim / 8)
		count := len(buf) / step
		out := make([]interface{}, count)
		for i := 0; i < count; i++ {
			out[i] = base64.StdEncoding.EncodeToString(buf[i*step : (i+1)*step])
		}
		return out, nil
	case schemapb.DataType_Int8Vector:
		buf := vf.GetInt8Vector()
		step := int(dim)
		count := len(buf) / step
		out := make([]interface{}, count)
		for i := 0; i < count; i++ {
			seg := buf[i*step : (i+1)*step]
			row := make([]int8, step)
			for j, b := range seg {
				row[j] = int8(b)
			}
			out[i] = row
		}
		return out, nil
	default:
		return nil, merr.WrapErrParameterInvalidMsg("unsupported vector element type %s", elemType)
	}
}

func convertFloatVectorToArray(vector [][]float32, dim int64) ([]float32, error) {
	floatArray := make([]float32, 0)
	for _, arr := range vector {
		if int64(len(arr)) != dim {
			return nil, merr.WrapErrParameterInvalidMsg("[]float32 size %d doesn't equal to vector dimension %d of %s",
				len(arr), dim, schemapb.DataType_name[int32(schemapb.DataType_FloatVector)])
		}
		for i := int64(0); i < dim; i++ {
			floatArray = append(floatArray, arr[i])
		}
	}
	return floatArray, nil
}

func convertBinaryVectorToArray(vector [][]byte, dim int64, dataType schemapb.DataType) ([]byte, error) {
	var bytesLen int64
	switch dataType {
	case schemapb.DataType_BinaryVector:
		bytesLen = dim / 8
	case schemapb.DataType_Float16Vector:
		bytesLen = dim * 2
	case schemapb.DataType_BFloat16Vector:
		bytesLen = dim * 2
	}
	binaryArray := make([]byte, 0, len(vector)*int(bytesLen))
	for _, arr := range vector {
		if int64(len(arr)) != bytesLen {
			return nil, merr.WrapErrParameterInvalidMsg("[]byte size %d doesn't equal to vector dimension %d of %s",
				len(arr), dim, schemapb.DataType_name[int32(dataType)])
		}
		for i := int64(0); i < bytesLen; i++ {
			binaryArray = append(binaryArray, arr[i])
		}
	}
	return binaryArray, nil
}

func convertInt8VectorToArray(vector [][]int8, dim int64) ([]byte, error) {
	byteArray := make([]byte, 0)
	for _, arr := range vector {
		if int64(len(arr)) != dim {
			return nil, merr.WrapErrParameterInvalidMsg("[]int8 size %d doesn't equal to vector dimension %d of %s",
				len(arr), dim, schemapb.DataType_name[int32(schemapb.DataType_Int8Vector)])
		}
		for i := int64(0); i < dim; i++ {
			byteArray = append(byteArray, byte(arr[i]))
		}
	}
	return byteArray, nil
}

type fieldCandi struct {
	name    string
	v       reflect.Value
	options map[string]string
}

func reflectValueCandi(v reflect.Value) (map[string]fieldCandi, error) {
	if v.Kind() == reflect.Ptr {
		v = v.Elem()
	}

	result := make(map[string]fieldCandi)
	switch v.Kind() {
	case reflect.Map: // map[string]interface{}
		iter := v.MapRange()
		for iter.Next() {
			key := iter.Key().String()
			result[key] = fieldCandi{
				name: key,
				v:    iter.Value(),
			}
		}
		return result, nil
	default:
		return nil, merr.WrapErrParameterInvalidMsg("unsupport row type: %s", v.Kind().String())
	}
}

func convertToIntArray(dataType schemapb.DataType, arr interface{}) []int32 {
	var res []int32
	switch dataType {
	case schemapb.DataType_Int8:
		for _, num := range arr.([]int8) {
			res = append(res, int32(num))
		}
	case schemapb.DataType_Int16:
		for _, num := range arr.([]int16) {
			res = append(res, int32(num))
		}
	}
	return res
}

func anyToColumns(rows []map[string]interface{}, validDataMap map[string][]bool, sch *schemapb.CollectionSchema, inInsert bool, partialUpdate bool) ([]*schemapb.FieldData, error) {
	rowsLen := len(rows)
	if rowsLen == 0 {
		return []*schemapb.FieldData{}, merr.WrapErrParameterInvalidMsg("no row need to be convert to columns")
	}

	isDynamic := sch.EnableDynamicField
	allowInsertAutoID, _ := common.IsAllowInsertAutoID(sch.GetProperties()...)
	isAutoIDPK := false
	pkFieldName := ""

	nameColumns := make(map[string]interface{})
	nameDims := make(map[string]int64)
	fieldData := make(map[string]*schemapb.FieldData)

	// Pre-compute the set of field names present across all rows,
	// so we can skip absent function output fields with a map lookup instead of scanning rows.
	presentFieldNames := make(map[string]struct{})
	for _, row := range rows {
		for name := range row {
			presentFieldNames[name] = struct{}{}
		}
	}

	for _, field := range sch.Fields {
		if field.IsPrimaryKey {
			pkFieldName = field.Name
			if field.AutoID {
				isAutoIDPK = true
			}
		}
		if (field.IsPrimaryKey && field.AutoID && inInsert && !allowInsertAutoID) || field.IsDynamic {
			continue
		}
		// skip function output field if no row provides data for it
		if field.GetIsFunctionOutput() {
			if _, ok := presentFieldNames[field.Name]; !ok {
				continue
			}
		}
		var data interface{}
		switch field.DataType {
		case schemapb.DataType_Bool:
			data = make([]bool, 0, rowsLen)
		case schemapb.DataType_Int8:
			data = make([]int8, 0, rowsLen)
		case schemapb.DataType_Int16:
			data = make([]int16, 0, rowsLen)
		case schemapb.DataType_Int32:
			data = make([]int32, 0, rowsLen)
		case schemapb.DataType_Int64:
			data = make([]int64, 0, rowsLen)
		case schemapb.DataType_Float:
			data = make([]float32, 0, rowsLen)
		case schemapb.DataType_Double:
			data = make([]float64, 0, rowsLen)
		case schemapb.DataType_Timestamptz:
			data = make([]string, 0, rowsLen)
		case schemapb.DataType_String:
			data = make([]string, 0, rowsLen)
		case schemapb.DataType_VarChar:
			data = make([]string, 0, rowsLen)
		case schemapb.DataType_Array:
			data = make([]*schemapb.ScalarField, 0, rowsLen)
		case schemapb.DataType_JSON:
			data = make([][]byte, 0, rowsLen)
		case schemapb.DataType_Geometry:
			data = make([]string, 0, rowsLen)
		case schemapb.DataType_FloatVector:
			data = make([][]float32, 0, rowsLen)
			dim, _ := getDim(field)
			nameDims[field.Name] = dim
		case schemapb.DataType_BinaryVector:
			data = make([][]byte, 0, rowsLen)
			dim, _ := getDim(field)
			nameDims[field.Name] = dim
		case schemapb.DataType_Float16Vector:
			data = make([][]byte, 0, rowsLen)
			dim, _ := getDim(field)
			nameDims[field.Name] = dim
		case schemapb.DataType_BFloat16Vector:
			data = make([][]byte, 0, rowsLen)
			dim, _ := getDim(field)
			nameDims[field.Name] = dim
		case schemapb.DataType_SparseFloatVector:
			data = make([][]byte, 0, rowsLen)
			nameDims[field.Name] = int64(0)
		case schemapb.DataType_Int8Vector:
			data = make([][]int8, 0, rowsLen)
			dim, _ := getDim(field)
			nameDims[field.Name] = dim
		default:
			return nil, merr.WrapErrParameterInvalidMsg("the type(%v) of field(%v) is not supported, use other sdk please", field.DataType, field.Name)
		}
		nameColumns[field.Name] = data
		fieldData[field.Name] = &schemapb.FieldData{
			Type:      field.DataType,
			FieldName: field.Name,
			FieldId:   field.FieldID,
			IsDynamic: field.IsDynamic,
		}
	}
	if len(nameDims) == 0 && len(sch.Functions) == 0 && !partialUpdate {
		return nil, merr.WrapErrParameterInvalidMsg("collection: %s has no vector field or functions", sch.Name)
	}

	dynamicCol := make([][]byte, 0, rowsLen)
	fieldLen := make(map[string]int)

	for _, row := range rows {
		// collection schema name need not be same, since receiver could have other names
		v := reflect.ValueOf(row)
		set, err := reflectValueCandi(v)
		if err != nil {
			return nil, err
		}
		for idx, field := range sch.Fields {
			if field.IsDynamic {
				continue
			}
			candi, ok := set[field.Name]
			if field.IsPrimaryKey && field.AutoID && inInsert {
				if !ok {
					continue
				}
				if !allowInsertAutoID {
					return nil, merr.WrapErrParameterInvalidMsg("no need to pass pk field(%s) when autoid==true in insert", field.Name)
				}
			}
			if (field.Nullable || field.DefaultValue != nil) && !ok {
				continue
			}
			if field.GetIsFunctionOutput() {
				if _, allocated := nameColumns[field.Name]; !allocated {
					continue
				}
			}
			if !ok {
				if partialUpdate {
					continue
				}
				return nil, merr.WrapErrParameterInvalidMsg("row %d does not has field %s", idx, field.Name)
			}
			fieldLen[field.Name] += 1
			switch field.DataType {
			case schemapb.DataType_Bool:
				nameColumns[field.Name] = append(nameColumns[field.Name].([]bool), candi.v.Interface().(bool))
			case schemapb.DataType_Int8:
				nameColumns[field.Name] = append(nameColumns[field.Name].([]int8), candi.v.Interface().(int8))
			case schemapb.DataType_Int16:
				nameColumns[field.Name] = append(nameColumns[field.Name].([]int16), candi.v.Interface().(int16))
			case schemapb.DataType_Int32:
				nameColumns[field.Name] = append(nameColumns[field.Name].([]int32), candi.v.Interface().(int32))
			case schemapb.DataType_Int64:
				nameColumns[field.Name] = append(nameColumns[field.Name].([]int64), candi.v.Interface().(int64))
			case schemapb.DataType_Float:
				nameColumns[field.Name] = append(nameColumns[field.Name].([]float32), candi.v.Interface().(float32))
			case schemapb.DataType_Timestamptz:
				nameColumns[field.Name] = append(nameColumns[field.Name].([]string), candi.v.Interface().(string))
			case schemapb.DataType_Double:
				nameColumns[field.Name] = append(nameColumns[field.Name].([]float64), candi.v.Interface().(float64))
			case schemapb.DataType_String:
				nameColumns[field.Name] = append(nameColumns[field.Name].([]string), candi.v.Interface().(string))
			case schemapb.DataType_VarChar:
				nameColumns[field.Name] = append(nameColumns[field.Name].([]string), candi.v.Interface().(string))
			case schemapb.DataType_Array:
				nameColumns[field.Name] = append(nameColumns[field.Name].([]*schemapb.ScalarField), candi.v.Interface().(*schemapb.ScalarField))
			case schemapb.DataType_JSON:
				nameColumns[field.Name] = append(nameColumns[field.Name].([][]byte), candi.v.Interface().([]byte))
			case schemapb.DataType_Geometry:
				nameColumns[field.Name] = append(nameColumns[field.Name].([]string), candi.v.Interface().(string))
			case schemapb.DataType_FloatVector:
				nameColumns[field.Name] = append(nameColumns[field.Name].([][]float32), candi.v.Interface().([]float32))
			case schemapb.DataType_BinaryVector:
				nameColumns[field.Name] = append(nameColumns[field.Name].([][]byte), candi.v.Interface().([]byte))
			case schemapb.DataType_Float16Vector:
				switch candi.v.Interface().(type) {
				case []byte:
					nameColumns[field.Name] = append(nameColumns[field.Name].([][]byte), candi.v.Interface().([]byte))
				case []float32:
					vec := typeutil.Float32ArrayToFloat16Bytes(candi.v.Interface().([]float32))
					nameColumns[field.Name] = append(nameColumns[field.Name].([][]byte), vec)
				default:
					return nil, merr.WrapErrParameterInvalidMsg("invalid type(%v) of field(%v) ", field.DataType, field.Name)
				}
			case schemapb.DataType_BFloat16Vector:
				switch candi.v.Interface().(type) {
				case []byte:
					nameColumns[field.Name] = append(nameColumns[field.Name].([][]byte), candi.v.Interface().([]byte))
				case []float32:
					vec := typeutil.Float32ArrayToBFloat16Bytes(candi.v.Interface().([]float32))
					nameColumns[field.Name] = append(nameColumns[field.Name].([][]byte), vec)
				default:
					return nil, merr.WrapErrParameterInvalidMsg("invalid type(%v) of field(%v) ", field.DataType, field.Name)
				}
			case schemapb.DataType_SparseFloatVector:
				content := candi.v.Interface().([]byte)
				rowSparseDim := typeutil.SparseFloatRowDim(content)
				if rowSparseDim > nameDims[field.Name] {
					nameDims[field.Name] = rowSparseDim
				}
				nameColumns[field.Name] = append(nameColumns[field.Name].([][]byte), content)
			case schemapb.DataType_Int8Vector:
				nameColumns[field.Name] = append(nameColumns[field.Name].([][]int8), candi.v.Interface().([]int8))
			default:
				return nil, merr.WrapErrParameterInvalidMsg("the type(%v) of field(%v) is not supported, use other sdk please", field.DataType, field.Name)
			}

			delete(set, field.Name)
		}
		for _, structField := range sch.GetStructArrayFields() {
			delete(set, structField.GetName())
		}
		// if is not dynamic, but pass more field, will throw err in /internal/distributed/proxy/httpserver/utils.go@checkAndSetData
		if isDynamic {
			m := make(map[string]interface{})
			for name, candi := range set {
				m[name] = candi.v.Interface()
			}
			bs, err := json.Marshal(m)
			if err != nil {
				return nil, merr.WrapErrParameterInvalidErr(err, "failed to marshal dynamic field")
			}
			dynamicCol = append(dynamicCol, bs)
		}
	}
	columns := make([]*schemapb.FieldData, 0, len(nameColumns))
	for name, column := range nameColumns {
		validData, hasValidData := validDataMap[name]
		if fieldLen[name] == 0 && name == pkFieldName && isAutoIDPK {
			continue
		}
		if fieldLen[name] == 0 && partialUpdate {
			if hasValidData {
				if len(validData) != rowsLen {
					mlog.Info(context.TODO(), "field len is not equal to rows len",
						mlog.String("fieldName", name),
						mlog.Int("fieldLen", len(validData)),
						mlog.Int("rowsLen", rowsLen))
					return nil, merr.WrapErrParameterInvalidMsg("column %s has length %d, expected %d", name, len(validData), rowsLen)
				}
			} else {
				mlog.Info(context.TODO(), "skip empty field for partial update",
					mlog.String("fieldName", name))
				continue
			}
		}
		if fieldLen[name] != rowsLen && partialUpdate && (!hasValidData || len(validData) != rowsLen) {
			// for partial update, if try to update different field in different rows, return error
			mlog.Info(context.TODO(), "field len is not equal to rows len",
				mlog.String("fieldName", name),
				mlog.Int("fieldLen", fieldLen[name]),
				mlog.Int("rowsLen", rowsLen))
			return nil, merr.WrapErrParameterInvalidMsg("column %s has length %d, expected %d", name, fieldLen[name], rowsLen)
		}

		colData := fieldData[name]
		switch colData.Type {
		case schemapb.DataType_Bool:
			colData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_BoolData{
						BoolData: &schemapb.BoolArray{
							Data: column.([]bool),
						},
					},
				},
			}
		case schemapb.DataType_Int8:
			colData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{
							Data: convertToIntArray(colData.Type, column),
						},
					},
				},
			}
		case schemapb.DataType_Int16:
			colData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{
							Data: convertToIntArray(colData.Type, column),
						},
					},
				},
			}
		case schemapb.DataType_Int32:
			colData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_IntData{
						IntData: &schemapb.IntArray{
							Data: column.([]int32),
						},
					},
				},
			}
		case schemapb.DataType_Int64:
			colData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_LongData{
						LongData: &schemapb.LongArray{
							Data: column.([]int64),
						},
					},
				},
			}
		case schemapb.DataType_Float:
			colData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_FloatData{
						FloatData: &schemapb.FloatArray{
							Data: column.([]float32),
						},
					},
				},
			}
		case schemapb.DataType_Double:
			colData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_DoubleData{
						DoubleData: &schemapb.DoubleArray{
							Data: column.([]float64),
						},
					},
				},
			}
		case schemapb.DataType_Timestamptz:
			colData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_StringData{
						StringData: &schemapb.StringArray{
							Data: column.([]string),
						},
					},
				},
			}
		case schemapb.DataType_String:
			colData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_StringData{
						StringData: &schemapb.StringArray{
							Data: column.([]string),
						},
					},
				},
			}
		case schemapb.DataType_VarChar:
			colData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_StringData{
						StringData: &schemapb.StringArray{
							Data: column.([]string),
						},
					},
				},
			}
		case schemapb.DataType_Array:
			colData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_ArrayData{
						ArrayData: &schemapb.ArrayArray{
							Data: column.([]*schemapb.ScalarField),
						},
					},
				},
			}
		case schemapb.DataType_JSON:
			colData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_JsonData{
						JsonData: &schemapb.JSONArray{
							Data: column.([][]byte),
						},
					},
				},
			}
		case schemapb.DataType_Geometry:
			colData.Field = &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_GeometryWktData{
						GeometryWktData: &schemapb.GeometryWktArray{
							Data: column.([]string),
						},
					},
				},
			}
		case schemapb.DataType_FloatVector:
			dim := nameDims[name]
			arr, err := convertFloatVectorToArray(column.([][]float32), dim)
			if err != nil {
				return nil, err
			}
			colData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_FloatVector{
						FloatVector: &schemapb.FloatArray{
							Data: arr,
						},
					},
				},
			}
		case schemapb.DataType_BinaryVector:
			dim := nameDims[name]
			arr, err := convertBinaryVectorToArray(column.([][]byte), dim, colData.Type)
			if err != nil {
				return nil, err
			}
			colData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_BinaryVector{
						BinaryVector: arr,
					},
				},
			}
		case schemapb.DataType_Float16Vector:
			dim := nameDims[name]
			arr, err := convertBinaryVectorToArray(column.([][]byte), dim, colData.Type)
			if err != nil {
				return nil, err
			}
			colData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Float16Vector{
						Float16Vector: arr,
					},
				},
			}
		case schemapb.DataType_BFloat16Vector:
			dim := nameDims[name]
			arr, err := convertBinaryVectorToArray(column.([][]byte), dim, colData.Type)
			if err != nil {
				return nil, err
			}
			colData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Bfloat16Vector{
						Bfloat16Vector: arr,
					},
				},
			}
		case schemapb.DataType_SparseFloatVector:
			colData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: nameDims[name],
					Data: &schemapb.VectorField_SparseFloatVector{
						SparseFloatVector: &schemapb.SparseFloatArray{
							Dim:      nameDims[name],
							Contents: column.([][]byte),
						},
					},
				},
			}
		case schemapb.DataType_Int8Vector:
			dim := nameDims[name]
			arr, err := convertInt8VectorToArray(column.([][]int8), dim)
			if err != nil {
				return nil, err
			}
			colData.Field = &schemapb.FieldData_Vectors{
				Vectors: &schemapb.VectorField{
					Dim: dim,
					Data: &schemapb.VectorField_Int8Vector{
						Int8Vector: arr,
					},
				},
			}
		default:
			return nil, merr.WrapErrParameterInvalidMsg("the type(%v) of field(%v) is not supported, use other sdk please", colData.Type, name)
		}
		colData.ValidData = validDataMap[name]
		columns = append(columns, colData)
	}
	if isDynamic {
		columns = append(columns, &schemapb.FieldData{
			Type:      schemapb.DataType_JSON,
			FieldName: "",
			Field: &schemapb.FieldData_Scalars{
				Scalars: &schemapb.ScalarField{
					Data: &schemapb.ScalarField_JsonData{
						JsonData: &schemapb.JSONArray{
							Data: dynamicCol,
						},
					},
				},
			},
			IsDynamic: true,
		})
	}
	for _, structField := range sch.GetStructArrayFields() {
		perRow := make([]structArrayRow, 0, rowsLen)
		for rowIdx, row := range rows {
			val, ok := row[structField.GetName()]
			if !ok {
				if partialUpdate {
					continue
				}
				return nil, merr.WrapErrParameterInvalidMsg("row %d does not has struct field %s", rowIdx, structField.GetName())
			}
			sr, ok := val.(structArrayRow)
			if !ok {
				return nil, merr.WrapErrParameterInvalidMsg("row %d struct field %s has unexpected payload type %T",
					rowIdx, structField.GetName(), val)
			}
			perRow = append(perRow, sr)
		}
		if len(perRow) == 0 {
			continue
		}
		structFieldData, err := buildStructArrayFieldData(structField, perRow)
		if err != nil {
			return nil, err
		}
		columns = append(columns, structFieldData)
	}
	return columns, nil
}

func serializeFloatVectors(vectorStr string, dataType schemapb.DataType, dimension, bytesLen int64, fpArrayToBytesFunc func([]float32) []byte) ([][]byte, error) {
	var fp32Values [][]float32
	err := json.Unmarshal([]byte(vectorStr), &fp32Values)
	if err != nil {
		return nil, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(dataType)], vectorStr, err.Error())
	}
	values := make([][]byte, 0, len(fp32Values))
	for _, vectorArray := range fp32Values {
		if int64(len(vectorArray)) != dimension {
			return nil, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(dataType)], vectorStr,
				fmt.Sprintf("dimension: %d, but length of []float: %d", dimension, len(vectorArray)))
		}
		vectorBytes := fpArrayToBytesFunc(vectorArray)
		values = append(values, vectorBytes)
	}
	return values, nil
}

func serializeByteVectors(vectorStr string, dataType schemapb.DataType, dimension, bytesLen int64) ([][]byte, error) {
	values := make([][]byte, 0)
	err := json.Unmarshal([]byte(vectorStr), &values)
	if err != nil {
		return nil, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(dataType)], vectorStr, err.Error())
	}
	for _, vectorArray := range values {
		if int64(len(vectorArray)) != bytesLen {
			return nil, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(dataType)], string(vectorArray),
				fmt.Sprintf("dimension: %d, bytesLen: %d, but length of []byte: %d", dimension, bytesLen, len(vectorArray)))
		}
	}
	return values, nil
}

// serializeFloatOrByteVectors serializes float32/float16/bfloat16 vectors.
// `[[1, 2, 3], [4.0, 5.0, 6.0]] is float32 vector,
// `["4z1jPgAAgL8=", "gD+AP4A/gD8="]` is float16/bfloat16 vector.
func serializeFloatOrByteVectors(jsonResult gjson.Result, dataType schemapb.DataType, dimension int64, fpArrayToBytesFunc func([]float32) []byte) ([][]byte, error) {
	firstElement := jsonResult.Get("0")

	// Clients may send float32 vector because they are inconvenient of processing float16 or bfloat16.
	// Float32 vector is an array in JSON format, like `[1.0, 2.0, 3.0]`, `[1, 2, 3]`, etc,
	// while float16 or bfloat16 vector is a string in JSON format, like `"4z1jPgAAgL8="`, `"gD+AP4A/gD8="`, etc.
	if firstElement.IsArray() {
		return serializeFloatVectors(jsonResult.Raw, dataType, dimension, dimension*2, fpArrayToBytesFunc)
	} else if firstElement.Type == gjson.String || !firstElement.Exists() {
		// consider corner case: `[]`
		return serializeByteVectors(jsonResult.Raw, dataType, dimension, dimension*2)
	}
	return nil, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(dataType)], jsonResult.Raw, "invalid type")
}

func serializeSparseFloatVectors(vectors []gjson.Result, dataType schemapb.DataType) ([][]byte, error) {
	values := make([][]byte, 0, len(vectors))
	for _, vector := range vectors {
		vectorBytes := []byte(vector.String())
		sparseVector, err := typeutil.CreateSparseFloatRowFromJSON(vectorBytes)
		if err != nil {
			return nil, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(dataType)], vector.String(), err.Error())
		}
		values = append(values, sparseVector)
	}
	return values, nil
}

func serializeInt8Vectors(vectorStr string, dataType schemapb.DataType, dimension int64, int8ArrayToBytesFunc func([]int8) []byte) ([][]byte, error) {
	var int8Values [][]int8
	err := json.Unmarshal([]byte(vectorStr), &int8Values)
	if err != nil {
		return nil, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(dataType)], vectorStr, err.Error())
	}
	values := make([][]byte, 0, len(int8Values))
	for _, vectorArray := range int8Values {
		if int64(len(vectorArray)) != dimension {
			return nil, merr.WrapErrParameterInvalid(schemapb.DataType_name[int32(dataType)], vectorStr,
				fmt.Sprintf("dimension: %d, but length of []int8: %d", dimension, len(vectorArray)))
		}
		vectorBytes := int8ArrayToBytesFunc(vectorArray)
		values = append(values, vectorBytes)
	}
	return values, nil
}

func convertQueries2Placeholder(body string, dataType schemapb.DataType, dimension int64) (*commonpb.PlaceholderValue, error) {
	var valueType commonpb.PlaceholderType
	var values [][]byte
	var err error
	switch dataType {
	case schemapb.DataType_FloatVector:
		valueType = commonpb.PlaceholderType_FloatVector
		values, err = serializeFloatVectors(gjson.Get(body, HTTPRequestData).Raw, dataType, dimension, dimension*4, typeutil.Float32ArrayToBytes)
	case schemapb.DataType_BinaryVector:
		valueType = commonpb.PlaceholderType_BinaryVector
		values, err = serializeByteVectors(gjson.Get(body, HTTPRequestData).Raw, dataType, dimension, dimension/8)
	case schemapb.DataType_Float16Vector:
		valueType = commonpb.PlaceholderType_Float16Vector
		values, err = serializeFloatOrByteVectors(gjson.Get(body, HTTPRequestData), dataType, dimension, typeutil.Float32ArrayToFloat16Bytes)
	case schemapb.DataType_BFloat16Vector:
		valueType = commonpb.PlaceholderType_BFloat16Vector
		values, err = serializeFloatOrByteVectors(gjson.Get(body, HTTPRequestData), dataType, dimension, typeutil.Float32ArrayToBFloat16Bytes)
	case schemapb.DataType_SparseFloatVector:
		valueType = commonpb.PlaceholderType_SparseFloatVector
		values, err = serializeSparseFloatVectors(gjson.Get(body, HTTPRequestData).Array(), dataType)
	case schemapb.DataType_Int8Vector:
		valueType = commonpb.PlaceholderType_Int8Vector
		values, err = serializeInt8Vectors(gjson.Get(body, HTTPRequestData).Raw, dataType, dimension, typeutil.Int8ArrayToBytes)
	case schemapb.DataType_VarChar:
		valueType = commonpb.PlaceholderType_VarChar
		res := gjson.Get(body, HTTPRequestData).Array()
		values = make([][]byte, 0, len(res))
		for _, v := range res {
			values = append(values, []byte(v.String()))
		}
	}
	if err != nil {
		return nil, err
	}
	return &commonpb.PlaceholderValue{
		Tag:    "$0",
		Type:   valueType,
		Values: values,
	}, nil
}

// todo: support [][]byte for BinaryVector
func vectors2PlaceholderGroupBytes(vectors [][]float32) []byte {
	var placeHolderType commonpb.PlaceholderType
	ph := &commonpb.PlaceholderValue{
		Tag:    "$0",
		Values: make([][]byte, 0, len(vectors)),
	}
	if len(vectors) != 0 {
		placeHolderType = commonpb.PlaceholderType_FloatVector

		ph.Type = placeHolderType
		for _, vector := range vectors {
			ph.Values = append(ph.Values, typeutil.Float32ArrayToBytes(vector))
		}
	}
	phg := &commonpb.PlaceholderGroup{
		Placeholders: []*commonpb.PlaceholderValue{
			ph,
		},
	}

	bs, _ := proto.Marshal(phg)
	return bs
}

// --------------------- get/query/search response --------------------- //
func genDynamicFields(fields []string, list []*schemapb.FieldData) []string {
	nonDynamicFieldNames := make(map[string]struct{})
	for _, field := range list {
		if !field.IsDynamic {
			nonDynamicFieldNames[field.FieldName] = struct{}{}
		}
	}
	dynamicFields := []string{}
	for _, fieldName := range fields {
		if _, exist := nonDynamicFieldNames[fieldName]; !exist {
			dynamicFields = append(dynamicFields, fieldName)
		}
	}
	return dynamicFields
}

func fieldDataRowCount(fieldData *schemapb.FieldData) (int64, error) {
	if len(fieldData.GetValidData()) > 0 {
		return int64(len(fieldData.GetValidData())), nil
	}
	return fieldDataValueCount(fieldData)
}

func fieldDataValueCount(fieldData *schemapb.FieldData) (int64, error) {
	switch fieldData.GetType() {
	case schemapb.DataType_Bool:
		return int64(len(fieldData.GetScalars().GetBoolData().GetData())), nil
	case schemapb.DataType_Int8, schemapb.DataType_Int16, schemapb.DataType_Int32:
		return int64(len(fieldData.GetScalars().GetIntData().GetData())), nil
	case schemapb.DataType_Int64:
		return int64(len(fieldData.GetScalars().GetLongData().GetData())), nil
	case schemapb.DataType_Float:
		return int64(len(fieldData.GetScalars().GetFloatData().GetData())), nil
	case schemapb.DataType_Double:
		return int64(len(fieldData.GetScalars().GetDoubleData().GetData())), nil
	case schemapb.DataType_Timestamptz:
		if fieldData.GetScalars().GetTimestamptzData() != nil {
			return int64(len(fieldData.GetScalars().GetTimestamptzData().GetData())), nil
		}
		return int64(len(fieldData.GetScalars().GetStringData().GetData())), nil
	case schemapb.DataType_String, schemapb.DataType_VarChar:
		return int64(len(fieldData.GetScalars().GetStringData().GetData())), nil
	case schemapb.DataType_Array:
		return int64(len(fieldData.GetScalars().GetArrayData().GetData())), nil
	case schemapb.DataType_JSON:
		return int64(len(fieldData.GetScalars().GetJsonData().GetData())), nil
	case schemapb.DataType_Geometry:
		if fieldData.GetScalars().GetGeometryData() != nil {
			return int64(len(fieldData.GetScalars().GetGeometryData().GetData())), nil
		}
		return int64(len(fieldData.GetScalars().GetGeometryWktData().GetData())), nil
	case schemapb.DataType_BinaryVector:
		dim := fieldData.GetVectors().GetDim()
		bytesPerRow := dim / 8
		if bytesPerRow <= 0 {
			return 0, merr.WrapErrParameterInvalidMsg("invalid binary vector dimension %d for field %s", dim, fieldData.GetFieldName())
		}
		return int64(len(fieldData.GetVectors().GetBinaryVector())) / bytesPerRow, nil
	case schemapb.DataType_FloatVector:
		dim := fieldData.GetVectors().GetDim()
		if dim <= 0 {
			return 0, merr.WrapErrParameterInvalidMsg("invalid float vector dimension %d for field %s", dim, fieldData.GetFieldName())
		}
		return int64(len(fieldData.GetVectors().GetFloatVector().GetData())) / dim, nil
	case schemapb.DataType_Float16Vector:
		dim := fieldData.GetVectors().GetDim()
		bytesPerRow := dim * 2
		if bytesPerRow <= 0 {
			return 0, merr.WrapErrParameterInvalidMsg("invalid float16 vector dimension %d for field %s", dim, fieldData.GetFieldName())
		}
		return int64(len(fieldData.GetVectors().GetFloat16Vector())) / bytesPerRow, nil
	case schemapb.DataType_BFloat16Vector:
		dim := fieldData.GetVectors().GetDim()
		bytesPerRow := dim * 2
		if bytesPerRow <= 0 {
			return 0, merr.WrapErrParameterInvalidMsg("invalid bfloat16 vector dimension %d for field %s", dim, fieldData.GetFieldName())
		}
		return int64(len(fieldData.GetVectors().GetBfloat16Vector())) / bytesPerRow, nil
	case schemapb.DataType_SparseFloatVector:
		return int64(len(fieldData.GetVectors().GetSparseFloatVector().GetContents())), nil
	case schemapb.DataType_Int8Vector:
		dim := fieldData.GetVectors().GetDim()
		if dim <= 0 {
			return 0, merr.WrapErrParameterInvalidMsg("invalid int8 vector dimension %d for field %s", dim, fieldData.GetFieldName())
		}
		return int64(len(fieldData.GetVectors().GetInt8Vector())) / dim, nil
	case schemapb.DataType_ArrayOfStruct:
		subs := fieldData.GetStructArrays().GetFields()
		if len(subs) == 0 {
			return 0, nil
		}
		switch subs[0].GetType() {
		case schemapb.DataType_Array:
			return int64(len(subs[0].GetScalars().GetArrayData().GetData())), nil
		case schemapb.DataType_ArrayOfVector:
			return int64(len(subs[0].GetVectors().GetVectorArray().GetData())), nil
		default:
			return 0, merr.WrapErrParameterInvalidMsg("unsupported struct sub-field type %s for field %s", subs[0].GetType(), fieldData.GetFieldName())
		}
	default:
		return 0, merr.WrapErrParameterInvalidMsg("the type(%v) of field(%v) is not supported, use other sdk please", fieldData.GetType(), fieldData.GetFieldName())
	}
}

type fieldDataRowAccessor struct {
	fieldData      *schemapb.FieldData
	validData      []bool
	compactIndices []int64
}

func newFieldDataRowAccessor(fieldData *schemapb.FieldData) (*fieldDataRowAccessor, error) {
	accessor := &fieldDataRowAccessor{
		fieldData: fieldData,
		validData: fieldData.GetValidData(),
	}
	if len(accessor.validData) == 0 {
		return accessor, nil
	}
	isNullableVector := typeutil.IsCompactNullableVectorFieldData(fieldData)

	compactIndices := make([]int64, len(accessor.validData))
	validCount := int64(0)
	for i, valid := range accessor.validData {
		if valid {
			compactIndices[i] = validCount
			validCount++
		} else {
			compactIndices[i] = -1
		}
	}
	if isNullableVector {
		if err := funcutil.ValidateNullableVectorFieldDataCompact(fieldData, uint64(len(accessor.validData)), true); err != nil {
			return nil, err
		}
		accessor.compactIndices = compactIndices
		return accessor, nil
	}
	if validCount == 0 {
		accessor.compactIndices = compactIndices
		return accessor, nil
	}

	valueCount, err := fieldDataValueCount(fieldData)
	if err != nil {
		return nil, err
	}
	if valueCount == int64(len(accessor.validData)) {
		return accessor, nil
	}
	if valueCount != validCount {
		return nil, merr.WrapErrParameterInvalidMsg("field %s has %d valid rows, but data length is %d", fieldData.GetFieldName(), validCount, valueCount)
	}
	accessor.compactIndices = compactIndices
	return accessor, nil
}

func (accessor *fieldDataRowAccessor) rowIndex(rowIdx int64) (int64, bool, error) {
	if len(accessor.validData) == 0 {
		return rowIdx, true, nil
	}
	if rowIdx >= int64(len(accessor.validData)) {
		return 0, false, merr.WrapErrParameterInvalidMsg("row index %d out of range for field %s valid data length %d", rowIdx, accessor.fieldData.GetFieldName(), len(accessor.validData))
	}
	if !accessor.validData[rowIdx] {
		return 0, false, nil
	}
	if accessor.compactIndices != nil {
		return accessor.compactIndices[rowIdx], true, nil
	}
	return rowIdx, true, nil
}

//nolint:gosec // G602: slice indices are bounded by rowsNum which is derived from the data length
func buildQueryResp(rowsNum int64, needFields []string, fieldDataList []*schemapb.FieldData, ids *schemapb.IDs,
	scores []float32, enableInt64 bool, collectionSchema *schemapb.CollectionSchema,
) ([]map[string]interface{}, error) {
	columnNum := len(fieldDataList)
	if rowsNum == int64(0) { // always
		if columnNum > 0 {
			var err error
			rowsNum, err = fieldDataRowCount(fieldDataList[0])
			if err != nil {
				return nil, err
			}
		} else if ids != nil {
			switch ids.GetIdField().(type) {
			case *schemapb.IDs_IntId:
				int64Pks := ids.GetIntId().GetData()
				rowsNum = int64(len(int64Pks))
			case *schemapb.IDs_StrId:
				stringPks := ids.GetStrId().GetData()
				rowsNum = int64(len(stringPks))
			default:
				return nil, merr.WrapErrParameterInvalidMsg("the type of primary key(id) is not supported, use other sdk please")
			}
		}
	}
	if rowsNum == int64(0) {
		return []map[string]interface{}{}, nil
	}
	fieldDataAccessors := make([]*fieldDataRowAccessor, 0, columnNum)
	for _, fieldData := range fieldDataList {
		accessor, err := newFieldDataRowAccessor(fieldData)
		if err != nil {
			return nil, err
		}
		fieldDataAccessors = append(fieldDataAccessors, accessor)
	}
	queryResp := make([]map[string]interface{}, 0, rowsNum)
	dynamicOutputFields := genDynamicFields(needFields, fieldDataList)

	pkFieldName := DefaultPrimaryFieldName
	if collectionSchema != nil {
		fieldsSchema := collectionSchema.GetFields()
		for _, field := range fieldsSchema {
			if field.GetIsPrimaryKey() {
				pkFieldName = field.GetName()
				break
			}
		}
	}
	for i := int64(0); i < rowsNum; i++ {
		row := map[string]interface{}{}
		if columnNum > 0 {
			for j := 0; j < columnNum; j++ {
				fieldData := fieldDataList[j]
				dataIdx, valid, err := fieldDataAccessors[j].rowIndex(i)
				if err != nil {
					return nil, err
				}
				if !valid {
					if !fieldData.GetIsDynamic() {
						row[fieldData.GetFieldName()] = nil
					}
					continue
				}
				switch fieldDataList[j].GetType() {
				case schemapb.DataType_Bool:
					row[fieldDataList[j].GetFieldName()] = fieldDataList[j].GetScalars().GetBoolData().GetData()[dataIdx]
				case schemapb.DataType_Int8:
					row[fieldDataList[j].GetFieldName()] = int8(fieldDataList[j].GetScalars().GetIntData().GetData()[dataIdx])
				case schemapb.DataType_Int16:
					row[fieldDataList[j].GetFieldName()] = int16(fieldDataList[j].GetScalars().GetIntData().GetData()[dataIdx])
				case schemapb.DataType_Int32:
					row[fieldDataList[j].GetFieldName()] = fieldDataList[j].GetScalars().GetIntData().GetData()[dataIdx]
				case schemapb.DataType_Int64:
					if enableInt64 {
						row[fieldDataList[j].GetFieldName()] = fieldDataList[j].GetScalars().GetLongData().GetData()[dataIdx]
					} else {
						row[fieldDataList[j].GetFieldName()] = strconv.FormatInt(fieldDataList[j].GetScalars().GetLongData().GetData()[dataIdx], 10)
					}
				case schemapb.DataType_Float:
					row[fieldDataList[j].GetFieldName()] = fieldDataList[j].GetScalars().GetFloatData().GetData()[dataIdx]
				case schemapb.DataType_Double:
					row[fieldDataList[j].GetFieldName()] = fieldDataList[j].GetScalars().GetDoubleData().GetData()[dataIdx]
				case schemapb.DataType_Timestamptz:
					if fieldDataList[j].GetScalars().GetTimestamptzData() != nil {
						row[fieldDataList[j].FieldName] = fieldDataList[j].GetScalars().GetTimestamptzData().GetData()[dataIdx]
					} else {
						row[fieldDataList[j].FieldName] = fieldDataList[j].GetScalars().GetStringData().GetData()[dataIdx]
					}
				case schemapb.DataType_String:
					row[fieldDataList[j].GetFieldName()] = fieldDataList[j].GetScalars().GetStringData().GetData()[dataIdx]
				case schemapb.DataType_VarChar:
					row[fieldDataList[j].GetFieldName()] = fieldDataList[j].GetScalars().GetStringData().GetData()[dataIdx]
				case schemapb.DataType_BinaryVector:
					row[fieldDataList[j].GetFieldName()] = fieldDataList[j].GetVectors().GetBinaryVector()[dataIdx*(fieldDataList[j].GetVectors().GetDim()/8) : (dataIdx+1)*(fieldDataList[j].GetVectors().GetDim()/8)]
				case schemapb.DataType_FloatVector:
					row[fieldDataList[j].GetFieldName()] = fieldDataList[j].GetVectors().GetFloatVector().GetData()[dataIdx*fieldDataList[j].GetVectors().GetDim() : (dataIdx+1)*fieldDataList[j].GetVectors().GetDim()]
				case schemapb.DataType_Float16Vector:
					row[fieldDataList[j].GetFieldName()] = fieldDataList[j].GetVectors().GetFloat16Vector()[dataIdx*(fieldDataList[j].GetVectors().GetDim()*2) : (dataIdx+1)*(fieldDataList[j].GetVectors().GetDim()*2)]
				case schemapb.DataType_BFloat16Vector:
					row[fieldDataList[j].GetFieldName()] = fieldDataList[j].GetVectors().GetBfloat16Vector()[dataIdx*(fieldDataList[j].GetVectors().GetDim()*2) : (dataIdx+1)*(fieldDataList[j].GetVectors().GetDim()*2)]
				case schemapb.DataType_SparseFloatVector:
					row[fieldDataList[j].GetFieldName()] = typeutil.SparseFloatBytesToMap(fieldDataList[j].GetVectors().GetSparseFloatVector().Contents[dataIdx])
				case schemapb.DataType_Int8Vector:
					row[fieldDataList[j].GetFieldName()] = fieldDataList[j].GetVectors().GetInt8Vector()[dataIdx*fieldDataList[j].GetVectors().GetDim() : (dataIdx+1)*fieldDataList[j].GetVectors().GetDim()]
				case schemapb.DataType_Array:
					row[fieldDataList[j].GetFieldName()] = fieldDataList[j].GetScalars().GetArrayData().GetData()[dataIdx]
				case schemapb.DataType_JSON:
					data, ok := fieldDataList[j].GetScalars().GetData().(*schemapb.ScalarField_JsonData)
					if ok && !fieldDataList[j].GetIsDynamic() {
						row[fieldDataList[j].GetFieldName()] = string(data.JsonData.GetData()[dataIdx])
					} else {
						var dataMap map[string]interface{}

						err := json.Unmarshal(fieldDataList[j].GetScalars().GetJsonData().Data[dataIdx], &dataMap)
						if err != nil {
							mlog.Error(context.TODO(),
								fmt.Sprintf("[BuildQueryResp] Unmarshal error %s", err.Error()))
							return nil, err
						}

						if containsString(dynamicOutputFields, fieldDataList[j].GetFieldName()) {
							for key, value := range dataMap {
								row[key] = value
							}
						} else {
							for _, dynamicField := range dynamicOutputFields {
								if _, ok := dataMap[dynamicField]; ok {
									row[dynamicField] = dataMap[dynamicField]
								}
							}
						}
					}
				case schemapb.DataType_Geometry:
					if fieldDataList[j].GetScalars().GetGeometryData() != nil {
						row[fieldDataList[j].FieldName] = fieldDataList[j].GetScalars().GetGeometryData().GetData()[dataIdx]
					} else {
						row[fieldDataList[j].FieldName] = fieldDataList[j].GetScalars().GetGeometryWktData().Data[dataIdx]
					}
				case schemapb.DataType_ArrayOfStruct:
					structRow, err := extractStructArrayRow(fieldDataList[j], int(dataIdx), collectionSchema)
					if err != nil {
						return nil, err
					}
					row[fieldDataList[j].GetFieldName()] = structRow
				default:
					row[fieldDataList[j].GetFieldName()] = ""
				}
			}
		}
		if ids != nil {
			switch ids.GetIdField().(type) {
			case *schemapb.IDs_IntId:
				int64Pks := ids.GetIntId().GetData()
				if enableInt64 {
					row[pkFieldName] = int64Pks[i]
				} else {
					row[pkFieldName] = strconv.FormatInt(int64Pks[i], 10)
				}
			case *schemapb.IDs_StrId:
				stringPks := ids.GetStrId().GetData()
				row[pkFieldName] = stringPks[i]
			default:
				return nil, merr.WrapErrParameterInvalidMsg("the type of primary key(id) is not supported, use other sdk please")
			}
		}
		if scores != nil && int64(len(scores)) > i {
			row[HTTPReturnDistance] = scores[i] // only 8 decimal places
		}
		queryResp = append(queryResp, row)
	}

	return queryResp, nil
}

func hasSearchAggregationResult(results *schemapb.SearchResultData) bool {
	return results != nil && (len(results.GetAggTopks()) > 0 || len(results.GetAggBuckets()) > 0)
}

func buildSearchAggregationResp(results *schemapb.SearchResultData, enableInt64 bool, collectionSchema *schemapb.CollectionSchema) ([]gin.H, error) {
	if results == nil {
		// The aggregation payload is produced by the server-side reduce, never
		// by the request: a malformed shape is an internal contract violation.
		return nil, merr.WrapErrServiceInternalMsg("search_aggregation result is nil")
	}
	aggTopks := results.GetAggTopks()
	pbBuckets := results.GetAggBuckets()
	if len(aggTopks) == 0 {
		return nil, merr.WrapErrServiceInternalMsg("search_aggregation response missing agg_topks")
	}
	if results.GetNumQueries() <= 0 {
		return nil, merr.WrapErrServiceInternalMsg("search_aggregation response missing nq")
	}
	if len(aggTopks) != int(results.GetNumQueries()) {
		return nil, merr.WrapErrServiceInternalMsg("search_aggregation agg_topks length %d does not match nq %d", len(aggTopks), results.GetNumQueries())
	}

	total := int64(0)
	for _, topk := range aggTopks {
		if topk < 0 {
			return nil, merr.WrapErrServiceInternalMsg("search_aggregation agg_topks cannot contain negative values")
		}
		total += topk
	}
	if total != int64(len(pbBuckets)) {
		return nil, merr.WrapErrServiceInternalMsg("search_aggregation agg_topks sum %d does not match bucket count %d", total, len(pbBuckets))
	}

	output := make([]gin.H, 0, len(aggTopks))
	offset := 0
	for _, topk := range aggTopks {
		buckets := make([]gin.H, 0, int(topk))
		for i := int64(0); i < topk; i++ {
			bucket, err := buildAggBucketResp(pbBuckets[offset], enableInt64, collectionSchema)
			if err != nil {
				return nil, err
			}
			buckets = append(buckets, bucket)
			offset++
		}
		output = append(output, gin.H{"buckets": buckets})
	}
	return output, nil
}

func buildAggBucketResp(pb *schemapb.AggBucket, enableInt64 bool, collectionSchema *schemapb.CollectionSchema) (gin.H, error) {
	if pb == nil {
		return nil, merr.WrapErrServiceInternalMsg("search_aggregation bucket is nil")
	}
	bucket := gin.H{
		"key":       buildAggBucketKeyResp(pb.GetKey(), enableInt64),
		"count":     formatRESTInt64(pb.GetCount(), enableInt64),
		"metrics":   buildAggMetricsResp(pb.GetMetrics(), enableInt64),
		"hits":      buildAggHitsResp(pb.GetHits(), enableInt64, collectionSchema),
		"subGroups": []gin.H{},
	}
	subGroups := make([]gin.H, 0, len(pb.GetSubGroups()))
	for _, sub := range pb.GetSubGroups() {
		subGroup, err := buildAggBucketResp(sub, enableInt64, collectionSchema)
		if err != nil {
			return nil, err
		}
		subGroups = append(subGroups, subGroup)
	}
	bucket["subGroups"] = subGroups
	return bucket, nil
}

func buildAggBucketKeyResp(keys []*schemapb.BucketKeyEntry, enableInt64 bool) []gin.H {
	resp := make([]gin.H, 0, len(keys))
	for _, key := range keys {
		if key == nil {
			resp = append(resp, gin.H{})
			continue
		}
		fieldName := key.GetFieldName()
		if fieldName == "" {
			fieldName = strconv.FormatInt(key.GetFieldId(), 10)
		}
		resp = append(resp, gin.H{
			"fieldName": fieldName,
			"fieldId":   formatRESTInt64(key.GetFieldId(), enableInt64),
			"value":     bucketKeyEntryValueToRESTAny(key, enableInt64),
		})
	}
	return resp
}

func buildAggMetricsResp(metrics map[string]*schemapb.MetricValue, enableInt64 bool) gin.H {
	resp := make(gin.H, len(metrics))
	for alias, metric := range metrics {
		resp[alias] = metricValueToRESTAny(metric, enableInt64)
	}
	return resp
}

func buildAggHitsResp(hits []*schemapb.AggHit, enableInt64 bool, collectionSchema *schemapb.CollectionSchema) []gin.H {
	resp := make([]gin.H, 0, len(hits))
	pkFieldName := getRESTPrimaryFieldName(collectionSchema)
	for _, hit := range hits {
		if hit == nil {
			resp = append(resp, gin.H{})
			continue
		}
		row := gin.H{
			pkFieldName:        aggHitPKToRESTAny(hit, enableInt64),
			HTTPReturnDistance: hit.GetScore(),
		}
		for _, field := range hit.GetFields() {
			if field == nil {
				continue
			}
			fieldName := field.GetFieldName()
			if fieldName == "" {
				fieldName = strconv.FormatInt(field.GetFieldId(), 10)
			}
			row[fieldName] = aggHitFieldValueToRESTAny(field, enableInt64)
		}
		resp = append(resp, row)
	}
	return resp
}

func getRESTPrimaryFieldName(collectionSchema *schemapb.CollectionSchema) string {
	if collectionSchema == nil {
		return DefaultPrimaryFieldName
	}
	for _, field := range collectionSchema.GetFields() {
		if field.GetIsPrimaryKey() {
			return field.GetName()
		}
	}
	return DefaultPrimaryFieldName
}

func formatRESTInt64(v int64, enableInt64 bool) interface{} {
	if enableInt64 {
		return v
	}
	return strconv.FormatInt(v, 10)
}

func metricValueToRESTAny(pb *schemapb.MetricValue, enableInt64 bool) interface{} {
	if pb == nil {
		return nil
	}
	switch v := pb.GetValue().(type) {
	case *schemapb.MetricValue_IntVal:
		return formatRESTInt64(v.IntVal, enableInt64)
	case *schemapb.MetricValue_DoubleVal:
		return v.DoubleVal
	case *schemapb.MetricValue_StringVal:
		return v.StringVal
	case *schemapb.MetricValue_BoolVal:
		return v.BoolVal
	default:
		return nil
	}
}

func bucketKeyEntryValueToRESTAny(pb *schemapb.BucketKeyEntry, enableInt64 bool) interface{} {
	if pb == nil {
		return nil
	}
	switch v := pb.GetValue().(type) {
	case *schemapb.BucketKeyEntry_IntVal:
		return formatRESTInt64(v.IntVal, enableInt64)
	case *schemapb.BucketKeyEntry_StringVal:
		return v.StringVal
	case *schemapb.BucketKeyEntry_BoolVal:
		return v.BoolVal
	default:
		return nil
	}
}

func aggHitPKToRESTAny(pb *schemapb.AggHit, enableInt64 bool) interface{} {
	if pb == nil {
		return nil
	}
	switch v := pb.GetPk().(type) {
	case *schemapb.AggHit_IntPk:
		return formatRESTInt64(v.IntPk, enableInt64)
	case *schemapb.AggHit_StrPk:
		return v.StrPk
	default:
		return nil
	}
}

func aggHitFieldValueToRESTAny(pb *schemapb.AggHitField, enableInt64 bool) interface{} {
	if pb == nil {
		return nil
	}
	switch v := pb.GetValue().(type) {
	case *schemapb.AggHitField_IntVal:
		return formatRESTInt64(v.IntVal, enableInt64)
	case *schemapb.AggHitField_BoolVal:
		return v.BoolVal
	case *schemapb.AggHitField_FloatVal:
		return v.FloatVal
	case *schemapb.AggHitField_DoubleVal:
		return v.DoubleVal
	case *schemapb.AggHitField_StringVal:
		return v.StringVal
	case *schemapb.AggHitField_BytesVal:
		return v.BytesVal
	default:
		return nil
	}
}

func formatInt64(intArray []int64) []string {
	stringArray := make([]string, 0, len(intArray))
	for _, i := range intArray {
		stringArray = append(stringArray, strconv.FormatInt(i, 10))
	}
	return stringArray
}

func CheckLimiter(ctx context.Context, req interface{}, pxy types.ProxyComponent) (any, error) {
	if !paramtable.Get().QuotaConfig.QuotaAndLimitsEnabled.GetAsBool() {
		return nil, nil
	}
	// apply limiter for http/http2 server
	limiter, err := pxy.GetRateLimiter()
	if err != nil {
		mlog.Error(ctx, "Get proxy rate limiter for httpV1/V2 server failed", mlog.Err(err))
		return nil, err
	}

	request, ok := req.(proto.Message)
	if !ok {
		return nil, merr.WrapErrParameterInvalidMsg("wrong req format when check limiter")
	}

	dbID, collectionIDToPartIDs, rt, n, err := proxy.GetRequestInfo(ctx, request)
	if err != nil {
		return nil, err
	}
	err = limiter.Check(dbID, collectionIDToPartIDs, rt, n)
	nodeID := strconv.FormatInt(paramtable.GetNodeID(), 10)
	metrics.ProxyRateLimitReqCount.WithLabelValues(nodeID, rt.String(), metrics.TotalLabel).Inc()
	if err != nil {
		metrics.ProxyRateLimitReqCount.WithLabelValues(nodeID, rt.String(), metrics.FailLabel).Inc()
		return proxy.GetFailedResponse(req, err), err
	}
	metrics.ProxyRateLimitReqCount.WithLabelValues(nodeID, rt.String(), metrics.SuccessLabel).Inc()
	return nil, nil
}

func convertConsistencyLevel(reqConsistencyLevel string) (commonpb.ConsistencyLevel, bool, error) {
	if reqConsistencyLevel != "" {
		level, ok := commonpb.ConsistencyLevel_value[reqConsistencyLevel]
		if !ok {
			return 0, false, merr.WrapErrParameterInvalidMsg("parameter:'%s' is incorrect, please check it", reqConsistencyLevel)
		}
		return commonpb.ConsistencyLevel(level), false, nil
	}
	// ConsistencyLevel_Bounded default in PyMilvus
	return commonpb.ConsistencyLevel_Bounded, true, nil
}

func convertDefaultValue(value interface{}, dataType schemapb.DataType) (*schemapb.ValueField, error) {
	if value == nil {
		return nil, nil
	}
	switch dataType {
	case schemapb.DataType_Bool:
		v, ok := value.(bool)
		if !ok {
			return nil, merr.WrapErrParameterInvalidMsg(`cannot use "%v"(type: %T) as bool default value`, value, value)
		}
		data := &schemapb.ValueField{
			Data: &schemapb.ValueField_BoolData{
				BoolData: v,
			},
		}
		return data, nil

	case schemapb.DataType_Int8, schemapb.DataType_Int16, schemapb.DataType_Int32:
		// all passed number is float64 type
		v, ok := value.(float64)
		if !ok {
			return nil, merr.WrapErrParameterInvalidMsg(`cannot use ""%v"(type: %T) as int default value`, value, value)
		}
		data := &schemapb.ValueField{
			Data: &schemapb.ValueField_IntData{
				IntData: int32(v),
			},
		}
		return data, nil

	case schemapb.DataType_Int64:
		v, ok := value.(float64)
		if !ok {
			return nil, merr.WrapErrParameterInvalidMsg(`cannot use "%v"(type: %T) as long default value`, value, value)
		}
		data := &schemapb.ValueField{
			Data: &schemapb.ValueField_LongData{
				LongData: int64(v),
			},
		}
		return data, nil

	case schemapb.DataType_Float:
		v, ok := value.(float64)
		if !ok {
			return nil, merr.WrapErrParameterInvalidMsg(`cannot use "%v"(type: %T) as float default value`, value, value)
		}
		data := &schemapb.ValueField{
			Data: &schemapb.ValueField_FloatData{
				FloatData: float32(v),
			},
		}
		return data, nil

	case schemapb.DataType_Double:
		v, ok := value.(float64)
		if !ok {
			return nil, merr.WrapErrParameterInvalidMsg(`cannot use "%v"(type: %T) as float default value`, value, value)
		}
		data := &schemapb.ValueField{
			Data: &schemapb.ValueField_DoubleData{
				DoubleData: v,
			},
		}
		return data, nil

	case schemapb.DataType_Timestamptz:
		v, ok := value.(string)
		if !ok {
			return nil, merr.WrapErrParameterInvalid("string", value, "Wrong defaultValue type")
		}
		data := &schemapb.ValueField{
			Data: &schemapb.ValueField_StringData{
				StringData: v,
			},
		}
		return data, nil

	case schemapb.DataType_Geometry:
		v, ok := value.(string)
		if !ok {
			return nil, merr.WrapErrParameterInvalidMsg(`cannot use "%v"(type: %T) as geometry default value`, value, value)
		}
		data := &schemapb.ValueField{
			Data: &schemapb.ValueField_StringData{
				StringData: v,
			},
		}
		return data, nil

	case schemapb.DataType_String, schemapb.DataType_VarChar:
		v, ok := value.(string)
		if !ok {
			return nil, merr.WrapErrParameterInvalidMsg(`cannot use "%v"(type: %T) as string default value`, value, value)
		}
		data := &schemapb.ValueField{
			Data: &schemapb.ValueField_StringData{
				StringData: v,
			},
		}
		return data, nil
	default:
		return nil, merr.WrapErrParameterInvalidMsg("Unexpected default value type: %s", dataType.String())
	}
}

func convertToExtraParams(indexParam IndexParam) ([]*commonpb.KeyValuePair, error) {
	var params []*commonpb.KeyValuePair
	if indexParam.IndexType != "" {
		params = append(params, &commonpb.KeyValuePair{Key: common.IndexTypeKey, Value: indexParam.IndexType})
	}
	if indexParam.IndexType == "" {
		for key, value := range indexParam.Params {
			if key == common.IndexTypeKey {
				params = append(params, &commonpb.KeyValuePair{Key: common.IndexTypeKey, Value: fmt.Sprintf("%v", value)})
				break
			}
		}
	}
	if indexParam.MetricType != "" {
		params = append(params, &commonpb.KeyValuePair{Key: common.MetricTypeKey, Value: indexParam.MetricType})
	}
	if len(indexParam.Params) != 0 {
		v, err := json.Marshal(indexParam.Params)
		if err != nil {
			return nil, err
		}
		params = append(params, &commonpb.KeyValuePair{Key: common.ParamsKey, Value: string(v)})
	}
	return params, nil
}

func getElementTypeParams(param interface{}) (string, error) {
	if str, ok := param.(string); ok {
		return str, nil
	}

	jsonBytes, err := json.Marshal(param)
	if err != nil {
		return "", err
	}
	return string(jsonBytes), nil
}

func MetricsHandlerFunc(c *gin.Context) {
	path := c.Request.URL.Path
	metrics.RestfulFunctionCall.WithLabelValues(
		strconv.FormatInt(paramtable.GetNodeID(), 10), path,
	).Inc()
	if c.Request.ContentLength >= 0 {
		metrics.RestfulReceiveBytes.WithLabelValues(
			strconv.FormatInt(paramtable.GetNodeID(), 10), path,
		).Add(float64(c.Request.ContentLength))
	}
	start := time.Now()

	// Process request
	c.Next()

	latency := time.Since(start)
	metrics.RestfulReqLatency.WithLabelValues(
		strconv.FormatInt(paramtable.GetNodeID(), 10), path,
	).Observe(float64(latency.Milliseconds()))

	// see https://github.com/milvus-io/milvus/issues/35767, counter cannot add negative value
	// when response is not written(say timeout/network broken), panicking may happen if not check
	if size := c.Writer.Size(); size > 0 {
		metrics.RestfulSendBytes.WithLabelValues(
			strconv.FormatInt(paramtable.GetNodeID(), 10), path,
		).Add(float64(c.Writer.Size()))
	}
}

func LoggerHandlerFunc() gin.HandlerFunc {
	notlogged := proxy.Params.ProxyCfg.GinLogSkipPaths.GetAsStrings()
	var skip map[string]struct{}
	if length := len(notlogged); length > 0 {
		skip = make(map[string]struct{}, length)
		for _, p := range notlogged {
			skip[p] = struct{}{}
		}
	}

	return func(c *gin.Context) {
		start := time.Now()
		path := c.Request.URL.Path
		raw := c.Request.URL.RawQuery

		c.Next()

		if _, ok := skip[path]; ok {
			return
		}

		param := gin.LogFormatterParams{
			Request:      c.Request,
			TimeStamp:    time.Now(),
			ClientIP:     c.ClientIP(),
			Method:       c.Request.Method,
			StatusCode:   c.Writer.Status(),
			ErrorMessage: c.Errors.ByType(gin.ErrorTypePrivate).String(),
			BodySize:     c.Writer.Size(),
		}
		param.Latency = param.TimeStamp.Sub(start)
		if param.Latency > time.Minute {
			param.Latency = param.Latency.Truncate(time.Second)
		}
		if raw != "" {
			path = path + "?" + raw
		}
		param.Path = path

		traceID, _ := c.Get("traceID")
		if traceID == nil {
			traceID = ""
		}

		accesslog.SetHTTPParams(c, &param)
		fmt.Fprintf(gin.DefaultWriter, "[%v] [GIN] [%s] [traceID=%s] [code=%3d] [latency=%v] [client=%s] [method=%s] [error=%s]\n",
			param.TimeStamp.Format("2006/01/02 15:04:05.000 Z07:00"),
			param.Path,
			traceID,
			param.StatusCode,
			param.Latency,
			param.ClientIP,
			param.Method,
			param.ErrorMessage,
		)
	}
}

func RequestHandlerFunc(c *gin.Context) {
	_, err := strconv.ParseBool(c.Request.Header.Get(mhttp.HTTPHeaderAllowInt64))
	if err != nil {
		if paramtable.Get().HTTPCfg.AcceptTypeAllowInt64.GetAsBool() {
			c.Request.Header.Set(mhttp.HTTPHeaderAllowInt64, "true")
		} else {
			c.Request.Header.Set(mhttp.HTTPHeaderAllowInt64, "false")
		}
	}
	c.Writer.Header().Set("Access-Control-Allow-Origin", "*")
	c.Writer.Header().Set("Access-Control-Allow-Credentials", "true")
	c.Writer.Header().Set("Access-Control-Allow-Headers", "Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, Authorization, accept, origin, Cache-Control, X-Requested-With")
	c.Writer.Header().Set("Access-Control-Allow-Methods", "GET, HEAD, POST, PUT, DELETE, OPTIONS, PATCH")
	c.Writer.Header().Set("X-Content-Type-Options", "nosniff") // Prevents MIME sniffing

	enableHSTS := paramtable.Get().HTTPCfg.EnableHSTS.GetAsBool()
	if enableHSTS {
		maxAge := paramtable.Get().HTTPCfg.HSTSMaxAge.GetValue()
		hstsValue := fmt.Sprintf("max-age=%s", maxAge)
		includeSubDomains := paramtable.Get().HTTPCfg.HSTSIncludeSubDomains.GetAsBool()
		if includeSubDomains {
			hstsValue += "; includeSubDomains"
		}
		c.Writer.Header().Set("Strict-Transport-Security", hstsValue)
	}

	if c.Request.Method == "OPTIONS" {
		c.AbortWithStatus(204)
		return
	}
	c.Next()
}

func generateTemplateArrayData(list []interface{}) *schemapb.TemplateArrayValue {
	dtype := getTemplateArrayType(list)
	var data *schemapb.TemplateArrayValue
	switch dtype {
	case schemapb.DataType_Bool:
		result := make([]bool, len(list))
		for i, item := range list {
			result[i] = item.(bool)
		}
		data = &schemapb.TemplateArrayValue{
			Data: &schemapb.TemplateArrayValue_BoolData{
				BoolData: &schemapb.BoolArray{
					Data: result,
				},
			},
		}
	case schemapb.DataType_String:
		result := make([]string, len(list))
		for i, item := range list {
			result[i] = item.(string)
		}
		data = &schemapb.TemplateArrayValue{
			Data: &schemapb.TemplateArrayValue_StringData{
				StringData: &schemapb.StringArray{
					Data: result,
				},
			},
		}
	case schemapb.DataType_Int64:
		result := make([]int64, len(list))
		for i, item := range list {
			result[i] = int64(item.(float64))
		}
		data = &schemapb.TemplateArrayValue{
			Data: &schemapb.TemplateArrayValue_LongData{
				LongData: &schemapb.LongArray{
					Data: result,
				},
			},
		}
	case schemapb.DataType_Float:
		result := make([]float64, len(list))
		for i, item := range list {
			result[i] = item.(float64)
		}
		data = &schemapb.TemplateArrayValue{
			Data: &schemapb.TemplateArrayValue_DoubleData{
				DoubleData: &schemapb.DoubleArray{
					Data: result,
				},
			},
		}
	case schemapb.DataType_Array:
		result := make([]*schemapb.TemplateArrayValue, len(list))
		for i, item := range list {
			result[i] = generateTemplateArrayData(item.([]interface{}))
		}
		data = &schemapb.TemplateArrayValue{
			Data: &schemapb.TemplateArrayValue_ArrayData{
				ArrayData: &schemapb.TemplateArrayValueArray{
					Data: result,
				},
			},
		}
	case schemapb.DataType_JSON:
		result := make([][]byte, len(list))
		for i, item := range list {
			bytes, err := json.Marshal(item)
			// won't happen
			if err != nil {
				panic(fmt.Sprintf("marshal data(%v) fail, please check it!", item))
			}
			result[i] = bytes
		}
		data = &schemapb.TemplateArrayValue{
			Data: &schemapb.TemplateArrayValue_JsonData{
				JsonData: &schemapb.JSONArray{
					Data: result,
				},
			},
		}
	// won't happen
	default:
		panic(fmt.Sprintf("Unexpected data(%v) type when generateTemplateArrayData, please check it!", list))
	}
	return data
}

func getTemplateArrayType(value []interface{}) schemapb.DataType {
	dtype := getTemplateType(value[0])

	for _, v := range value {
		if getTemplateType(v) != dtype {
			return schemapb.DataType_JSON
		}
	}
	return dtype
}

func getTemplateType(value interface{}) schemapb.DataType {
	switch v := value.(type) {
	case bool:
		return schemapb.DataType_Bool
	case string:
		return schemapb.DataType_String
	case float64:
		// note: all passed number is float64 type
		// if field type is float64, but value in ExpressionTemplate is int64, it's ok to use TemplateValue_Int64Val to store it
		// it will convert to float64 in ./internal/parser/planparserv2/utils.go, Line 233
		if v == math.Trunc(v) && v >= math.MinInt64 && v <= math.MaxInt64 {
			return schemapb.DataType_Int64
		}
		return schemapb.DataType_Float
	// it won't happen
	// case int64:
	case []interface{}:
		return schemapb.DataType_Array
	default:
		panic(fmt.Sprintf("Unexpected data(%v) when getTemplateType, please check it!", value))
	}
}

func generateExpressionTemplate(params map[string]interface{}) map[string]*schemapb.TemplateValue {
	expressionTemplate := make(map[string]*schemapb.TemplateValue, len(params))

	for name, value := range params {
		dtype := getTemplateType(value)
		var data *schemapb.TemplateValue
		switch dtype {
		case schemapb.DataType_Bool:
			data = &schemapb.TemplateValue{
				Val: &schemapb.TemplateValue_BoolVal{
					BoolVal: value.(bool),
				},
			}
		case schemapb.DataType_String:
			data = &schemapb.TemplateValue{
				Val: &schemapb.TemplateValue_StringVal{
					StringVal: value.(string),
				},
			}
		case schemapb.DataType_Int64:
			data = &schemapb.TemplateValue{
				Val: &schemapb.TemplateValue_Int64Val{
					Int64Val: int64(value.(float64)),
				},
			}
		case schemapb.DataType_Float:
			data = &schemapb.TemplateValue{
				Val: &schemapb.TemplateValue_FloatVal{
					FloatVal: value.(float64),
				},
			}
		case schemapb.DataType_Array:
			data = &schemapb.TemplateValue{
				Val: &schemapb.TemplateValue_ArrayVal{
					ArrayVal: generateTemplateArrayData(value.([]interface{})),
				},
			}
		default:
			panic(fmt.Sprintf("Unexpected data(%v) when generateExpressionTemplate, please check it!", data))
		}
		expressionTemplate[name] = data
	}
	return expressionTemplate
}

func WrapErrorToResponse(err error) *milvuspb.BoolResponse {
	return &milvuspb.BoolResponse{
		Status: merr.Status(err),
	}
}

func searchParamsContainAny(reqSearchParams map[string]interface{}, keys ...string) bool {
	for _, key := range keys {
		if _, ok := reqSearchParams[key]; ok {
			return true
		}
	}

	params, ok := reqSearchParams[Params]
	if !ok {
		return false
	}
	paramsMap, ok := params.(map[string]interface{})
	if !ok {
		return false
	}
	for _, key := range keys {
		if _, ok := paramsMap[key]; ok {
			return true
		}
	}
	return false
}

// after 2.5.2, all parameters of search_params can be written into one layer
// no more parameters will be written searchParams.params
// to ensure compatibility and milvus can still get a json format parameter
// try to write all the parameters under searchParams into searchParams.Params
func generateSearchParams(reqSearchParams map[string]interface{}) ([]*commonpb.KeyValuePair, error) {
	var searchParams []*commonpb.KeyValuePair
	var params interface{}
	if val, ok := reqSearchParams[Params]; ok {
		params = val
	}

	paramsMap := make(map[string]interface{})
	if params != nil {
		var ok bool
		if paramsMap, ok = params.(map[string]interface{}); !ok {
			return nil, merr.WrapErrParameterInvalidMsg("searchParams.params must be a dict")
		}
	}

	deepEqual := func(value1, value2 interface{}) bool {
		// try to handle 10.0==10
		switch v1 := value1.(type) {
		case float64:
			if v2, ok := value2.(int); ok {
				return v1 == float64(v2)
			}
		case int:
			if v2, ok := value2.(float64); ok {
				return float64(v1) == v2
			}
		}
		return reflect.DeepEqual(value1, value2)
	}

	for key, value := range reqSearchParams {
		if val, ok := paramsMap[key]; ok {
			if !deepEqual(val, value) {
				return nil, merr.WrapErrParameterInvalidMsg("ambiguous parameter: %s, in search_param: %v, in search_param.params: %v", key, value, val)
			}
		} else if key != Params {
			paramsMap[key] = value
		}
	}

	bs, _ := json.Marshal(paramsMap)
	searchParams = append(searchParams, &commonpb.KeyValuePair{Key: Params, Value: string(bs)})

	for key, value := range reqSearchParams {
		if key != Params {
			// for compatibility
			if key == "ignoreGrowing" {
				key = common.IgnoreGrowing
			}
			searchParams = append(searchParams, &commonpb.KeyValuePair{Key: key, Value: fmt.Sprintf("%v", value)})
		}
	}
	// need to exposure ParamRoundDecimal in req?
	searchParams = append(searchParams, &commonpb.KeyValuePair{Key: ParamRoundDecimal, Value: "-1"})
	return searchParams, nil
}

func convertSearchAggregationReq(req *SearchAggregationReq) (*commonpb.SearchAggregationSpec, error) {
	if req == nil {
		return nil, nil
	}
	if len(req.Fields) == 0 {
		return nil, merr.WrapErrParameterInvalidMsg("searchAggregation.fields must be non-empty")
	}
	fields := make([]string, 0, len(req.Fields))
	for _, field := range req.Fields {
		field = strings.TrimSpace(field)
		if field == "" {
			return nil, merr.WrapErrParameterInvalidMsg("searchAggregation.fields must contain non-empty field names")
		}
		fields = append(fields, field)
	}
	if req.Size <= 0 {
		return nil, merr.WrapErrParameterInvalidMsg("searchAggregation.size must be positive")
	}
	if req.SearchSize < 0 {
		return nil, merr.WrapErrParameterInvalidMsg("searchAggregation.searchSize must be non-negative")
	}
	if req.SearchSize > 0 && req.SearchSize < req.Size {
		return nil, merr.WrapErrParameterInvalidMsg("searchAggregation.searchSize must be greater than or equal to size")
	}

	spec := &commonpb.SearchAggregationSpec{
		Fields:     fields,
		Size:       req.Size,
		SearchSize: req.SearchSize,
	}

	if len(req.Metrics) > 0 {
		spec.Metrics = make(map[string]*commonpb.MetricAggSpec, len(req.Metrics))
	}
	for alias, metric := range req.Metrics {
		alias = strings.TrimSpace(alias)
		op := strings.TrimSpace(metric.Op)
		fieldName := strings.TrimSpace(metric.FieldName)
		if alias == "" {
			return nil, merr.WrapErrParameterInvalidMsg("searchAggregation.metrics alias must be non-empty")
		}
		if op == "" {
			return nil, merr.WrapErrParameterInvalidMsg("searchAggregation.metrics.%s.op must be non-empty", alias)
		}
		if fieldName == "" {
			return nil, merr.WrapErrParameterInvalidMsg("searchAggregation.metrics.%s.fieldName must be non-empty", alias)
		}
		spec.Metrics[alias] = &commonpb.MetricAggSpec{Op: op, FieldName: fieldName}
	}

	for _, order := range req.Order {
		key := strings.TrimSpace(order.Key)
		direction := strings.TrimSpace(order.Direction)
		if key == "" {
			return nil, merr.WrapErrParameterInvalidMsg("searchAggregation.order key must be non-empty")
		}
		if direction == "" {
			return nil, merr.WrapErrParameterInvalidMsg("searchAggregation.order direction must be non-empty")
		}
		spec.Order = append(spec.Order, &commonpb.OrderSpec{Key: key, Direction: direction})
	}

	if req.TopHits != nil {
		topHits, err := convertTopHitsReq(req.TopHits)
		if err != nil {
			return nil, err
		}
		spec.TopHits = topHits
	}

	if req.SubAggregation != nil {
		sub, err := convertSearchAggregationReq(req.SubAggregation)
		if err != nil {
			return nil, err
		}
		spec.SubAggregation = sub
	}

	return spec, nil
}

func convertTopHitsReq(req *TopHitsReq) (*commonpb.TopHitsSpec, error) {
	if req.Size <= 0 {
		return nil, merr.WrapErrParameterInvalidMsg("searchAggregation.topHits.size must be positive")
	}
	spec := &commonpb.TopHitsSpec{Size: req.Size}
	for _, sort := range req.Sort {
		fieldName := strings.TrimSpace(sort.FieldName)
		direction := strings.TrimSpace(sort.Direction)
		if fieldName == "" {
			return nil, merr.WrapErrParameterInvalidMsg("searchAggregation.topHits.sort fieldName must be non-empty")
		}
		if direction == "" {
			return nil, merr.WrapErrParameterInvalidMsg("searchAggregation.topHits.sort direction must be non-empty")
		}
		spec.Sort = append(spec.Sort, &commonpb.SortSpec{FieldName: fieldName, Direction: direction})
	}
	return spec, nil
}

func genFunctionSchema(ctx context.Context, function *FunctionSchema) (*schemapb.FunctionSchema, error) {
	functionTypeValue, ok := schemapb.FunctionType_value[function.FunctionType]
	if !ok {
		mlog.Warn(ctx, "function's data type is invalid(case sensitive).", mlog.Any("function.DataType", function.FunctionType), mlog.Any("function", function))
		return nil, merr.WrapErrParameterInvalidMsg("Unsupported function type: %s", function.FunctionType)
	}
	functionType := schemapb.FunctionType(functionTypeValue)
	description := function.Description
	params := []*commonpb.KeyValuePair{}
	for key, value := range function.Params {
		if reflect.TypeOf(value).Kind() == reflect.Slice || reflect.TypeOf(value).Kind() == reflect.Map {
			bs, err := json.Marshal(value)
			if err != nil {
				return nil, merr.WrapErrParameterInvalidMsg("Marshal function params fail, please check it!")
			}
			params = append(params, &commonpb.KeyValuePair{Key: key, Value: string(bs)})
		} else {
			params = append(params, &commonpb.KeyValuePair{Key: key, Value: fmt.Sprintf("%v", value)})
		}
	}
	return &schemapb.FunctionSchema{
		Name:             function.FunctionName,
		Description:      description,
		Type:             functionType,
		InputFieldNames:  function.InputFieldNames,
		OutputFieldNames: function.OutputFieldNames,
		Params:           params,
	}, nil
}

func genFunctionScore(ctx context.Context, functionScore *FunctionScore) (*schemapb.FunctionScore, error) {
	fScore := schemapb.FunctionScore{
		Functions: []*schemapb.FunctionSchema{},
		Params:    []*commonpb.KeyValuePair{},
	}
	for _, function := range functionScore.Functions {
		f, err := genFunctionSchema(ctx, &function)
		if err != nil {
			return nil, err
		}
		fScore.Functions = append(fScore.Functions, f)
	}
	for key, value := range functionScore.Params {
		fScore.Params = append(fScore.Params, &commonpb.KeyValuePair{Key: key, Value: fmt.Sprintf("%v", value)})
	}
	return &fScore, nil
}

func genFunctionChains(chains []FunctionChainReq) ([]*schemapb.FunctionChain, error) {
	result := make([]*schemapb.FunctionChain, 0, len(chains))
	for i, chainReq := range chains {
		chainPB, err := genFunctionChain(chainReq)
		if err != nil {
			return nil, merr.WrapErrParameterInvalidMsg("functionChains[%d]: %v", i, err)
		}
		result = append(result, chainPB)
	}
	return result, nil
}

func genFunctionChain(req FunctionChainReq) (*schemapb.FunctionChain, error) {
	stage, err := genFunctionChainStage(req.Stage)
	if err != nil {
		return nil, err
	}

	ops := make([]*schemapb.FunctionChainOp, 0, len(req.Ops))
	for i, opReq := range req.Ops {
		opPB, err := genFunctionChainOp(opReq)
		if err != nil {
			return nil, merr.WrapErrParameterInvalidMsg("ops[%d]: %v", i, err)
		}
		ops = append(ops, opPB)
	}

	return &schemapb.FunctionChain{
		Name:  strings.TrimSpace(req.Name),
		Stage: stage,
		Ops:   ops,
	}, nil
}

func genFunctionChainStage(stageName string) (schemapb.FunctionChainStage, error) {
	stageName = strings.TrimSpace(stageName)
	stageValue, ok := schemapb.FunctionChainStage_value[stageName]
	if !ok {
		return schemapb.FunctionChainStage_FunctionChainStageUnspecified, merr.WrapErrParameterInvalidMsg("unsupported function chain stage: %s", stageName)
	}
	stage := schemapb.FunctionChainStage(stageValue)
	if _, err := chain.ProtoStageToReprStage(stage); err != nil {
		return schemapb.FunctionChainStage_FunctionChainStageUnspecified, err
	}
	return stage, nil
}

func genFunctionChainOp(req FunctionChainOpReq) (*schemapb.FunctionChainOp, error) {
	opName := strings.TrimSpace(req.Op)
	if opName == "" {
		return nil, merr.WrapErrParameterInvalidMsg("op name is empty")
	}

	paramMap, err := genFunctionParamMap(req.Params)
	if err != nil {
		return nil, merr.WrapErrParameterInvalidMsg("params: %v", err)
	}

	var exprPB *schemapb.FunctionChainExpr
	if req.Expr != nil {
		exprPB, err = genFunctionChainExpr(*req.Expr)
		if err != nil {
			return nil, merr.WrapErrParameterInvalidMsg("expr: %v", err)
		}
	}

	return &schemapb.FunctionChainOp{
		Op:      opName,
		Expr:    exprPB,
		Inputs:  trimStringList(req.Inputs),
		Outputs: trimStringList(req.Outputs),
		Params:  paramMap,
	}, nil
}

func genFunctionChainExpr(req FunctionChainExprReq) (*schemapb.FunctionChainExpr, error) {
	name := strings.TrimSpace(req.Name)
	if name == "" {
		return nil, merr.WrapErrParameterInvalidMsg("expr name is empty")
	}

	args := make([]*schemapb.FunctionChainExprArg, 0, len(req.Args))
	for i, argReq := range req.Args {
		argPB, err := genFunctionChainExprArg(argReq)
		if err != nil {
			return nil, merr.WrapErrParameterInvalidMsg("args[%d]: %v", i, err)
		}
		args = append(args, argPB)
	}

	params, err := genFunctionParamMap(req.Params)
	if err != nil {
		return nil, merr.WrapErrParameterInvalidMsg("params: %v", err)
	}

	return &schemapb.FunctionChainExpr{
		Name:   name,
		Args:   args,
		Params: params,
	}, nil
}

func genFunctionChainExprArg(req FunctionChainExprArgReq) (*schemapb.FunctionChainExprArg, error) {
	hasColumn := req.Column != nil
	hasLiteral := req.Literal != nil
	if hasColumn == hasLiteral {
		return nil, merr.WrapErrParameterInvalidMsg("exactly one of column or literal is required")
	}
	if hasColumn {
		name := strings.TrimSpace(*req.Column)
		if name == "" {
			return nil, merr.WrapErrParameterInvalidMsg("column name is empty")
		}
		return &schemapb.FunctionChainExprArg{
			Arg: &schemapb.FunctionChainExprArg_Column{
				Column: &schemapb.FunctionChainColumnArg{Name: name},
			},
		}, nil
	}

	literal, err := genFunctionParamValue(req.Literal)
	if err != nil {
		return nil, merr.WrapErrParameterInvalidMsg("literal: %v", err)
	}
	return &schemapb.FunctionChainExprArg{
		Arg: &schemapb.FunctionChainExprArg_Literal{Literal: literal},
	}, nil
}

func genFunctionParamMap(params map[string]interface{}) (map[string]*schemapb.FunctionParamValue, error) {
	result := make(map[string]*schemapb.FunctionParamValue, len(params))
	for key, value := range params {
		paramName := strings.TrimSpace(key)
		if paramName == "" {
			return nil, merr.WrapErrParameterInvalidMsg("param name is empty")
		}
		paramValue, err := genFunctionParamValue(value)
		if err != nil {
			return nil, merr.WrapErrParameterInvalidMsg("param %q: %v", key, err)
		}
		result[paramName] = paramValue
	}
	return result, nil
}

func genFunctionParamValue(value interface{}) (*schemapb.FunctionParamValue, error) {
	switch v := value.(type) {
	case nil:
		return nil, merr.WrapErrParameterInvalidMsg("function param value is nil")
	case bool:
		return &schemapb.FunctionParamValue{Value: &schemapb.FunctionParamValue_BoolValue{BoolValue: v}}, nil
	case float64:
		if math.Trunc(v) == v && v >= math.MinInt64 && v <= math.MaxInt64 {
			return &schemapb.FunctionParamValue{Value: &schemapb.FunctionParamValue_Int64Value{Int64Value: int64(v)}}, nil
		}
		return &schemapb.FunctionParamValue{Value: &schemapb.FunctionParamValue_DoubleValue{DoubleValue: v}}, nil
	case string:
		return &schemapb.FunctionParamValue{Value: &schemapb.FunctionParamValue_StringValue{StringValue: v}}, nil
	case []interface{}:
		values := make([]*schemapb.FunctionParamValue, 0, len(v))
		for i, item := range v {
			converted, err := genFunctionParamValue(item)
			if err != nil {
				return nil, merr.WrapErrParameterInvalidMsg("array[%d]: %v", i, err)
			}
			values = append(values, converted)
		}
		return &schemapb.FunctionParamValue{Value: &schemapb.FunctionParamValue_ArrayValue{ArrayValue: &schemapb.FunctionParamArray{Values: values}}}, nil
	case map[string]interface{}:
		fields := make(map[string]*schemapb.FunctionParamValue, len(v))
		for key, item := range v {
			fieldName := strings.TrimSpace(key)
			if fieldName == "" {
				return nil, merr.WrapErrParameterInvalidMsg("object field name is empty")
			}
			converted, err := genFunctionParamValue(item)
			if err != nil {
				return nil, merr.WrapErrParameterInvalidMsg("object field %q: %v", key, err)
			}
			fields[fieldName] = converted
		}
		return &schemapb.FunctionParamValue{Value: &schemapb.FunctionParamValue_ObjectValue{ObjectValue: &schemapb.FunctionParamObject{Fields: fields}}}, nil
	default:
		return nil, merr.WrapErrParameterInvalidMsg("unsupported function param value type %T", value)
	}
}

func trimStringList(values []string) []string {
	trimmed := make([]string, len(values))
	for i, value := range values {
		trimmed[i] = strings.TrimSpace(value)
	}
	return trimmed
}