milvus-io--milvus/internal/proxy/task_index.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 proxy

import (
	"context"
	"fmt"
	"strings"

	"github.com/cockroachdb/errors"

	"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/types"
	"github.com/milvus-io/milvus/internal/util/indexparamcheck"
	"github.com/milvus-io/milvus/internal/util/vecindexmgr"
	"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/util/commonpbutil"
	"github.com/milvus-io/milvus/pkg/v3/util/funcutil"
	"github.com/milvus-io/milvus/pkg/v3/util/indexparams"
	"github.com/milvus-io/milvus/pkg/v3/util/merr"
	"github.com/milvus-io/milvus/pkg/v3/util/metric"
	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
)

const (
	CreateIndexTaskName           = "CreateIndexTask"
	AlterIndexTaskName            = "AlterIndexTask"
	DescribeIndexTaskName         = "DescribeIndexTask"
	DropIndexTaskName             = "DropIndexTask"
	GetIndexStateTaskName         = "GetIndexStateTask"
	GetIndexBuildProgressTaskName = "GetIndexBuildProgressTask"

	AutoIndexName = common.AutoIndexName
	DimKey        = common.DimKey
	IsSparseKey   = common.IsSparseKey

	RefineTypeKey = "refine_type"
)

// mapVectorMetricToEmbListMetric maps element-level vector metrics to EmbList-level metrics for ArrayOfVector.
// Autoindex configs use element-level metrics (e.g., IP, L2, COSINE), but ArrayOfVector fields
// require EmbList metrics (e.g., MaxSimIP, MaxSimL2, MaxSimCosine).
func mapVectorMetricToEmbListMetric(metricType string) string {
	switch strings.ToUpper(metricType) {
	case strings.ToUpper(metric.COSINE):
		return metric.MaxSimCosine
	case strings.ToUpper(metric.L2):
		return metric.MaxSimL2
	case strings.ToUpper(metric.IP):
		return metric.MaxSimIP
	case strings.ToUpper(metric.HAMMING):
		return metric.MaxSimHamming
	case strings.ToUpper(metric.JACCARD):
		return metric.MaxSimJaccard
	default:
		return metricType
	}
}

// adjustAutoIndexParamsByDataType adjusts autoindex params based on vector data type
// If data_type is bf16 and refine_type is fp16/fp32, adjust to BF16
// If data_type is fp16 and refine_type is bf16/fp32, adjust to FP16
// Other refine_type values (e.g., sq8) are not modified
func adjustAutoIndexParamsByDataType(config map[string]string, dataType schemapb.DataType) map[string]string {
	if config == nil {
		return config
	}
	refineType, hasRefine := config[RefineTypeKey]
	if !hasRefine {
		return config
	}

	refineTypeLower := strings.ToLower(refineType)
	var requiredRefineType string

	switch dataType {
	case schemapb.DataType_Float16Vector:
		// fp16 data requires fp16 refine, adjust if refine_type is bf16/fp32
		if refineTypeLower == "bf16" || refineTypeLower == "fp32" {
			requiredRefineType = "FP16"
		}
	case schemapb.DataType_BFloat16Vector:
		// bf16 data requires bf16 refine, adjust if refine_type is fp16/fp32
		if refineTypeLower == "fp16" || refineTypeLower == "fp32" {
			requiredRefineType = "BF16"
		}
	}

	if requiredRefineType == "" {
		return config
	}

	adjusted := make(map[string]string, len(config))
	for k, v := range config {
		adjusted[k] = v
	}
	adjusted[RefineTypeKey] = requiredRefineType
	return adjusted
}

func getDenseFloatAutoIndexParams(collectionProperties []*commonpb.KeyValuePair) map[string]string {
	// autoindex is enabled only for cloud instance.
	// and large_topk query mode is set in collection properties.
	// we will use large_topk index params for dense float vector index when these two conditions are met.
	if Params.AutoIndexConfig.Enable.GetAsBool() && common.IsQueryModeLargeTopK(collectionProperties...) {
		return Params.AutoIndexConfig.LargeTopKIndexParams.GetAsJSONMap()
	}

	return Params.AutoIndexConfig.IndexParams.GetAsJSONMap()
}

type createIndexTask struct {
	baseTask
	Condition
	req      *milvuspb.CreateIndexRequest
	ctx      context.Context
	mixCoord types.MixCoordClient
	result   *commonpb.Status

	isAutoIndex    bool
	newIndexParams []*commonpb.KeyValuePair
	newTypeParams  []*commonpb.KeyValuePair
	newExtraParams []*commonpb.KeyValuePair

	collectionID                     UniqueID
	functionSchema                   *schemapb.FunctionSchema
	fieldSchema                      *schemapb.FieldSchema
	userAutoIndexMetricTypeSpecified bool
	collectionProperties             []*commonpb.KeyValuePair
}

func (cit *createIndexTask) TraceCtx() context.Context {
	return cit.ctx
}

func (cit *createIndexTask) ID() UniqueID {
	return cit.req.GetBase().GetMsgID()
}

func (cit *createIndexTask) SetID(uid UniqueID) {
	cit.req.GetBase().MsgID = uid
}

func (cit *createIndexTask) Name() string {
	return CreateIndexTaskName
}

func (cit *createIndexTask) Type() commonpb.MsgType {
	return cit.req.GetBase().GetMsgType()
}

func (cit *createIndexTask) BeginTs() Timestamp {
	return cit.req.GetBase().GetTimestamp()
}

func (cit *createIndexTask) EndTs() Timestamp {
	return cit.req.GetBase().GetTimestamp()
}

func (cit *createIndexTask) SetTs(ts Timestamp) {
	cit.req.Base.Timestamp = ts
}

func (cit *createIndexTask) OnEnqueue() error {
	if cit.req.Base == nil {
		cit.req.Base = commonpbutil.NewMsgBase()
	}
	cit.req.Base.MsgType = commonpb.MsgType_CreateIndex
	cit.req.Base.SourceID = paramtable.GetNodeID()
	return nil
}

func wrapUserIndexParams(metricType string) []*commonpb.KeyValuePair {
	return []*commonpb.KeyValuePair{
		{
			Key:   common.IndexTypeKey,
			Value: AutoIndexName,
		},
		{
			Key:   common.MetricTypeKey,
			Value: metricType,
		},
	}
}

func (cit *createIndexTask) parseFunctionParamsToIndex(indexParamsMap map[string]string) error {
	if !cit.fieldSchema.GetIsFunctionOutput() {
		return nil
	}
	return indexparamcheck.FillFunctionOutputIndexParams(cit.functionSchema.GetType(), indexParamsMap)
}

func (cit *createIndexTask) parseIndexParams(ctx context.Context) error {
	cit.newExtraParams = cit.req.GetExtraParams()
	if err := indexparamcheck.ValidateIndexParamsSize(cit.newExtraParams...); err != nil {
		return err
	}

	isVecIndex := typeutil.IsVectorType(cit.fieldSchema.DataType)
	indexParamsMap, err := indexparamcheck.ExpandIndexParams(cit.req.GetExtraParams())
	if err != nil {
		return err
	}

	if jsonCastType, exist := indexParamsMap[common.JSONCastTypeKey]; exist {
		indexParamsMap[common.JSONCastTypeKey] = strings.ToUpper(strings.TrimSpace(jsonCastType))
	}
	if jsonCastFunction, exist := indexParamsMap[common.JSONCastFunctionKey]; exist {
		indexParamsMap[common.JSONCastFunctionKey] = strings.ToUpper(strings.TrimSpace(jsonCastFunction))
	}

	if err := indexparamcheck.ValidateIndexParamsMapSize(indexParamsMap); err != nil {
		return err
	}

	if cit.fieldSchema.GetDataType() == schemapb.DataType_Text {
		return merr.WrapErrParameterInvalidMsg("TEXT field does not support user-created scalar index")
	}

	if err := ValidateAutoIndexMmapConfig(isVecIndex, indexParamsMap); err != nil {
		return err
	}

	specifyIndexType, exist := indexParamsMap[common.IndexTypeKey]
	if exist && specifyIndexType != "" {
		// todo(SpadeA): mmap check for struct array index
		if err := indexparamcheck.ValidateMmapIndexParams(specifyIndexType, indexParamsMap); err != nil {
			mlog.Warn(ctx, "Invalid mmap type params", mlog.String(common.IndexTypeKey, specifyIndexType), mlog.Err(err))
			return merr.WrapErrParameterInvalidMsg("invalid mmap type params: %s", err.Error())
		}
		// todo(SpadeA): check for struct array index
		checker, err := indexparamcheck.GetIndexCheckerMgrInstance().GetChecker(specifyIndexType)
		// not enable hybrid index for user, used in milvus internally
		if err != nil || indexparamcheck.IsHYBRIDChecker(checker) {
			mlog.Warn(ctx, "Failed to get index checker", mlog.String(common.IndexTypeKey, specifyIndexType), mlog.Err(err))
			return merr.WrapErrParameterInvalid("valid index", fmt.Sprintf("invalid index type: %s", specifyIndexType))
		}
	}

	// Validate warmup policy if specified
	if err := indexparamcheck.ValidateWarmupIndexParams(indexParamsMap); err != nil {
		mlog.Warn(ctx, "Invalid warmup params", mlog.Err(err))
		return merr.WrapErrParameterInvalidMsg("invalid warmup params: %s", err.Error())
	}

	if !isVecIndex {
		specifyIndexType, exist := indexParamsMap[common.IndexTypeKey]
		autoIndexEnable := Params.AutoIndexConfig.ScalarAutoIndexEnable.GetAsBool()

		if autoIndexEnable || !exist || specifyIndexType == AutoIndexName {
			getPrimitiveIndexType := func(dataType schemapb.DataType) string {
				if typeutil.IsBoolType(dataType) {
					return Params.AutoIndexConfig.ScalarBoolIndexType.GetValue()
				} else if typeutil.IsIntegerType(dataType) {
					return Params.AutoIndexConfig.ScalarIntIndexType.GetValue()
				} else if typeutil.IsFloatingType(dataType) {
					return Params.AutoIndexConfig.ScalarFloatIndexType.GetValue()
				} else if typeutil.IsTimestamptzType(dataType) {
					return Params.AutoIndexConfig.ScalarTimestampTzIndexType.GetValue()
				}
				return Params.AutoIndexConfig.ScalarVarcharIndexType.GetValue()
			}

			indexType, err := func() (string, error) {
				dataType := cit.fieldSchema.DataType
				if typeutil.IsPrimitiveType(dataType) {
					return getPrimitiveIndexType(dataType), nil
				} else if typeutil.IsArrayType(dataType) {
					return getPrimitiveIndexType(cit.fieldSchema.ElementType), nil
				} else if typeutil.IsJSONType(dataType) {
					return Params.AutoIndexConfig.ScalarJSONIndexType.GetValue(), nil
				} else if typeutil.IsGeometryType(dataType) {
					return Params.AutoIndexConfig.ScalarGeometryIndexType.GetValue(), nil
				}
				return "", merr.WrapErrParameterInvalidMsg("create auto index on type:%s is not supported", dataType.String())
			}()
			if err != nil {
				return merr.WrapErrParameterInvalid("supported field", err.Error())
			}

			if typeutil.IsJSONType(cit.fieldSchema.DataType) && indexType == indexparamcheck.IndexHybrid {
				// Full JSON cast is only implemented by the inverted JSON
				// index; HYBRID supports typed JSON path indexes.
				if castType, ok := indexParamsMap[common.JSONCastTypeKey]; ok && strings.EqualFold(castType, "JSON") {
					indexType = indexparamcheck.IndexINVERTED
				}
			}
			indexParamsMap[common.IndexTypeKey] = indexType
			cit.isAutoIndex = true
		}
	} else {
		specifyIndexType, exist := indexParamsMap[common.IndexTypeKey]
		if Params.AutoIndexConfig.Enable.GetAsBool() { // `enable` only for cloud instance.
			mlog.Info(ctx, "create index trigger AutoIndex",
				mlog.String("original type", specifyIndexType),
				mlog.String("final type", Params.AutoIndexConfig.AutoIndexTypeName.GetValue()))

			metricType, metricTypeExist := indexParamsMap[common.MetricTypeKey]

			if typeutil.IsDenseFloatVectorType(cit.fieldSchema.DataType) ||
				(typeutil.IsArrayOfVectorType(cit.fieldSchema.DataType) && typeutil.IsDenseFloatVectorType(cit.fieldSchema.ElementType)) {
				autoIndexParams := getDenseFloatAutoIndexParams(cit.collectionProperties)
				// override float vector index params by autoindex
				// filter incompatible refine_type for fp16/bf16 vectors
				dataType := cit.fieldSchema.DataType
				if typeutil.IsArrayOfVectorType(cit.fieldSchema.DataType) {
					dataType = cit.fieldSchema.ElementType
				}
				autoIndexParams = adjustAutoIndexParamsByDataType(autoIndexParams, dataType)
				for k, v := range autoIndexParams {
					indexParamsMap[k] = v
				}
			} else if typeutil.IsSparseFloatVectorType(cit.fieldSchema.DataType) ||
				(typeutil.IsArrayOfVectorType(cit.fieldSchema.DataType) && typeutil.IsSparseFloatVectorType(cit.fieldSchema.ElementType)) {
				// override sparse float vector index params by autoindex
				for k, v := range Params.AutoIndexConfig.SparseIndexParams.GetAsJSONMap() {
					indexParamsMap[k] = v
				}
			} else if typeutil.IsBinaryVectorType(cit.fieldSchema.DataType) ||
				(typeutil.IsArrayOfVectorType(cit.fieldSchema.DataType) && typeutil.IsBinaryVectorType(cit.fieldSchema.ElementType)) {
				if metricTypeExist && funcutil.SliceContain(indexparamcheck.DeduplicateMetrics, metricType) {
					if !Params.AutoIndexConfig.EnableDeduplicateIndex.GetAsBool() {
						mlog.Warn(ctx, "Deduplicate index is not enabled, but metric type is deduplicate.")
						return merr.WrapErrParameterInvalidMsg("Deduplicate index is not enabled, but metric type is deduplicate.")
					}
					// override binary vector index params by autoindex deduplicate params
					for k, v := range Params.AutoIndexConfig.DeduplicateIndexParams.GetAsJSONMap() {
						indexParamsMap[k] = v
					}
				} else {
					// override binary vector index params by autoindex
					for k, v := range Params.AutoIndexConfig.BinaryIndexParams.GetAsJSONMap() {
						indexParamsMap[k] = v
					}
				}
			} else if typeutil.IsIntVectorType(cit.fieldSchema.DataType) ||
				(typeutil.IsArrayOfVectorType(cit.fieldSchema.DataType) && typeutil.IsIntVectorType(cit.fieldSchema.ElementType)) {
				// override int vector index params by autoindex
				for k, v := range Params.AutoIndexConfig.IntVectorIndexParams.GetAsJSONMap() {
					indexParamsMap[k] = v
				}
			}

			if metricTypeExist {
				// make the users' metric type first class citizen.
				indexParamsMap[common.MetricTypeKey] = metricType
				cit.userAutoIndexMetricTypeSpecified = true
			} else if typeutil.IsArrayOfVectorType(cit.fieldSchema.DataType) {
				// When user does not specify metric, autoindex config provides element-level metrics.
				// Map them to EmbList metrics since ArrayOfVector requires EmbList metrics.
				if m, ok := indexParamsMap[common.MetricTypeKey]; ok {
					indexParamsMap[common.MetricTypeKey] = mapVectorMetricToEmbListMetric(m)
				}
			}
		} else { // behavior change after 2.2.9, adapt autoindex logic here.
			useAutoIndex := func(autoIndexConfig map[string]string) {
				fields := make([]mlog.Field, 0, len(autoIndexConfig))
				for k, v := range autoIndexConfig {
					indexParamsMap[k] = v
					fields = append(fields, mlog.String(k, v))
				}
				mlog.Info(ctx, "AutoIndex triggered", fields...)
			}
			metricType, metricTypeExist := indexParamsMap[common.MetricTypeKey]

			handle := func(numberParams int, autoIndexConfig map[string]string) error {
				// empty case.
				if len(indexParamsMap) == numberParams {
					// though we already know there must be metric type, how to make this safer to avoid crash?
					metricType := autoIndexConfig[common.MetricTypeKey]
					cit.newExtraParams = wrapUserIndexParams(metricType)
					useAutoIndex(autoIndexConfig)
					return nil
				}

				if len(indexParamsMap) > numberParams+1 {
					return merr.WrapErrParameterInvalidMsg("only metric type can be passed when use AutoIndex")
				}

				if len(indexParamsMap) == numberParams+1 {
					if !metricTypeExist {
						return merr.WrapErrParameterInvalidMsg("only metric type can be passed when use AutoIndex")
					}

					// only metric type is passed.
					cit.newExtraParams = wrapUserIndexParams(metricType)
					useAutoIndex(autoIndexConfig)
					// make the users' metric type first class citizen.
					indexParamsMap[common.MetricTypeKey] = metricType
					cit.userAutoIndexMetricTypeSpecified = true
				}

				return nil
			}

			var config map[string]string
			if typeutil.IsDenseFloatVectorType(cit.fieldSchema.DataType) ||
				(typeutil.IsArrayOfVectorType(cit.fieldSchema.DataType) && typeutil.IsDenseFloatVectorType(cit.fieldSchema.ElementType)) {
				config = getDenseFloatAutoIndexParams(cit.collectionProperties)
				// filter incompatible refine_type for fp16/bf16 vectors
				dataType := cit.fieldSchema.DataType
				if typeutil.IsArrayOfVectorType(cit.fieldSchema.DataType) {
					dataType = cit.fieldSchema.ElementType
				}
				config = adjustAutoIndexParamsByDataType(config, dataType)
			} else if typeutil.IsSparseFloatVectorType(cit.fieldSchema.DataType) ||
				(typeutil.IsArrayOfVectorType(cit.fieldSchema.DataType) && typeutil.IsSparseFloatVectorType(cit.fieldSchema.ElementType)) {
				// override sparse float vector index params by autoindex
				config = Params.AutoIndexConfig.SparseIndexParams.GetAsJSONMap()
			} else if typeutil.IsBinaryVectorType(cit.fieldSchema.DataType) ||
				(typeutil.IsArrayOfVectorType(cit.fieldSchema.DataType) && typeutil.IsBinaryVectorType(cit.fieldSchema.ElementType)) {
				if metricTypeExist && funcutil.SliceContain(indexparamcheck.DeduplicateMetrics, metricType) {
					config = Params.AutoIndexConfig.DeduplicateIndexParams.GetAsJSONMap()
				} else {
					// override binary vector index params by autoindex
					config = Params.AutoIndexConfig.BinaryIndexParams.GetAsJSONMap()
				}
			} else if typeutil.IsIntVectorType(cit.fieldSchema.DataType) ||
				(typeutil.IsArrayOfVectorType(cit.fieldSchema.DataType) && typeutil.IsIntVectorType(cit.fieldSchema.ElementType)) {
				// override int vector index params by autoindex
				config = Params.AutoIndexConfig.IntVectorIndexParams.GetAsJSONMap()
			}
			if !exist {
				if err := handle(0, config); err != nil {
					return err
				}
			} else if specifyIndexType == AutoIndexName {
				if err := handle(1, config); err != nil {
					return err
				}
			}
			// When user does not specify metric, map autoindex config's element-level metrics to EmbList
			if !metricTypeExist && typeutil.IsArrayOfVectorType(cit.fieldSchema.DataType) {
				if m, ok := indexParamsMap[common.MetricTypeKey]; ok {
					indexParamsMap[common.MetricTypeKey] = mapVectorMetricToEmbListMetric(m)
				}
			}
		}

		// fill index param for Functions
		if err := cit.parseFunctionParamsToIndex(indexParamsMap); err != nil {
			return err
		}

		indexType, exist := indexParamsMap[common.IndexTypeKey]
		if !exist {
			return merr.WrapErrParameterMissingMsg("IndexType not specified")
		}
		//  index parameters defined in the YAML file are merged with the user-provided parameters during create stage
		if Params.KnowhereConfig.Enable.GetAsBool() {
			var err error
			indexParamsMap, err = Params.KnowhereConfig.MergeIndexParams(indexType, paramtable.BuildStage, indexParamsMap)
			if err != nil {
				return err
			}
		}
		if vecindexmgr.GetVecIndexMgrInstance().IsDiskANN(indexType) {
			err := indexparams.FillDiskIndexParams(Params, indexParamsMap)
			if err != nil {
				return err
			}
		}
		metricType, metricTypeExist := indexParamsMap[common.MetricTypeKey]
		if !metricTypeExist {
			return merr.WrapErrParameterInvalid("valid index params", "invalid index params", "metric type not set for vector index")
		}
		if typeutil.IsDenseFloatVectorType(cit.fieldSchema.DataType) {
			if !funcutil.SliceContain(indexparamcheck.FloatVectorMetrics, metricType) {
				return merr.WrapErrParameterInvalid("valid index params", "invalid index params", "float vector index does not support metric type: "+metricType)
			}
		} else if typeutil.IsSparseFloatVectorType(cit.fieldSchema.DataType) {
			if !funcutil.SliceContain(indexparamcheck.SparseFloatVectorMetrics, metricType) {
				return merr.WrapErrParameterInvalid("valid index params", "invalid index params", "only IP&BM25 is the supported metric type for sparse index")
			}
			if metricType == metric.BM25 && cit.functionSchema.GetType() != schemapb.FunctionType_BM25 {
				return merr.WrapErrParameterInvalid("valid index params", "invalid index params", "only BM25 Function output field support BM25 metric type")
			}
		} else if typeutil.IsBinaryVectorType(cit.fieldSchema.DataType) {
			if !funcutil.SliceContain(indexparamcheck.BinaryVectorMetrics, metricType) {
				return merr.WrapErrParameterInvalid("valid index params", "invalid index params", "binary vector index does not support metric type: "+metricType)
			}
		} else if typeutil.IsIntVectorType(cit.fieldSchema.DataType) {
			if !funcutil.SliceContain(indexparamcheck.IntVectorMetrics, metricType) {
				return merr.WrapErrParameterInvalid("valid index params", "invalid index params", "int vector index does not support metric type: "+metricType)
			}
		} else if typeutil.IsArrayOfVectorType(cit.fieldSchema.DataType) {
			if err := indexparamcheck.ValidateArrayOfVectorMetricType(cit.fieldSchema.ElementType, metricType); err != nil {
				return merr.WrapErrParameterInvalid("valid index params", "invalid index params", err.Error())
			}
		}
	}

	// autofill json path with field name if not specified for json index
	if typeutil.IsJSONType(cit.fieldSchema.DataType) {
		if _, exist := indexParamsMap[common.JSONPathKey]; !exist {
			indexParamsMap[common.JSONPathKey] = cit.req.FieldName
		}
	}

	if err := indexparamcheck.ValidateIndexParamsMapSize(indexParamsMap); err != nil {
		return err
	}

	err = checkTrain(ctx, cit.fieldSchema, indexParamsMap)
	if err != nil {
		// checkTrain may propagate errors from indexparamcheck (not yet
		// merr-standardized). Already-merr errors (leaves / fillDimension /
		// the merr-returning checkers) pass through with their real code;
		// only legacy plain errors get wrapped once into ParameterInvalid.
		if merr.IsMilvusError(err) {
			return err
		}
		return merr.WrapErrParameterInvalid("valid index params", "invalid index params", err.Error())
	}

	typeParams := cit.fieldSchema.GetTypeParams()
	typeParamsMap := make(map[string]string)
	for _, pair := range typeParams {
		typeParamsMap[pair.Key] = pair.Value
	}

	for k, v := range indexParamsMap {
		// Currently, it is required that type_params and index_params do not have same keys.
		if k == DimKey || k == common.MaxLengthKey {
			delete(indexParamsMap, k)
			continue
		}
		cit.newIndexParams = append(cit.newIndexParams, &commonpb.KeyValuePair{Key: k, Value: v})
	}

	for k, v := range typeParamsMap {
		if _, ok := indexParamsMap[k]; ok {
			continue
		}
		cit.newTypeParams = append(cit.newTypeParams, &commonpb.KeyValuePair{Key: k, Value: v})
	}

	return nil
}

func (cit *createIndexTask) getIndexedFieldAndFunction(ctx context.Context) error {
	schema, err := globalMetaCache.GetCollectionSchema(ctx, cit.req.GetDbName(), cit.req.GetCollectionName())
	if err != nil {
		mlog.Error(ctx, "failed to get collection schema", mlog.Err(err))
		return merr.Wrap(err, "failed to get collection schema")
	}

	field, err := schema.schemaHelper.GetFieldFromNameDefaultJSON(cit.req.GetFieldName())
	if err != nil {
		mlog.Error(ctx, "create index on non-exist field", mlog.Err(err))
		return merr.WrapErrParameterInvalidMsg("cannot create index on non-exist field: %s", cit.req.GetFieldName())
	}

	if field.IsFunctionOutput {
		function, err := schema.schemaHelper.GetFunctionByOutputField(field)
		if err != nil {
			mlog.Error(ctx, "create index failed, cannot find function of function output field", mlog.Err(err))
			return merr.WrapErrParameterInvalidMsg("create index failed, cannot find function of function output field: %s", cit.req.GetFieldName())
		}
		cit.functionSchema = function
	}
	cit.fieldSchema = field
	return nil
}

func checkTrain(ctx context.Context, field *schemapb.FieldSchema, indexParams map[string]string) error {
	indexType := indexParams[common.IndexTypeKey]

	if indexType == indexparamcheck.IndexHybrid {
		_, exist := indexParams[common.BitmapCardinalityLimitKey]
		if !exist {
			indexParams[common.BitmapCardinalityLimitKey] = paramtable.Get().AutoIndexConfig.BitmapCardinalityLimit.GetValue()
		}
		// Does not allow the user to specify the index type for hybrid index. This is by design.
		indexParams[common.HybridLowCardinalityIndexTypeKey] = paramtable.Get().DataCoordCfg.HybridIndexLowCardinalityIndexType.GetValue()
		indexParams[common.HybridHighCardinalityIndexTypeKey] = paramtable.Get().DataCoordCfg.HybridIndexHighCardinalityIndexType.GetValue()
	}

	if _, err := indexparamcheck.GetIndexCheckerMgrInstance().GetChecker(indexType); err != nil {
		mlog.Warn(ctx, "Failed to get index checker", mlog.String(common.IndexTypeKey, indexType))
		return merr.WrapErrParameterInvalidMsg("invalid index type: %s", indexType)
	}

	// For ArrayOfVector with non-EmbList metrics (e.g., COSINE, L2, IP), each embedding
	// in the array is indexed independently as a regular vector. The index only needs to
	// support the element vector type, not the EmbeddingList capability.
	// Resolve the effective data type used for index compatibility checks.
	effectiveDataType := field.DataType
	effectiveElementType := field.ElementType
	if typeutil.IsArrayOfVectorType(field.DataType) &&
		!funcutil.SliceContain(indexparamcheck.EmbListMetrics, indexParams[common.MetricTypeKey]) {
		effectiveDataType = field.ElementType
		effectiveElementType = schemapb.DataType_None
	}

	if typeutil.IsVectorType(field.DataType) && indexType != indexparamcheck.AutoIndex {
		exist := CheckVecIndexWithDataTypeExist(indexType, effectiveDataType, effectiveElementType)
		if !exist {
			return merr.WrapErrParameterInvalidMsg("data type %s can't build with this index %s", schemapb.DataType_name[int32(field.GetDataType())], indexType)
		}
	}

	if err := indexparamcheck.ValidateFieldIndexParams(field, indexParams); err != nil {
		mlog.Info(ctx, "create index with invalid parameters", mlog.Err(err), mlog.String("data_type", field.GetDataType().String()))
		return err
	}

	return nil
}

func (cit *createIndexTask) PreExecute(ctx context.Context) error {
	collName := cit.req.GetCollectionName()

	collID, err := globalMetaCache.GetCollectionID(ctx, cit.req.GetDbName(), collName)
	if err != nil {
		return err
	}
	cit.collectionID = collID

	collInfo, err := globalMetaCache.GetCollectionInfo(ctx, cit.req.GetDbName(), cit.req.GetCollectionName(), cit.collectionID)
	if err != nil {
		return err
	}
	cit.collectionProperties = collInfo.properties

	if err = validateIndexName(cit.req.GetIndexName()); err != nil {
		return err
	}

	err = cit.getIndexedFieldAndFunction(ctx)
	if err != nil {
		return err
	}

	// check index param, not accurate, only some static rules
	err = cit.parseIndexParams(ctx)
	if err != nil {
		return err
	}

	return nil
}

func (cit *createIndexTask) Execute(ctx context.Context) error {
	mlog.Info(ctx, "proxy create index", mlog.Int64("collectionID", cit.collectionID), mlog.Int64("fieldID", cit.fieldSchema.GetFieldID()),
		mlog.String("indexName", cit.req.GetIndexName()), mlog.Any("typeParams", cit.fieldSchema.GetTypeParams()),
		mlog.Any("indexParams", cit.req.GetExtraParams()),
		mlog.Any("newExtraParams", cit.newExtraParams),
	)

	var err error
	req := &indexpb.CreateIndexRequest{
		CollectionID:                     cit.collectionID,
		FieldID:                          cit.fieldSchema.GetFieldID(),
		IndexName:                        cit.req.GetIndexName(),
		TypeParams:                       cit.newTypeParams,
		IndexParams:                      cit.newIndexParams,
		IsAutoIndex:                      cit.isAutoIndex,
		UserIndexParams:                  cit.newExtraParams,
		Timestamp:                        cit.BeginTs(),
		UserAutoindexMetricTypeSpecified: cit.userAutoIndexMetricTypeSpecified,
	}
	cit.result, err = cit.mixCoord.CreateIndex(ctx, req)
	if err = merr.CheckRPCCall(cit.result, err); err != nil {
		return err
	}
	return nil
}

func (cit *createIndexTask) PostExecute(ctx context.Context) error {
	return nil
}

type alterIndexTask struct {
	baseTask
	Condition
	req      *milvuspb.AlterIndexRequest
	ctx      context.Context
	mixCoord types.MixCoordClient
	result   *commonpb.Status

	collectionID UniqueID
}

func (t *alterIndexTask) TraceCtx() context.Context {
	return t.ctx
}

func (t *alterIndexTask) ID() UniqueID {
	return t.req.GetBase().GetMsgID()
}

func (t *alterIndexTask) SetID(uid UniqueID) {
	t.req.GetBase().MsgID = uid
}

func (t *alterIndexTask) Name() string {
	return AlterIndexTaskName
}

func (t *alterIndexTask) Type() commonpb.MsgType {
	return t.req.GetBase().GetMsgType()
}

func (t *alterIndexTask) BeginTs() Timestamp {
	return t.req.GetBase().GetTimestamp()
}

func (t *alterIndexTask) EndTs() Timestamp {
	return t.req.GetBase().GetTimestamp()
}

func (t *alterIndexTask) SetTs(ts Timestamp) {
	t.req.Base.Timestamp = ts
}

func (t *alterIndexTask) OnEnqueue() error {
	if t.req.Base == nil {
		t.req.Base = commonpbutil.NewMsgBase()
	}
	t.req.Base.MsgType = commonpb.MsgType_AlterIndex
	t.req.Base.SourceID = paramtable.GetNodeID()
	return nil
}

func (t *alterIndexTask) PreExecute(ctx context.Context) error {
	if len(t.req.GetDeleteKeys()) > 0 && len(t.req.GetExtraParams()) > 0 {
		return merr.WrapErrParameterInvalidMsg("cannot provide both DeleteKeys and ExtraParams")
	}

	if len(t.req.GetExtraParams()) > 0 {
		for _, param := range t.req.GetExtraParams() {
			if !indexparams.IsConfigableIndexParam(param.GetKey()) {
				return merr.WrapErrParameterInvalidMsg("%s is not a configable index property", param.GetKey())
			}
		}
		// Validate warmup policy if specified
		indexParamsMap := funcutil.KeyValuePair2Map(t.req.GetExtraParams())
		if err := indexparamcheck.ValidateWarmupIndexParams(indexParamsMap); err != nil {
			return merr.WrapErrParameterInvalidMsg("invalid warmup params: %s", err.Error())
		}
	} else if len(t.req.GetDeleteKeys()) > 0 {
		for _, param := range t.req.GetDeleteKeys() {
			if !indexparams.IsConfigableIndexParam(param) {
				return merr.WrapErrParameterInvalidMsg("%s is not a configable index property", param)
			}
		}
	}

	collName := t.req.GetCollectionName()

	collection, err := globalMetaCache.GetCollectionID(ctx, t.req.GetDbName(), collName)
	if err != nil {
		return err
	}
	t.collectionID = collection

	if len(t.req.GetIndexName()) == 0 {
		return merr.WrapErrParameterInvalidMsg("index name is empty")
	}

	// TODO fubang should implement it when the alter index is reconstructed
	// typeParams := funcutil.KeyValuePair2Map(t.req.GetExtraParams())
	// if err = ValidateAutoIndexMmapConfig(typeParams); err != nil {
	// 	return err
	// }

	loaded, err := isCollectionLoaded(ctx, t.mixCoord, collection)
	if err != nil {
		return err
	}
	if loaded {
		return merr.WrapErrCollectionLoaded(collName, "can't alter index on loaded collection, please release the collection first")
	}

	return nil
}

func (t *alterIndexTask) Execute(ctx context.Context) error {
	log := mlog.With(
		mlog.String("collection", t.req.GetCollectionName()),
		mlog.String("indexName", t.req.GetIndexName()),
		mlog.Any("params", t.req.GetExtraParams()),
		mlog.Any("deletekeys", t.req.GetDeleteKeys()),
	)

	log.Info(ctx, "alter index")

	var err error
	req := &indexpb.AlterIndexRequest{
		CollectionID: t.collectionID,
		IndexName:    t.req.GetIndexName(),
		Params:       t.req.GetExtraParams(),
		DeleteKeys:   t.req.GetDeleteKeys(),
	}
	t.result, err = t.mixCoord.AlterIndex(ctx, req)
	if err = merr.CheckRPCCall(t.result, err); err != nil {
		return err
	}
	return nil
}

func (t *alterIndexTask) PostExecute(ctx context.Context) error {
	return nil
}

type describeIndexTask struct {
	baseTask
	Condition
	*milvuspb.DescribeIndexRequest
	ctx      context.Context
	mixCoord types.MixCoordClient
	result   *milvuspb.DescribeIndexResponse

	collectionID UniqueID
}

func (dit *describeIndexTask) TraceCtx() context.Context {
	return dit.ctx
}

func (dit *describeIndexTask) ID() UniqueID {
	return dit.Base.MsgID
}

func (dit *describeIndexTask) SetID(uid UniqueID) {
	dit.Base.MsgID = uid
}

func (dit *describeIndexTask) Name() string {
	return DescribeIndexTaskName
}

func (dit *describeIndexTask) Type() commonpb.MsgType {
	return dit.Base.MsgType
}

func (dit *describeIndexTask) BeginTs() Timestamp {
	return dit.Base.Timestamp
}

func (dit *describeIndexTask) EndTs() Timestamp {
	return dit.Base.Timestamp
}

func (dit *describeIndexTask) SetTs(ts Timestamp) {
	dit.Base.Timestamp = ts
}

func (dit *describeIndexTask) OnEnqueue() error {
	dit.Base = commonpbutil.NewMsgBase()
	dit.Base.MsgType = commonpb.MsgType_DescribeIndex
	dit.Base.SourceID = paramtable.GetNodeID()
	return nil
}

func (dit *describeIndexTask) PreExecute(ctx context.Context) error {
	if err := validateCollectionName(dit.CollectionName); err != nil {
		return err
	}

	collID, err := globalMetaCache.GetCollectionID(ctx, dit.GetDbName(), dit.CollectionName)
	if err != nil {
		return err
	}
	dit.collectionID = collID
	return nil
}

func (dit *describeIndexTask) Execute(ctx context.Context) error {
	schema, err := globalMetaCache.GetCollectionSchema(ctx, dit.GetDbName(), dit.GetCollectionName())
	if err != nil {
		mlog.Error(ctx, "failed to get collection schema", mlog.Err(err))
		return merr.Wrap(err, "failed to get collection schema")
	}

	resp, err := dit.mixCoord.DescribeIndex(ctx, &indexpb.DescribeIndexRequest{CollectionID: dit.collectionID, IndexName: dit.IndexName, Timestamp: dit.Timestamp})
	if err != nil {
		return err
	}

	dit.result = &milvuspb.DescribeIndexResponse{}
	dit.result.Status = resp.GetStatus()
	err = merr.Error(resp.GetStatus())
	if err != nil {
		if errors.Is(err, merr.ErrIndexNotFound) && len(dit.GetIndexName()) == 0 {
			err = merr.WrapErrIndexNotFoundForCollection(dit.GetCollectionName())
			dit.result.Status = merr.Status(err)
		}
		return err
	}
	for _, indexInfo := range resp.IndexInfos {
		field, err := schema.schemaHelper.GetFieldFromID(indexInfo.FieldID)
		if err != nil {
			mlog.Error(ctx, "failed to get collection field", mlog.Err(err))
			return merr.WrapErrParameterInvalidMsg("failed to get collection field: %d", indexInfo.FieldID)
		}
		params := indexInfo.GetUserIndexParams()
		if params == nil {
			metricType, err := funcutil.GetAttrByKeyFromRepeatedKV(MetricTypeKey, indexInfo.GetIndexParams())
			if err == nil {
				params = wrapUserIndexParams(metricType)
			}
		}
		fieldName := field.Name
		if field.IsDynamic {
			jsonPath, err := funcutil.GetAttrByKeyFromRepeatedKV(common.JSONPathKey, indexInfo.GetIndexParams())
			if err != nil {
				mlog.Warn(ctx, "failed to get json path for dynamic field", mlog.Err(err))
			} else if jsonPath != "" {
				// Skip leading "/" and find next "/" to get first path segment
				trimmedPath := strings.TrimPrefix(jsonPath, "/")
				slashIndex := strings.Index(trimmedPath, "/")
				if slashIndex == -1 {
					fieldName = trimmedPath // Use full remaining path if no more "/"
				} else {
					fieldName = trimmedPath[:slashIndex]
				}
				// Unescape JSON Pointer path: ~1 -> / and ~0 -> ~
				fieldName = strings.ReplaceAll(fieldName, "~1", "/")
				fieldName = strings.ReplaceAll(fieldName, "~0", "~")
			}
		}
		desc := &milvuspb.IndexDescription{
			IndexName:            indexInfo.GetIndexName(),
			IndexID:              indexInfo.GetIndexID(),
			FieldName:            fieldName,
			Params:               params,
			IndexedRows:          indexInfo.GetIndexedRows(),
			TotalRows:            indexInfo.GetTotalRows(),
			PendingIndexRows:     indexInfo.GetPendingIndexRows(),
			State:                indexInfo.GetState(),
			IndexStateFailReason: indexInfo.GetIndexStateFailReason(),
			MinIndexVersion:      indexInfo.GetMinIndexVersion(),
			MaxIndexVersion:      indexInfo.GetMaxIndexVersion(),
		}
		dit.result.IndexDescriptions = append(dit.result.IndexDescriptions, desc)
	}
	return err
}

func (dit *describeIndexTask) PostExecute(ctx context.Context) error {
	return nil
}

type getIndexStatisticsTask struct {
	baseTask
	Condition
	*milvuspb.GetIndexStatisticsRequest
	ctx      context.Context
	mixCoord types.MixCoordClient
	result   *milvuspb.GetIndexStatisticsResponse

	nodeID       int64
	collectionID UniqueID
}

func (dit *getIndexStatisticsTask) TraceCtx() context.Context {
	return dit.ctx
}

func (dit *getIndexStatisticsTask) ID() UniqueID {
	return dit.Base.MsgID
}

func (dit *getIndexStatisticsTask) SetID(uid UniqueID) {
	dit.Base.MsgID = uid
}

func (dit *getIndexStatisticsTask) Name() string {
	return DescribeIndexTaskName
}

func (dit *getIndexStatisticsTask) Type() commonpb.MsgType {
	return dit.Base.MsgType
}

func (dit *getIndexStatisticsTask) BeginTs() Timestamp {
	return dit.Base.Timestamp
}

func (dit *getIndexStatisticsTask) EndTs() Timestamp {
	return dit.Base.Timestamp
}

func (dit *getIndexStatisticsTask) SetTs(ts Timestamp) {
	dit.Base.Timestamp = ts
}

func (dit *getIndexStatisticsTask) OnEnqueue() error {
	dit.Base = commonpbutil.NewMsgBase()
	dit.Base.MsgType = commonpb.MsgType_GetIndexStatistics
	dit.Base.SourceID = paramtable.GetNodeID()
	return nil
}

func (dit *getIndexStatisticsTask) PreExecute(ctx context.Context) error {
	if err := validateCollectionName(dit.CollectionName); err != nil {
		return err
	}

	collID, err := globalMetaCache.GetCollectionID(ctx, dit.GetDbName(), dit.CollectionName)
	if err != nil {
		return err
	}
	dit.collectionID = collID
	return nil
}

func (dit *getIndexStatisticsTask) Execute(ctx context.Context) error {
	schema, err := globalMetaCache.GetCollectionSchema(ctx, dit.GetDbName(), dit.GetCollectionName())
	if err != nil {
		mlog.Error(ctx, "failed to get collection schema", mlog.String("collection_name", dit.GetCollectionName()), mlog.Err(err))
		return merr.Wrap(err, "failed to get collection schema")
	}
	schemaHelper := schema.schemaHelper

	resp, err := dit.mixCoord.GetIndexStatistics(ctx, &indexpb.GetIndexStatisticsRequest{
		CollectionID: dit.collectionID, IndexName: dit.IndexName,
	})
	if err := merr.CheckRPCCall(resp, err); err != nil {
		return err
	}
	dit.result = &milvuspb.GetIndexStatisticsResponse{}
	dit.result.Status = resp.GetStatus()
	for _, indexInfo := range resp.IndexInfos {
		field, err := schemaHelper.GetFieldFromID(indexInfo.FieldID)
		if err != nil {
			mlog.Error(ctx, "failed to get collection field", mlog.Int64("field_id", indexInfo.FieldID), mlog.Err(err))
			return merr.WrapErrParameterInvalidMsg("failed to get collection field: %d", indexInfo.FieldID)
		}
		params := indexInfo.GetUserIndexParams()
		if params == nil {
			params = indexInfo.GetIndexParams()
		}
		desc := &milvuspb.IndexDescription{
			IndexName:            indexInfo.GetIndexName(),
			IndexID:              indexInfo.GetIndexID(),
			FieldName:            field.Name,
			Params:               params,
			IndexedRows:          indexInfo.GetIndexedRows(),
			TotalRows:            indexInfo.GetTotalRows(),
			State:                indexInfo.GetState(),
			IndexStateFailReason: indexInfo.GetIndexStateFailReason(),
			MinIndexVersion:      indexInfo.GetMinIndexVersion(),
			MaxIndexVersion:      indexInfo.GetMaxIndexVersion(),
		}
		dit.result.IndexDescriptions = append(dit.result.IndexDescriptions, desc)
	}
	return err
}

func (dit *getIndexStatisticsTask) PostExecute(ctx context.Context) error {
	return nil
}

type dropIndexTask struct {
	baseTask
	Condition
	ctx context.Context
	*milvuspb.DropIndexRequest
	mixCoord types.MixCoordClient
	result   *commonpb.Status

	collectionID UniqueID
}

func (dit *dropIndexTask) TraceCtx() context.Context {
	return dit.ctx
}

func (dit *dropIndexTask) ID() UniqueID {
	return dit.Base.MsgID
}

func (dit *dropIndexTask) SetID(uid UniqueID) {
	dit.Base.MsgID = uid
}

func (dit *dropIndexTask) Name() string {
	return DropIndexTaskName
}

func (dit *dropIndexTask) Type() commonpb.MsgType {
	return dit.Base.MsgType
}

func (dit *dropIndexTask) BeginTs() Timestamp {
	return dit.Base.Timestamp
}

func (dit *dropIndexTask) EndTs() Timestamp {
	return dit.Base.Timestamp
}

func (dit *dropIndexTask) SetTs(ts Timestamp) {
	dit.Base.Timestamp = ts
}

func (dit *dropIndexTask) OnEnqueue() error {
	if dit.Base == nil {
		dit.Base = commonpbutil.NewMsgBase()
	}
	dit.Base.MsgType = commonpb.MsgType_DropIndex
	dit.Base.SourceID = paramtable.GetNodeID()
	return nil
}

func (dit *dropIndexTask) PreExecute(ctx context.Context) error {
	collID, err := globalMetaCache.GetCollectionID(ctx, dit.GetDbName(), dit.CollectionName)
	if err != nil {
		return err
	}
	dit.collectionID = collID

	return nil
}

func (dit *dropIndexTask) Execute(ctx context.Context) error {
	ctxLog := mlog.With()
	ctxLog.Info(ctx, "proxy drop index", mlog.Int64("collID", dit.collectionID),
		mlog.String("field_name", dit.FieldName),
		mlog.String("index_name", dit.IndexName),
		mlog.String("db_name", dit.DbName),
	)

	var err error
	dit.result, err = dit.mixCoord.DropIndex(ctx, &indexpb.DropIndexRequest{
		CollectionID: dit.collectionID,
		PartitionIDs: nil,
		IndexName:    dit.IndexName,
		DropAll:      false,
	})
	if err = merr.CheckRPCCall(dit.result, err); err != nil {
		ctxLog.Warn(ctx, "drop index failed", mlog.Err(err))
		return err
	}
	return nil
}

func (dit *dropIndexTask) PostExecute(ctx context.Context) error {
	return nil
}

// Deprecated: use describeIndexTask instead
type getIndexBuildProgressTask struct {
	baseTask
	Condition
	*milvuspb.GetIndexBuildProgressRequest
	ctx      context.Context
	mixCoord types.MixCoordClient
	result   *milvuspb.GetIndexBuildProgressResponse

	collectionID UniqueID
}

func (gibpt *getIndexBuildProgressTask) TraceCtx() context.Context {
	return gibpt.ctx
}

func (gibpt *getIndexBuildProgressTask) ID() UniqueID {
	return gibpt.Base.MsgID
}

func (gibpt *getIndexBuildProgressTask) SetID(uid UniqueID) {
	gibpt.Base.MsgID = uid
}

func (gibpt *getIndexBuildProgressTask) Name() string {
	return GetIndexBuildProgressTaskName
}

func (gibpt *getIndexBuildProgressTask) Type() commonpb.MsgType {
	return gibpt.Base.MsgType
}

func (gibpt *getIndexBuildProgressTask) BeginTs() Timestamp {
	return gibpt.Base.Timestamp
}

func (gibpt *getIndexBuildProgressTask) EndTs() Timestamp {
	return gibpt.Base.Timestamp
}

func (gibpt *getIndexBuildProgressTask) SetTs(ts Timestamp) {
	gibpt.Base.Timestamp = ts
}

func (gibpt *getIndexBuildProgressTask) OnEnqueue() error {
	gibpt.Base = commonpbutil.NewMsgBase()
	gibpt.Base.MsgType = commonpb.MsgType_GetIndexBuildProgress
	gibpt.Base.SourceID = paramtable.GetNodeID()
	return nil
}

func (gibpt *getIndexBuildProgressTask) PreExecute(ctx context.Context) error {
	if err := validateCollectionName(gibpt.CollectionName); err != nil {
		return err
	}

	return nil
}

func (gibpt *getIndexBuildProgressTask) Execute(ctx context.Context) error {
	collectionName := gibpt.CollectionName
	collectionID, err := globalMetaCache.GetCollectionID(ctx, gibpt.GetDbName(), collectionName)
	if err != nil { // err is not nil if collection not exists
		return err
	}
	gibpt.collectionID = collectionID

	resp, err := gibpt.mixCoord.GetIndexBuildProgress(ctx, &indexpb.GetIndexBuildProgressRequest{
		CollectionID: collectionID,
		IndexName:    gibpt.IndexName,
	})
	if err = merr.CheckRPCCall(resp, err); err != nil {
		return err
	}

	gibpt.result = &milvuspb.GetIndexBuildProgressResponse{
		Status:      resp.Status,
		TotalRows:   resp.GetTotalRows(),
		IndexedRows: resp.GetIndexedRows(),
	}

	return nil
}

func (gibpt *getIndexBuildProgressTask) PostExecute(ctx context.Context) error {
	return nil
}

// Deprecated: use describeIndexTask instead
type getIndexStateTask struct {
	baseTask
	Condition
	*milvuspb.GetIndexStateRequest
	ctx      context.Context
	mixCoord types.MixCoordClient
	result   *milvuspb.GetIndexStateResponse

	collectionID UniqueID
}

func (gist *getIndexStateTask) TraceCtx() context.Context {
	return gist.ctx
}

func (gist *getIndexStateTask) ID() UniqueID {
	return gist.Base.MsgID
}

func (gist *getIndexStateTask) SetID(uid UniqueID) {
	gist.Base.MsgID = uid
}

func (gist *getIndexStateTask) Name() string {
	return GetIndexStateTaskName
}

func (gist *getIndexStateTask) Type() commonpb.MsgType {
	return gist.Base.MsgType
}

func (gist *getIndexStateTask) BeginTs() Timestamp {
	return gist.Base.Timestamp
}

func (gist *getIndexStateTask) EndTs() Timestamp {
	return gist.Base.Timestamp
}

func (gist *getIndexStateTask) SetTs(ts Timestamp) {
	gist.Base.Timestamp = ts
}

func (gist *getIndexStateTask) OnEnqueue() error {
	gist.Base = commonpbutil.NewMsgBase()
	gist.Base.MsgType = commonpb.MsgType_GetIndexState
	gist.Base.SourceID = paramtable.GetNodeID()
	return nil
}

func (gist *getIndexStateTask) PreExecute(ctx context.Context) error {
	if err := validateCollectionName(gist.CollectionName); err != nil {
		return err
	}

	return nil
}

func (gist *getIndexStateTask) Execute(ctx context.Context) error {
	collectionID, err := globalMetaCache.GetCollectionID(ctx, gist.GetDbName(), gist.CollectionName)
	if err != nil {
		return err
	}

	state, err := gist.mixCoord.GetIndexState(ctx, &indexpb.GetIndexStateRequest{
		CollectionID: collectionID,
		IndexName:    gist.IndexName,
	})
	if err = merr.CheckRPCCall(state, err); err != nil {
		return err
	}

	gist.result = &milvuspb.GetIndexStateResponse{
		Status:     merr.Success(),
		State:      state.GetState(),
		FailReason: state.GetFailReason(),
	}
	return nil
}

func (gist *getIndexStateTask) PostExecute(ctx context.Context) error {
	return nil
}