milvus-io--milvus/internal/rootcoord/create_collection_task.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 rootcoord

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"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"
	"github.com/milvus-io/milvus/internal/coordinator/snmanager"
	"github.com/milvus-io/milvus/internal/json"
	"github.com/milvus-io/milvus/internal/storagev2/packed"
	"github.com/milvus-io/milvus/internal/streamingcoord/server/balancer"
	"github.com/milvus-io/milvus/internal/util/hookutil"
	"github.com/milvus-io/milvus/pkg/v3/common"
	"github.com/milvus-io/milvus/pkg/v3/mlog"
	"github.com/milvus-io/milvus/pkg/v3/proto/indexpb"
	"github.com/milvus-io/milvus/pkg/v3/proto/querypb"
	"github.com/milvus-io/milvus/pkg/v3/streaming/util/message"
	util "github.com/milvus-io/milvus/pkg/v3/util"
	"github.com/milvus-io/milvus/pkg/v3/util/externalspec"
	"github.com/milvus-io/milvus/pkg/v3/util/funcutil"
	"github.com/milvus-io/milvus/pkg/v3/util/merr"
	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
	"github.com/milvus-io/milvus/pkg/v3/util/retry"
	"github.com/milvus-io/milvus/pkg/v3/util/timestamptz"
	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
)

type createCollectionTask struct {
	*Core
	Req             *milvuspb.CreateCollectionRequest
	header          *message.CreateCollectionMessageHeader
	body            *message.CreateCollectionRequest
	preserveFieldID bool

	// heldFileResourceIds tracks file resources whose refCnt was incremented
	// during validation, to be released if the task fails before Broadcast.
	heldFileResourceIds []int64
}

// releaseFileResources decrements refCnt for file resources that were
// incremented during validation. Called when the task fails before Broadcast.
func (t *createCollectionTask) releaseFileResources() {
	if len(t.heldFileResourceIds) > 0 {
		t.meta.DecFileResourceRefCnt(t.heldFileResourceIds)
		t.heldFileResourceIds = nil
	}
}

func (t *createCollectionTask) validate(ctx context.Context) error {
	if t.Req == nil {
		return merr.WrapErrServiceInternalMsg("empty requests")
	}
	Params := paramtable.Get()

	// 1. check shard number
	shardsNum := t.Req.GetShardsNum()
	var cfgMaxShardNum int32
	if Params.CommonCfg.PreCreatedTopicEnabled.GetAsBool() {
		cfgMaxShardNum = int32(len(Params.CommonCfg.TopicNames.GetAsStrings()))
	} else {
		cfgMaxShardNum = Params.RootCoordCfg.DmlChannelNum.GetAsInt32()
	}
	if shardsNum > cfgMaxShardNum {
		return merr.WrapErrParameterInvalidMsg("shard num (%d) exceeds max configuration (%d)", shardsNum, cfgMaxShardNum)
	}

	cfgShardLimit := Params.ProxyCfg.MaxShardNum.GetAsInt32()
	if shardsNum > cfgShardLimit {
		return merr.WrapErrParameterInvalidMsg("shard num (%d) exceeds system limit (%d)", shardsNum, cfgShardLimit)
	}

	// 2. check db-collection capacity
	db2CollIDs := t.meta.ListAllAvailCollections(ctx)
	if err := t.checkMaxCollectionsPerDB(ctx, db2CollIDs); err != nil {
		return err
	}

	// 3. check total collection number
	totalCollections := 0
	for _, collIDs := range db2CollIDs {
		totalCollections += len(collIDs)
	}

	maxCollectionNum := Params.QuotaConfig.MaxCollectionNum.GetAsInt()
	if totalCollections >= maxCollectionNum {
		mlog.Warn(ctx, "unable to create collection because the number of collection has reached the limit", mlog.Int("max_collection_num", maxCollectionNum))
		return merr.WrapErrCollectionNumLimitExceeded(t.Req.GetDbName(), maxCollectionNum)
	}

	// 4. check collection * shard * partition
	var newPartNum int64 = 1
	if t.Req.GetNumPartitions() > 0 {
		newPartNum = t.Req.GetNumPartitions()
	}
	return checkGeneralCapacity(ctx, 1, newPartNum, t.Req.GetShardsNum(), t.Core)
}

// checkMaxCollectionsPerDB DB properties take precedence over quota configurations for max collections.
func (t *createCollectionTask) checkMaxCollectionsPerDB(ctx context.Context, db2CollIDs map[int64][]int64) error {
	Params := paramtable.Get()

	collIDs, ok := db2CollIDs[t.header.DbId]
	if !ok {
		mlog.Warn(ctx, "can not found DB ID", mlog.String("collection", t.Req.GetCollectionName()), mlog.String("dbName", t.Req.GetDbName()))
		return merr.WrapErrDatabaseNotFound(t.Req.GetDbName(), "failed to create collection")
	}

	db, err := t.meta.GetDatabaseByName(ctx, t.Req.GetDbName(), typeutil.MaxTimestamp)
	if err != nil {
		mlog.Warn(ctx, "can not found DB ID", mlog.String("collection", t.Req.GetCollectionName()), mlog.String("dbName", t.Req.GetDbName()))
		return merr.WrapErrDatabaseNotFound(t.Req.GetDbName(), "failed to create collection")
	}

	check := func(maxColNumPerDB int) error {
		if len(collIDs) >= maxColNumPerDB {
			mlog.Warn(ctx, "unable to create collection because the number of collection has reached the limit in DB", mlog.Int("maxCollectionNumPerDB", maxColNumPerDB))
			return merr.WrapErrCollectionNumLimitExceeded(t.Req.GetDbName(), maxColNumPerDB)
		}
		return nil
	}

	maxColNumPerDBStr := db.GetProperty(common.DatabaseMaxCollectionsKey)
	if maxColNumPerDBStr != "" {
		maxColNumPerDB, err := strconv.Atoi(maxColNumPerDBStr)
		if err != nil {
			mlog.Warn(ctx, "parse value of property fail", mlog.String("key", common.DatabaseMaxCollectionsKey),
				mlog.String("value", maxColNumPerDBStr), mlog.Err(err))
			return merr.WrapErrServiceInternalMsg("parse value of property fail, key:%s, value:%s", common.DatabaseMaxCollectionsKey, maxColNumPerDBStr)
		}
		return check(maxColNumPerDB)
	}

	maxColNumPerDB := Params.QuotaConfig.MaxCollectionNumPerDB.GetAsInt()
	return check(maxColNumPerDB)
}

func checkGeometryDefaultValue(value string) error {
	if _, err := common.ConvertWKTToWKB(value); err != nil {
		mlog.Warn(context.TODO(), "invalid default value for geometry field", mlog.Err(err))
		return merr.WrapErrParameterInvalidMsg("invalid default value for geometry field")
	}

	return nil
}

func hasSystemFields(schema *schemapb.CollectionSchema, systemFields []string) bool {
	for _, f := range schema.GetFields() {
		if funcutil.SliceContain(systemFields, f.GetName()) {
			return true
		}
	}
	return false
}

func (t *createCollectionTask) validateSchema(ctx context.Context, schema *schemapb.CollectionSchema) error {
	mlog.With(mlog.String("CollectionName", t.Req.CollectionName))
	if t.Req.GetCollectionName() != schema.GetName() {
		mlog.Error(ctx, "collection name not matches schema name", mlog.String("SchemaName", schema.Name))
		msg := fmt.Sprintf("collection name = %s, schema.Name=%s", t.Req.GetCollectionName(), schema.Name)
		return merr.WrapErrParameterInvalid("collection name matches schema name", "don't match", msg)
	}

	if err := checkFieldSchema(schema.GetFields()); err != nil {
		return err
	}

	// Validate default
	if err := timestamptz.CheckAndRewriteTimestampTzDefaultValue(schema); err != nil {
		return err
	}

	if err := checkStructArrayFieldSchema(schema.GetStructArrayFields()); err != nil {
		return err
	}

	// Note: this call mutates schema (sets nullable=true on every user field).
	// See NormalizeAndValidateExternalCollectionSchema for the rationale.
	if err := typeutil.NormalizeAndValidateExternalCollectionSchema(schema); err != nil {
		return err
	}
	if err := typeutil.ValidateTextRequiresStorageV3(schema, Params.CommonCfg.UseLoonFFI.GetAsBool()); err != nil {
		return merr.WrapErrParameterInvalidMsg("%s", err.Error())
	}

	// For external collections, validate the source URL scheme allowlist and
	// the JSON spec structure (extfs allowlist + format whitelist) at the
	// RootCoord side as well — defense in depth in case a request bypasses
	// proxy-side validation.
	// Skip validation for empty ExternalSource — IsExternalCollection allows
	// it (external fields without a concrete source), so the validator must
	// too. When a source is eventually supplied via alter/refresh, validation
	// runs there.
	if typeutil.IsExternalCollection(schema) && schema.GetExternalSource() != "" {
		if err := externalspec.ValidateSourceAndSpec(schema.GetExternalSource(), schema.GetExternalSpec()); err != nil {
			return err
		}
	}

	if hasSystemFields(schema, []string{RowIDFieldName, TimeStampFieldName, MetaFieldName, NamespaceFieldName}) {
		mlog.Error(ctx, "schema contains system field",
			mlog.String("RowIDFieldName", RowIDFieldName),
			mlog.String("TimeStampFieldName", TimeStampFieldName),
			mlog.String("MetaFieldName", MetaFieldName),
			mlog.String("NamespaceFieldName", NamespaceFieldName))
		msg := fmt.Sprintf("schema contains system field: %s, %s, %s, %s", RowIDFieldName, TimeStampFieldName, MetaFieldName, NamespaceFieldName)
		return merr.WrapErrParameterInvalid("schema don't contains system field", "contains", msg)
	}

	if err := validateStructArrayFieldDataType(schema.GetStructArrayFields()); err != nil {
		return err
	}

	// check analyzer was vaild
	analyzerInfos := make([]*querypb.AnalyzerInfo, 0)
	for _, field := range schema.GetFields() {
		err := validateAnalyzer(schema, field, &analyzerInfos)
		if err != nil {
			return err
		}
	}

	// validate analyzer params at any streaming node
	// and set file resource ids to schema
	if len(analyzerInfos) > 0 {
		err := retry.Do(ctx, func() error {
			if t.fileResourceObserver == nil {
				return nil
			}
			if err := t.fileResourceObserver.CheckAllQnReady(); err != nil {
				return err
			}
			return nil
		}, retry.Attempts(10), retry.Sleep(3*time.Second))
		if err != nil {
			return err
		}

		resp, err := t.mixCoord.ValidateAnalyzer(t.ctx, &querypb.ValidateAnalyzerRequest{
			AnalyzerInfos: analyzerInfos,
		})
		if err != nil {
			return err
		}

		if err := merr.Error(resp.GetStatus()); err != nil {
			return err
		}
		schema.FileResourceIds = resp.GetResourceIds()

		// Bind file resources to collection lifecycle: refCnt++ now, refCnt-- on
		// drop. Under ddLock, atomic with RemoveFileResource. See #48612.
		if len(schema.FileResourceIds) > 0 {
			if err := t.meta.IncFileResourceRefCnt(schema.FileResourceIds); err != nil {
				return err
			}
			t.heldFileResourceIds = schema.FileResourceIds
		}
	}

	return validateFieldDataType(schema.GetFields())
}

func (t *createCollectionTask) assignFieldAndFunctionID(schema *schemapb.CollectionSchema) error {
	idx := 0
	for _, field := range schema.GetFields() {
		field.FieldID = int64(idx + StartOfUserFieldID)
		idx++
	}

	for _, structArrayField := range schema.GetStructArrayFields() {
		structArrayField.FieldID = int64(idx + StartOfUserFieldID)
		idx++

		for _, field := range structArrayField.GetFields() {
			field.FieldID = int64(idx + StartOfUserFieldID)
			idx++
		}
	}

	return assignFunctionIDsFromFieldNames(schema)
}

// assignFunctionIDsFromFieldNames resolves function input/output field IDs
// after field IDs have been assigned or aligned from a source snapshot.
func assignFunctionIDsFromFieldNames(schema *schemapb.CollectionSchema) error {
	name2id := map[string]int64{}
	for _, field := range schema.GetFields() {
		name2id[field.GetName()] = field.GetFieldID()
	}
	for _, structArrayField := range schema.GetStructArrayFields() {
		for _, field := range structArrayField.GetFields() {
			name2id[field.GetName()] = field.GetFieldID()
		}
	}

	for fidx, function := range schema.GetFunctions() {
		function.InputFieldIds = make([]int64, len(function.InputFieldNames))
		function.Id = int64(fidx) + StartOfUserFunctionID
		for idx, name := range function.InputFieldNames {
			fieldId, ok := name2id[name]
			if !ok {
				return merr.WrapErrParameterInvalidMsg("input field %s of function %s not found", name, function.GetName())
			}
			function.InputFieldIds[idx] = fieldId
		}

		function.OutputFieldIds = make([]int64, len(function.OutputFieldNames))
		for idx, name := range function.OutputFieldNames {
			fieldId, ok := name2id[name]
			if !ok {
				return merr.WrapErrParameterInvalidMsg("output field %s of function %s not found", name, function.GetName())
			}
			function.OutputFieldIds[idx] = fieldId
		}
	}

	return nil
}

func (t *createCollectionTask) appendDynamicField(ctx context.Context, schema *schemapb.CollectionSchema) {
	if schema.EnableDynamicField {
		schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
			Name:        MetaFieldName,
			Description: "dynamic schema",
			DataType:    schemapb.DataType_JSON,
			IsDynamic:   true,
			Nullable:    true,
			DefaultValue: &schemapb.ValueField{
				Data: &schemapb.ValueField_BytesData{
					BytesData: []byte("{}"),
				},
			},
		})
		mlog.Info(ctx, "append dynamic field", mlog.String("collection", schema.Name))
	}
}

func (t *createCollectionTask) appendConsistecyLevel() {
	if ok, _ := getConsistencyLevel(t.Req.Properties...); ok {
		return
	}
	for _, p := range t.Req.Properties {
		if p.GetKey() == common.ConsistencyLevel {
			// if there's already a consistency level, overwrite it.
			p.Value = strconv.Itoa(int(t.Req.ConsistencyLevel))
			return
		}
	}
	// append consistency level into schema properties
	t.Req.Properties = append(t.Req.Properties, &commonpb.KeyValuePair{
		Key:   common.ConsistencyLevel,
		Value: strconv.Itoa(int(t.Req.ConsistencyLevel)),
	})
}

func (t *createCollectionTask) appendNamespaceShardingEnabled() error {
	if err := common.ValidateNamespaceShardingEnabled(t.Req.Properties...); err != nil {
		return err
	}
	if common.IsNamespaceShardingEnabledKeyExists(t.Req.Properties...) {
		return nil
	}
	t.Req.Properties = append(t.Req.Properties, &commonpb.KeyValuePair{
		Key:   common.NamespaceShardingEnabledKey,
		Value: "false",
	})
	return nil
}

func (t *createCollectionTask) handleNamespaceField(ctx context.Context, schema *schemapb.CollectionSchema) error {
	hasIsolation := hasIsolationProperty(t.Req.Properties...)
	_, err := typeutil.GetPartitionKeyFieldSchema(schema)
	hasPartitionKey := err == nil
	if !schema.GetEnableNamespace() {
		return nil
	}

	if common.IsNamespaceModePartition(t.Req.GetProperties()...) {
		if hasPartitionKey {
			return merr.WrapErrParameterInvalidMsg("namespace is not supported with partition key mode")
		}
		return nil
	}

	if typeutil.IsExternalCollection(schema) {
		return merr.WrapErrParameterInvalidMsg("external collection does not support namespace field")
	}

	if hasIsolation {
		iso, err := common.IsPartitionKeyIsolationKvEnabled(t.Req.Properties...)
		if err != nil {
			return err
		}
		if !iso {
			return merr.WrapErrCollectionIllegalSchema(t.Req.CollectionName,
				"isolation property is false when namespace enabled")
		}
	}

	if hasPartitionKey {
		return merr.WrapErrParameterInvalidMsg("namespace is not supported with partition key mode")
	}

	if common.IsNamespaceModePartition(t.Req.GetProperties()...) {
		return nil
	}

	schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
		Name:           common.NamespaceFieldName,
		IsPartitionKey: true,
		DataType:       schemapb.DataType_VarChar,
		TypeParams: []*commonpb.KeyValuePair{
			{Key: common.MaxLengthKey, Value: fmt.Sprintf("%d", paramtable.Get().ProxyCfg.MaxVarCharLength.GetAsInt())},
		},
	})
	t.Req.Properties = append(t.Req.Properties, &commonpb.KeyValuePair{
		Key:   common.PartitionKeyIsolationKey,
		Value: "true",
	})
	mlog.Info(ctx, "added namespace field",
		mlog.String("collectionName", t.Req.CollectionName),
		mlog.String("fieldName", common.NamespaceFieldName))
	return nil
}

func hasIsolationProperty(props ...*commonpb.KeyValuePair) bool {
	for _, p := range props {
		if p.GetKey() == common.PartitionKeyIsolationKey {
			return true
		}
	}
	return false
}

func (t *createCollectionTask) appendSysFields(schema *schemapb.CollectionSchema) {
	schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
		FieldID:      int64(RowIDField),
		Name:         RowIDFieldName,
		IsPrimaryKey: false,
		Description:  "row id",
		DataType:     schemapb.DataType_Int64,
	})
	schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
		FieldID:      int64(TimeStampField),
		Name:         TimeStampFieldName,
		IsPrimaryKey: false,
		Description:  "time stamp",
		DataType:     schemapb.DataType_Int64,
	})
}

// prepareMilvusTableSnapshotSchema validates the source snapshot and aligns
// target field IDs before normal create-collection field ID assignment runs.
func (t *createCollectionTask) prepareMilvusTableSnapshotSchema(ctx context.Context) error {
	schema := t.body.CollectionSchema
	if schema == nil || schema.GetExternalSource() == "" || schema.GetExternalSpec() == "" {
		return nil
	}
	// Validate before reading snapshot metadata so RootCoord keeps the same
	// external source boundary even if a request bypasses Proxy.
	if err := externalspec.ValidateSourceAndSpec(schema.GetExternalSource(), schema.GetExternalSpec()); err != nil {
		return err
	}
	spec, err := externalspec.ParseExternalSpec(schema.GetExternalSpec())
	if err != nil {
		return err
	}
	if spec.Format != externalspec.FormatMilvusTable {
		return nil
	}
	if t.preserveFieldID {
		// DDL replay carries the schema after milvus-table field-ID alignment.
		// Re-reading the source snapshot here would make RootCoord recovery
		// depend on the external bucket and credentials still being available.
		return nil
	}

	metadata, err := packed.ReadMilvusTableSnapshotMetadata(
		schema.GetExternalSource(),
		schema.GetExternalSpec(),
		createMilvusTableSnapshotStorageConfig(),
		packed.ExternalSpecContext{
			Source: schema.GetExternalSource(),
			Spec:   schema.GetExternalSpec(),
		},
	)
	if err != nil {
		return merr.Wrap(err, "read milvus-table snapshot metadata for schema alignment")
	}
	sourceSchema := metadata.GetCollection().GetSchema()
	if sourceSchema == nil {
		return merr.WrapErrParameterInvalidMsg("milvus-table snapshot metadata missing collection schema")
	}
	if typeutil.IsExternalCollection(sourceSchema) {
		// Avoid external-table chaining. A chained source would require refresh
		// and read paths to chase another collection's external source/storage
		// contract, which is not part of the milvus-table snapshot contract.
		return merr.WrapErrParameterInvalidMsg("milvus-table external collection cannot use an external collection snapshot as source")
	}
	if err := typeutil.ValidateMilvusTableSchemaIdentity(schema, sourceSchema, false); err != nil {
		return merr.Wrap(err, "milvus-table target schema must match source snapshot schema")
	}

	sourceFields := milvusTableSourceFieldsByName(sourceSchema)
	nextTargetOnlyFieldID := nextMilvusTableTargetOnlyFieldID(sourceSchema)
	for _, field := range schema.GetFields() {
		if field.GetName() == common.VirtualPKFieldName || typeutil.IsFunctionOutputField(schema, field) {
			// Milvus-table mapped fields must reuse source field IDs because
			// source manifests store physical columns by field ID. Target-only
			// fields, including virtual PK and target function outputs, are not
			// read from the source manifest and therefore need IDs outside the
			// source snapshot range.
			field.FieldID = nextTargetOnlyFieldID
			nextTargetOnlyFieldID++
			continue
		}
		if typeutil.IsExternalSystemOrVirtualField(field.GetName()) {
			continue
		}
		sourceField := sourceFields[field.GetExternalField()]
		if sourceField == nil {
			return merr.WrapErrParameterInvalidMsg("milvus-table target field %q maps to missing source field %q", field.GetName(), field.GetExternalField())
		}
		field.FieldID = sourceField.GetFieldID()
	}
	if err := assignFunctionIDsFromFieldNames(schema); err != nil {
		return merr.Wrap(err, "align milvus-table function field IDs")
	}
	t.preserveFieldID = true
	t.Req.Properties = upsertCreateCollectionProperty(t.Req.GetProperties(), util.PreserveFieldIdsKey, "true")

	mlog.Info(ctx, "aligned milvus-table external collection field IDs with source snapshot",
		mlog.String("collection", t.Req.GetCollectionName()),
		mlog.String("externalSource", schema.GetExternalSource()))
	return nil
}

// createMilvusTableSnapshotStorageConfig builds the local Milvus storage config
// used to read source snapshot metadata during create collection.
func createMilvusTableSnapshotStorageConfig() *indexpb.StorageConfig {
	params := paramtable.Get()
	if params.CommonCfg.StorageType.GetValue() == "local" {
		return &indexpb.StorageConfig{
			RootPath:    params.LocalStorageCfg.Path.GetValue(),
			StorageType: params.CommonCfg.StorageType.GetValue(),
		}
	}
	return &indexpb.StorageConfig{
		Address:           params.MinioCfg.Address.GetValue(),
		AccessKeyID:       params.MinioCfg.AccessKeyID.GetValue(),
		SecretAccessKey:   params.MinioCfg.SecretAccessKey.GetValue(),
		UseSSL:            params.MinioCfg.UseSSL.GetAsBool(),
		SslCACert:         params.MinioCfg.SslCACert.GetValue(),
		BucketName:        params.MinioCfg.BucketName.GetValue(),
		RootPath:          params.MinioCfg.RootPath.GetValue(),
		UseIAM:            params.MinioCfg.UseIAM.GetAsBool(),
		IAMEndpoint:       params.MinioCfg.IAMEndpoint.GetValue(),
		StorageType:       params.CommonCfg.StorageType.GetValue(),
		Region:            params.MinioCfg.Region.GetValue(),
		UseVirtualHost:    params.MinioCfg.UseVirtualHost.GetAsBool(),
		CloudProvider:     params.MinioCfg.CloudProvider.GetValue(),
		RequestTimeoutMs:  params.MinioCfg.RequestTimeoutMs.GetAsInt64(),
		GcpCredentialJSON: params.MinioCfg.GcpCredentialJSON.GetValue(),
		SslTlsMinVersion:  params.MinioCfg.SslTLSMinVersion.GetValue(),
		UseCrc32CChecksum: params.MinioCfg.UseCRC32C.GetAsBool(),
	}
}

// nextMilvusTableTargetOnlyFieldID returns the first field ID above all source
// snapshot IDs so target-only fields cannot collide with source columns.
func nextMilvusTableTargetOnlyFieldID(schemas ...*schemapb.CollectionSchema) int64 {
	next := int64(StartOfUserFieldID)
	for _, schema := range schemas {
		for _, field := range schema.GetFields() {
			if field.GetFieldID() >= next {
				next = field.GetFieldID() + 1
			}
		}
	}
	return next
}

// milvusTableSourceFieldsByName indexes source user fields that can be targets
// of ExternalField mappings.
func milvusTableSourceFieldsByName(schema *schemapb.CollectionSchema) map[string]*schemapb.FieldSchema {
	fields := make(map[string]*schemapb.FieldSchema, len(schema.GetFields()))
	for _, field := range schema.GetFields() {
		if typeutil.IsExternalSystemOrVirtualField(field.GetName()) {
			continue
		}
		fields[field.GetName()] = field
	}
	return fields
}

// upsertCreateCollectionProperty inserts or updates one create-collection
// property without disturbing the remaining request properties.
func upsertCreateCollectionProperty(properties []*commonpb.KeyValuePair, key, value string) []*commonpb.KeyValuePair {
	for _, property := range properties {
		if property.GetKey() == key {
			property.Value = value
			return properties
		}
	}
	return append(properties, &commonpb.KeyValuePair{Key: key, Value: value})
}

func (t *createCollectionTask) prepareSchema(ctx context.Context) error {
	if err := t.prepareMilvusTableSnapshotSchema(ctx); err != nil {
		return err
	}

	// if schema comes from restore snapshot
	preservedDynamicFieldID := int64(-1)
	preservedNamespaceFieldID := int64(-1)
	if t.preserveFieldID {
		mlog.Info(ctx, "preserve field IDs from schema during create collection", mlog.String("collection", t.Req.CollectionName))
		fields := make([]*schemapb.FieldSchema, 0)
		// filter out system fields
		for _, field := range t.body.CollectionSchema.Fields {
			if field.Name != RowIDFieldName && field.GetFieldID() == 0 {
				mlog.Info(context.TODO(), "field id 0 is not allowed when preserve field ids", mlog.String("field", field.Name))
				return merr.WrapErrParameterInvalidMsg("field id 0 is not allowed when preserve field ids, field: %s", field.Name)
			}

			if field.GetName() == MetaFieldName {
				preservedDynamicFieldID = field.GetFieldID()
				continue
			}
			if field.GetName() == NamespaceFieldName {
				preservedNamespaceFieldID = field.GetFieldID()
				continue
			}
			if field.GetName() == TimeStampFieldName || field.GetName() == RowIDFieldName {
				continue
			}
			fields = append(fields, field)
		}
		t.body.CollectionSchema.Fields = fields
	}

	if err := t.validateSchema(ctx, t.body.CollectionSchema); err != nil {
		return err
	}

	t.appendConsistecyLevel()
	if err := t.appendNamespaceShardingEnabled(); err != nil {
		return err
	}
	t.appendDynamicField(ctx, t.body.CollectionSchema)
	if err := common.ValidateNamespaceMode(t.Req.GetProperties()...); err != nil {
		return err
	}
	if err := t.handleNamespaceField(ctx, t.body.CollectionSchema); err != nil {
		return err
	}

	if t.preserveFieldID {
		// cause dynamic field is system field without internal id allocation
		// we need to restore its field id here
		for _, field := range t.body.CollectionSchema.Fields {
			if field.GetName() == MetaFieldName {
				field.FieldID = preservedDynamicFieldID
			}

			if field.GetName() == NamespaceFieldName {
				field.FieldID = preservedNamespaceFieldID
			}
		}
	} else {
		if err := t.assignFieldAndFunctionID(t.body.CollectionSchema); err != nil {
			return err
		}
	}
	if err := typeutil.ValidateExternalCollectionResolvedSchema(t.body.CollectionSchema); err != nil {
		return err
	}

	// Validate timezone
	tz, exist := funcutil.TryGetAttrByKeyFromRepeatedKV(common.TimezoneKey, t.Req.GetProperties())
	if exist && !timestamptz.IsTimezoneValid(tz) {
		return merr.WrapErrParameterInvalidMsg("unknown or invalid IANA Time Zone ID: %s", tz)
	}

	// Set properties for persistent
	t.body.CollectionSchema.Properties = updateMaxFieldIDProperty(t.Req.GetProperties(), maxAssignedFieldIDFromSchema(t.body.CollectionSchema))
	t.body.CollectionSchema.Version = 0
	t.appendSysFields(t.body.CollectionSchema)
	return nil
}

func (t *createCollectionTask) assignCollectionID() error {
	var err error
	t.header.CollectionId, err = t.idAllocator.AllocOne()
	t.body.CollectionID = t.header.CollectionId
	return err
}

func (t *createCollectionTask) assignPartitionIDs(ctx context.Context) error {
	Params := paramtable.Get()

	partitionNames := make([]string, 0, t.Req.GetNumPartitions())
	defaultPartitionName := Params.CommonCfg.DefaultPartitionName.GetValue()

	if _, err := typeutil.GetPartitionKeyFieldSchema(t.body.CollectionSchema); err == nil {
		// only when enabling partition key mode, we allow to create multiple partitions.
		partitionNums := t.Req.GetNumPartitions()
		// double check, default num of physical partitions should be greater than 0
		if partitionNums <= 0 {
			return merr.WrapErrParameterInvalidMsg("the specified partitions should be greater than 0 if partition key is used")
		}

		cfgMaxPartitionNum := Params.RootCoordCfg.MaxPartitionNum.GetAsInt64()
		if partitionNums > cfgMaxPartitionNum {
			return merr.WrapErrParameterInvalidMsg("partition number (%d) exceeds max configuration (%d), collection: %s",
				partitionNums, cfgMaxPartitionNum, t.Req.CollectionName)
		}

		for i := int64(0); i < partitionNums; i++ {
			partitionNames = append(partitionNames, fmt.Sprintf("%s_%d", defaultPartitionName, i))
		}
	} else {
		// compatible with old versions <= 2.2.8
		partitionNames = append(partitionNames, defaultPartitionName)
	}

	// allocate partition ids
	start, end, err := t.idAllocator.Alloc(uint32(len(partitionNames)))
	if err != nil {
		return err
	}
	t.header.PartitionIds = make([]int64, len(partitionNames))
	t.body.PartitionIDs = make([]int64, len(partitionNames))
	for i := start; i < end; i++ {
		t.header.PartitionIds[i-start] = i
		t.body.PartitionIDs[i-start] = i
	}
	t.body.PartitionNames = partitionNames

	mlog.Info(ctx, "assign partitions when create collection",
		mlog.String("collectionName", t.Req.GetCollectionName()),
		mlog.Int64s("partitionIds", t.header.PartitionIds),
		mlog.Strings("partitionNames", t.body.PartitionNames))
	return nil
}

func (t *createCollectionTask) assignChannels(ctx context.Context) error {
	vchannels, err := snmanager.StaticStreamingNodeManager.AllocVirtualChannels(ctx, balancer.AllocVChannelParam{
		CollectionID: t.header.GetCollectionId(),
		Num:          int(t.Req.GetShardsNum()),
	})
	if err != nil {
		return merr.Wrapf(err, "failed to allocate vchannels for collection %d (shards=%d)",
			t.header.GetCollectionId(), t.Req.GetShardsNum())
	}

	for _, vchannel := range vchannels {
		t.body.PhysicalChannelNames = append(t.body.PhysicalChannelNames, funcutil.ToPhysicalChannel(vchannel))
		t.body.VirtualChannelNames = append(t.body.VirtualChannelNames, vchannel)
	}
	return nil
}

func (t *createCollectionTask) Prepare(ctx context.Context) error {
	t.body.Base = &commonpb.MsgBase{
		MsgType: commonpb.MsgType_CreateCollection,
	}

	db, err := t.meta.GetDatabaseByName(ctx, t.Req.GetDbName(), typeutil.MaxTimestamp)
	if err != nil {
		return err
	}
	// set collection timezone
	properties := t.Req.GetProperties()
	_, ok := funcutil.TryGetAttrByKeyFromRepeatedKV(common.TimezoneKey, properties)
	if !ok {
		dbTz, ok2 := funcutil.TryGetAttrByKeyFromRepeatedKV(common.TimezoneKey, db.Properties)
		if !ok2 {
			dbTz = common.DefaultTimezone
		}
		timezoneKV := &commonpb.KeyValuePair{Key: common.TimezoneKey, Value: dbTz}
		t.Req.Properties = append(properties, timezoneKV)
	}

	t.Req.Properties = hookutil.TidyCollPropsByDBProps(t.Req.Properties, db.Properties)

	t.header.DbId = db.ID
	t.body.DbID = t.header.DbId
	if err := t.validate(ctx); err != nil {
		return err
	}

	if err := t.prepareSchema(ctx); err != nil {
		return err
	}

	if err := t.assignCollectionID(); err != nil {
		return err
	}

	if err := t.assignPartitionIDs(ctx); err != nil {
		return err
	}

	if err := t.assignChannels(ctx); err != nil {
		return err
	}

	return t.validateIfCollectionExists(ctx)
}

func (t *createCollectionTask) validateIfCollectionExists(ctx context.Context) error {
	// Check if the collection name duplicates an alias.
	if _, err := t.meta.DescribeAlias(ctx, t.Req.GetDbName(), t.Req.GetCollectionName(), typeutil.MaxTimestamp); err == nil {
		err2 := merr.WrapErrAsInputError(merr.WrapErrAliasCollectionNameConflict(t.Req.GetDbName(), t.Req.GetCollectionName(), "please choose a unique name"))
		mlog.Warn(ctx, "create collection failed", mlog.String("database", t.Req.GetDbName()), mlog.Err(err2))
		return err2
	}

	// Check if the collection already exists.
	existedCollInfo, err := t.meta.GetCollectionByName(ctx, t.Req.GetDbName(), t.Req.GetCollectionName(), typeutil.MaxTimestamp, false)
	if err == nil {
		newCollInfo := newCollectionModel(t.header, t.body, 0)
		if equal := existedCollInfo.Equal(*newCollInfo); !equal {
			return merr.WrapErrParameterInvalidMsg("create duplicate collection with different parameters, collection: %s", t.Req.GetCollectionName())
		}
		return errIgnoredCreateCollection
	}
	return nil
}

func validateMultiAnalyzerParams(params string, coll *schemapb.CollectionSchema, fieldSchema *schemapb.FieldSchema, infos *[]*querypb.AnalyzerInfo) error {
	var m map[string]json.RawMessage
	var analyzerMap map[string]json.RawMessage
	var mFileName string

	err := json.Unmarshal([]byte(params), &m)
	if err != nil {
		return err
	}

	mfield, ok := m["by_field"]
	if !ok {
		return merr.WrapErrParameterInvalidMsg("multi analyzer params now must set by_field to specify with field decide analyzer")
	}

	err = json.Unmarshal(mfield, &mFileName)
	if err != nil {
		return merr.WrapErrParameterInvalidMsg("multi analyzer params by_field must be string but now: %s", mfield)
	}

	// check field exist
	fieldExist := false
	for _, field := range coll.GetFields() {
		if field.GetName() == mFileName {
			// only support string field now
			if field.GetDataType() != schemapb.DataType_VarChar {
				return merr.WrapErrParameterInvalidMsg("multi analyzer params now only support by string field, but field %s is not string", field.GetName())
			}
			fieldExist = true
			break
		}
	}

	if !fieldExist {
		return merr.WrapErrParameterInvalidMsg("multi analyzer dependent field %s not exist in collection %s", string(mfield), coll.GetName())
	}

	if value, ok := m["alias"]; ok {
		mapping := map[string]string{}
		err = json.Unmarshal(value, &mapping)
		if err != nil {
			return merr.WrapErrParameterInvalidMsg("multi analyzer alias must be string map but now: %s", value)
		}
	}

	analyzers, ok := m["analyzers"]
	if !ok {
		return merr.WrapErrParameterInvalidMsg("multi analyzer params must set analyzers ")
	}

	err = json.Unmarshal(analyzers, &analyzerMap)
	if err != nil {
		return merr.WrapErrParameterInvalidMsg("unmarshal analyzers failed: %s", err)
	}

	hasDefault := false
	for name, bytes := range analyzerMap {
		*infos = append(*infos, &querypb.AnalyzerInfo{
			Name:   name,
			Field:  fieldSchema.GetName(),
			Params: string(bytes),
		})
		if name == "default" {
			hasDefault = true
		}
	}

	if !hasDefault {
		return merr.WrapErrParameterInvalidMsg("multi analyzer must set default analyzer for all unknown value")
	}
	return nil
}

func validateAnalyzer(collSchema *schemapb.CollectionSchema, fieldSchema *schemapb.FieldSchema, analyzerInfos *[]*querypb.AnalyzerInfo) error {
	h := typeutil.CreateFieldSchemaHelper(fieldSchema)
	if !h.EnableMatch() && !typeutil.IsBm25FunctionInputField(collSchema, fieldSchema) {
		return nil
	}

	if !h.EnableAnalyzer() {
		return merr.WrapErrParameterInvalidMsg("field %s which has enable_match or is input of BM25 function must also enable_analyzer", fieldSchema.Name)
	}

	if params, ok := h.GetMultiAnalyzerParams(); ok {
		if h.EnableMatch() {
			return merr.WrapErrParameterInvalidMsg("multi analyzer now only support for bm25, but now field %s enable match", fieldSchema.Name)
		}
		if h.HasAnalyzerParams() {
			return merr.WrapErrParameterInvalidMsg("field %s analyzer params should be none if has multi analyzer params", fieldSchema.Name)
		}

		return validateMultiAnalyzerParams(params, collSchema, fieldSchema, analyzerInfos)
	}

	for _, kv := range fieldSchema.GetTypeParams() {
		if kv.GetKey() == "analyzer_params" {
			*analyzerInfos = append(*analyzerInfos, &querypb.AnalyzerInfo{
				Field:  fieldSchema.GetName(),
				Params: kv.GetValue(),
			})
		}
	}
	// return nil when use default analyzer
	return nil
}