wehub-resource-sync
498b235461
Build and test / Build and test AMD64 Ubuntu 22.04 (push) Failing after 0s
Publish Builder / amazonlinux2023 (push) Failing after 1s
Build and test / UT for Go (push) Has been skipped
Publish KRTE Images / KRTE (push) Failing after 1s
Build and test / Integration Test (push) Has been skipped
Build and test / Upload Code Coverage (push) Has been skipped
Publish Builder / rockylinux9 (push) Failing after 1s
Publish Builder / ubuntu22.04 (push) Failing after 0s
Publish Builder / ubuntu24.04 (push) Failing after 0s
Publish Gpu Builder / publish-gpu-builder (push) Failing after 1s
Publish Test Images / PyTest (push) Failing after 0s
Build and test / UT for Cpp (push) Has been cancelled
3861 lines
127 KiB
Go
3861 lines
127 KiB
Go
// Licensed to the LF AI & Data foundation under one
|
|
// or more contributor license agreements. See the NOTICE file
|
|
// distributed with this work for additional information
|
|
// regarding copyright ownership. The ASF licenses this file
|
|
// to you under the Apache License, Version 2.0 (the
|
|
// "License"); you may not use this file except in compliance
|
|
// with the License. You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package 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, ¶m)
|
|
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
|
|
}
|