fix: require sufficient rows and vectors before building ArrayOfVector EmbList indexes (#50625)

issue: https://github.com/milvus-io/milvus/issues/42148

---------

Signed-off-by: SpadeA <tangchenjie1210@gmail.com>

internal/datacoord/task_index.go

@@ -19,6 +19,7 @@ package datacoord
 import (
 	"context"
 	"path"
+	"strings"
 	"time"
 
 	"github.com/cockroachdb/errors"
@@ -34,11 +35,13 @@ import (
 	"github.com/milvus-io/milvus/internal/util/vecindexmgr"
 	"github.com/milvus-io/milvus/pkg/v3/common"
 	"github.com/milvus-io/milvus/pkg/v3/log"
+	"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v3/proto/indexpb"
 	"github.com/milvus-io/milvus/pkg/v3/proto/workerpb"
 	"github.com/milvus-io/milvus/pkg/v3/taskcommon"
 	"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"
 )
@@ -58,6 +61,8 @@ type indexBuildTask struct {
 
 var _ globalTask.Task = (*indexBuildTask)(nil)
 
+var errVectorArrayFieldBinlogNotFound = errors.New("vector array field binlog not found")
+
 func newIndexBuildTask(segIndex *model.SegmentIndex,
 	taskSlot int64,
 	meta *meta,
@@ -169,18 +174,74 @@ func (it *indexBuildTask) CreateTaskOnWorker(nodeID int64, cluster session.Clust
 	indexParams := it.meta.indexMeta.GetIndexParams(segIndex.CollectionID, segIndex.IndexID)
 	indexType := GetIndexType(indexParams)
 	effectiveRows := segIndex.NumRows
+	estimatedVectorArrayVectors := int64(0)
+	isVectorArrayIndex := false
+	isEmbeddingListIndex := false
 	if fieldID := it.meta.indexMeta.GetFieldIDByIndexID(segIndex.CollectionID, segIndex.IndexID); fieldID > 0 {
 		if collectionInfo, err := it.handler.GetCollection(ctx, segIndex.CollectionID); err == nil {
 			for _, f := range typeutil.GetAllFieldSchemas(collectionInfo.Schema) {
-				if f.FieldID == fieldID && f.GetNullable() && typeutil.IsVectorType(f.GetDataType()) {
-					effectiveRows = segmentutil.CalcValidRowCountFromFieldBinLog(segment.SegmentInfo, fieldID)
+				if f.FieldID == fieldID {
+					if f.GetNullable() && typeutil.IsVectorType(f.GetDataType()) {
+						effectiveRows = segmentutil.CalcValidRowCountFromFieldBinLog(segment.SegmentInfo, fieldID)
+					}
+					isVectorArrayIndex = typeutil.IsVectorArrayType(f.GetDataType())
+					isEmbeddingListIndex = isVectorArrayIndex && isEmbeddingListMetric(indexParams)
+					if isVectorArrayIndex {
+						estimate, err := estimateVectorArrayElementCountForIndexBuild(segment.SegmentInfo, collectionInfo.Schema, f)
+						if err != nil {
+							failReason := "failed to estimate vector array element count, count is unknown: " + err.Error()
+							log.Warn("failed to estimate vector array element count",
+								zap.Int64("fieldID", f.GetFieldID()),
+								zap.String("fieldName", f.GetName()),
+								zap.String("failReason", failReason),
+								zap.Error(err))
+							if updateErr := it.UpdateStateWithMeta(indexpb.JobState_JobStateFailed, failReason); updateErr != nil {
+								log.Warn("failed to update vector array index task state to Failed",
+									zap.String("failReason", failReason),
+									zap.Error(updateErr))
+							}
+							return
+						}
+						estimatedVectorArrayVectors = estimate.vectorCount
+						if estimate.emptyOnStaleSchema {
+							effectiveRows = 0
+							log.Info("vector array field binlog is absent on stale schema segment, treating as empty field",
+								zap.Int64("fieldID", f.GetFieldID()),
+								zap.String("fieldName", f.GetName()),
+								zap.Int32("segmentSchemaVersion", segment.GetSchemaVersion()),
+								zap.Int32("collectionSchemaVersion", collectionInfo.Schema.GetVersion()))
+						}
+					}
 					break
 				}
 			}
 		}
 	}
-	if isNoTrainIndex(indexType) || effectiveRows < Params.DataCoordCfg.MinSegmentNumRowsToEnableIndex.GetAsInt64() {
-		log.Info("segment does not need index really, marking as finished", zap.Int64("numRows", segIndex.NumRows), zap.Int64("effectiveRows", effectiveRows))
+	minRowsToBuildIndex := Params.DataCoordCfg.MinSegmentNumRowsToEnableIndex.GetAsInt64()
+	rowCountBelowThreshold := effectiveRows < minRowsToBuildIndex
+	vectorArrayVectorCountBelowThreshold := false
+	indexDataBelowThreshold := rowCountBelowThreshold
+	if isVectorArrayIndex {
+		// Element-level ArrayOfVector indexes are built from flattened inner vectors,
+		// so row count alone should not block index building. MaxSim metrics build
+		// EmbList indexes and additionally require enough logical rows.
+		vectorArrayVectorCountBelowThreshold = estimatedVectorArrayVectors < minRowsToBuildIndex
+		indexDataBelowThreshold = vectorArrayVectorCountBelowThreshold ||
+			(isEmbeddingListIndex && rowCountBelowThreshold)
+	}
+	if isNoTrainIndex(indexType) || indexDataBelowThreshold {
+		log.Info("segment does not need index really, marking as finished",
+			zap.Int64("numRows", segIndex.NumRows),
+			zap.Int64("effectiveRows", effectiveRows),
+			zap.Int64("estimatedVectorArrayVectors", estimatedVectorArrayVectors),
+			zap.Int64("minRowsToBuildIndex", minRowsToBuildIndex),
+			zap.String("indexType", indexType),
+			zap.Bool("vectorArrayIndex", isVectorArrayIndex),
+			zap.Bool("embeddingListIndex", isEmbeddingListIndex),
+			zap.Bool("rowCountBelowThreshold", rowCountBelowThreshold),
+			zap.Bool("vectorArrayVectorCountBelowThreshold", vectorArrayVectorCountBelowThreshold),
+			zap.Bool("indexDataBelowThreshold", indexDataBelowThreshold),
+		)
 		now := time.Now()
 		it.SetTaskTime(taskcommon.TimeStart, now)
 		it.SetTaskTime(taskcommon.TimeEnd, now)
@@ -222,6 +283,133 @@ func (it *indexBuildTask) CreateTaskOnWorker(nodeID int64, cluster session.Clust
 	log.Info("index task assigned successfully")
 }
 
+func isEmbeddingListMetric(indexParams []*commonpb.KeyValuePair) bool {
+	metricType, err := getIndexParam(indexParams, common.MetricTypeKey)
+	if err != nil {
+		return false
+	}
+	switch strings.ToUpper(metricType) {
+	case metric.MaxSim,
+		metric.MaxSimCosine,
+		metric.MaxSimL2,
+		metric.MaxSimIP,
+		metric.MaxSimHamming,
+		metric.MaxSimJaccard:
+		return true
+	default:
+		return false
+	}
+}
+
+type vectorArrayElementCountEstimate struct {
+	vectorCount        int64
+	emptyOnStaleSchema bool
+}
+
+func estimateVectorArrayElementCountForIndexBuild(segment *datapb.SegmentInfo, schema *schemapb.CollectionSchema, field *schemapb.FieldSchema) (vectorArrayElementCountEstimate, error) {
+	count, err := estimateVectorArrayElementCount(segment, field)
+	if err == nil {
+		return vectorArrayElementCountEstimate{vectorCount: count}, nil
+	}
+	if isMissingVectorArrayFieldOnStaleSchema(err, segment, schema, field) {
+		// A nullable field added after this segment was written has no binlog in the
+		// stale segment. For index build purposes it contributes zero vectors and
+		// should be fake-finished by the threshold check below.
+		return vectorArrayElementCountEstimate{emptyOnStaleSchema: true}, nil
+	}
+	return vectorArrayElementCountEstimate{}, err
+}
+
+func estimateVectorArrayElementCount(segment *datapb.SegmentInfo, field *schemapb.FieldSchema) (int64, error) {
+	if segment == nil {
+		return 0, merr.WrapErrServiceInternalMsg("segment info is nil")
+	}
+	if field == nil {
+		return 0, merr.WrapErrServiceInternalMsg("field schema is nil")
+	}
+	elementSize, err := vectorArrayElementSize(field)
+	if err != nil {
+		return 0, err
+	}
+
+	var totalPayloadBytes int64
+	var seenFieldLog bool
+	for _, fieldBinlog := range segment.GetBinlogs() {
+		if !fieldBinlogContainsField(fieldBinlog, field.GetFieldID()) {
+			continue
+		}
+
+		seenFieldLog = true
+		for _, binlog := range fieldBinlog.GetBinlogs() {
+			payloadBytes := binlog.GetMemorySize()
+			if payloadBytes <= 0 {
+				continue
+			}
+
+			// VectorArrayFieldData stores each logical row with a 4-byte inner-vector byte length.
+			payloadBytes -= binlog.GetEntriesNum() * 4
+			if field.GetNullable() {
+				payloadBytes -= binlog.GetEntriesNum()
+			}
+			if payloadBytes > 0 {
+				totalPayloadBytes += payloadBytes
+			}
+		}
+	}
+
+	if !seenFieldLog {
+		return 0, merr.WrapErrServiceInternalErr(errVectorArrayFieldBinlogNotFound, "fieldID=%d", field.GetFieldID())
+	}
+	return totalPayloadBytes / elementSize, nil
+}
+
+func isMissingVectorArrayFieldOnStaleSchema(err error, segment *datapb.SegmentInfo, schema *schemapb.CollectionSchema, field *schemapb.FieldSchema) bool {
+	return errors.Is(err, errVectorArrayFieldBinlogNotFound) &&
+		segment != nil &&
+		schema != nil &&
+		field != nil &&
+		field.GetNullable() &&
+		segment.GetSchemaVersion() < schema.GetVersion()
+}
+
+func fieldBinlogContainsField(fieldBinlog *datapb.FieldBinlog, fieldID int64) bool {
+	if fieldBinlog.GetFieldID() == fieldID {
+		return true
+	}
+	for _, childFieldID := range fieldBinlog.GetChildFields() {
+		if childFieldID == fieldID {
+			return true
+		}
+	}
+	return false
+}
+
+func vectorArrayElementSize(field *schemapb.FieldSchema) (int64, error) {
+	if field == nil {
+		return 0, merr.WrapErrServiceInternalMsg("field schema is nil")
+	}
+	dim, err := storage.GetDimFromParams(field.GetTypeParams())
+	if err != nil {
+		return 0, merr.WrapErrServiceInternalErr(err, "invalid vector array dim, fieldID=%d", field.GetFieldID())
+	}
+	if dim <= 0 {
+		return 0, merr.WrapErrParameterInvalidMsg("invalid vector array dim %d, fieldID=%d", dim, field.GetFieldID())
+	}
+
+	switch field.GetElementType() {
+	case schemapb.DataType_FloatVector:
+		return int64(dim) * 4, nil
+	case schemapb.DataType_BinaryVector:
+		return int64(dim+7) / 8, nil
+	case schemapb.DataType_Float16Vector, schemapb.DataType_BFloat16Vector:
+		return int64(dim) * 2, nil
+	case schemapb.DataType_Int8Vector:
+		return int64(dim), nil
+	default:
+		return 0, merr.WrapErrParameterInvalidMsg("unsupported vector array element type %s, fieldID=%d", field.GetElementType().String(), field.GetFieldID())
+	}
+}
+
 // Helper method to prepare job request
 func (it *indexBuildTask) prepareJobRequest(ctx context.Context, segment *SegmentInfo, segIndex *model.SegmentIndex,
 	indexParams []*commonpb.KeyValuePair, indexType string,

internal/datacoord/task_index_test.go

@@ -30,6 +30,7 @@ import (
 	catalogmocks "github.com/milvus-io/milvus/internal/metastore/mocks"
 	"github.com/milvus-io/milvus/internal/metastore/model"
 	"github.com/milvus-io/milvus/internal/mocks"
+	"github.com/milvus-io/milvus/pkg/v3/common"
 	"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v3/proto/indexpb"
 	"github.com/milvus-io/milvus/pkg/v3/proto/workerpb"
@@ -239,6 +240,315 @@ func (s *indexTaskSuite) TestCreateTaskOnWorker() {
 	})
 }
 
+func (s *indexTaskSuite) TestCreateTaskOnWorkerVectorArrayMaxSimRequiresEnoughVectors() {
+	const (
+		dim        = 128
+		enoughRows = int64(10000)
+	)
+	elementBytes := int64(dim * 4)
+
+	tests := []struct {
+		name              string
+		metricType        string
+		numRows           int64
+		innerVectorCount  int64
+		expectCreateIndex bool
+		expectedState     indexpb.JobState
+	}{
+		{
+			name:              "maxsim skips when inner vector count is below threshold",
+			metricType:        "MAX_SIM_COSINE",
+			numRows:           enoughRows,
+			innerVectorCount:  0,
+			expectCreateIndex: false,
+			expectedState:     indexpb.JobState_JobStateFinished,
+		},
+		{
+			name:              "maxsim skips when row count is below threshold",
+			metricType:        "MAX_SIM_COSINE",
+			numRows:           1023,
+			innerVectorCount:  2048,
+			expectCreateIndex: false,
+			expectedState:     indexpb.JobState_JobStateFinished,
+		},
+		{
+			name:              "maxsim builds when row count and inner vector count are enough",
+			metricType:        "MAX_SIM_COSINE",
+			numRows:           enoughRows,
+			innerVectorCount:  2048,
+			expectCreateIndex: true,
+			expectedState:     indexpb.JobState_JobStateInProgress,
+		},
+		{
+			name:              "element level metric skips when inner vector count is below threshold",
+			metricType:        "COSINE",
+			numRows:           enoughRows,
+			innerVectorCount:  0,
+			expectCreateIndex: false,
+			expectedState:     indexpb.JobState_JobStateFinished,
+		},
+		{
+			name:              "element level metric builds when inner vector count is enough",
+			metricType:        "COSINE",
+			numRows:           enoughRows,
+			innerVectorCount:  2048,
+			expectCreateIndex: true,
+			expectedState:     indexpb.JobState_JobStateInProgress,
+		},
+		{
+			name:              "element level metric builds when row count is below threshold but inner vector count is enough",
+			metricType:        "COSINE",
+			numRows:           1023,
+			innerVectorCount:  2048,
+			expectCreateIndex: true,
+			expectedState:     indexpb.JobState_JobStateInProgress,
+		},
+	}
+
+	for _, tc := range tests {
+		s.Run(tc.name, func() {
+			catalog := catalogmocks.NewDataCoordCatalog(s.T())
+			catalog.EXPECT().AlterSegmentIndexes(mock.Anything, mock.Anything).Return(nil).Maybe()
+
+			mt := &meta{
+				segments: &SegmentsInfo{
+					segments: map[int64]*SegmentInfo{
+						s.segID: {
+							SegmentInfo: &datapb.SegmentInfo{
+								ID:            s.segID,
+								CollectionID:  s.collID,
+								PartitionID:   s.partID,
+								InsertChannel: "ch1",
+								NumOfRows:     tc.numRows,
+								State:         commonpb.SegmentState_Flushed,
+								MaxRowNum:     tc.numRows,
+								Level:         datapb.SegmentLevel_L2,
+								Binlogs: []*datapb.FieldBinlog{{
+									FieldID: s.fieldID,
+									Binlogs: []*datapb.Binlog{{
+										EntriesNum: tc.numRows,
+										MemorySize: tc.numRows*4 + tc.innerVectorCount*elementBytes + 1,
+									}},
+								}},
+							},
+						},
+					},
+				},
+				indexMeta: createIndexMetaWithSegment(catalog, s.collID, s.partID, s.segID, s.indexID, s.fieldID, s.taskID),
+			}
+			mt.indexMeta.indexes[s.collID][s.indexID].IndexParams = []*commonpb.KeyValuePair{
+				{Key: common.IndexTypeKey, Value: "HNSW"},
+				{Key: common.MetricTypeKey, Value: tc.metricType},
+			}
+			mt.indexMeta.indexes[s.collID][s.indexID].TypeParams = []*commonpb.KeyValuePair{
+				{Key: common.DimKey, Value: fmt.Sprintf("%d", dim)},
+			}
+			segIndex, ok := mt.indexMeta.segmentBuildInfo.Get(s.taskID)
+			s.True(ok)
+			segIndex.NumRows = tc.numRows
+
+			field := &schemapb.FieldSchema{
+				Name:        "array_vector",
+				FieldID:     s.fieldID,
+				DataType:    schemapb.DataType_ArrayOfVector,
+				ElementType: schemapb.DataType_FloatVector,
+				TypeParams: []*commonpb.KeyValuePair{
+					{Key: common.DimKey, Value: fmt.Sprintf("%d", dim)},
+				},
+			}
+			handler := NewNMockHandler(s.T())
+			handler.EXPECT().GetCollection(mock.Anything, s.collID).Return(&collectionInfo{
+				ID: s.collID,
+				Schema: &schemapb.CollectionSchema{
+					Fields: []*schemapb.FieldSchema{field},
+				},
+				Partitions: []int64{s.partID},
+			}, nil).Maybe()
+
+			cm := mocks.NewChunkManager(s.T())
+			cm.EXPECT().RootPath().Return("root").Maybe()
+
+			it := newIndexBuildTask(segIndex, 1, mt, handler, cm, newIndexEngineVersionManager())
+			cluster := session.NewMockCluster(s.T())
+			if tc.expectCreateIndex {
+				cluster.EXPECT().CreateIndex(mock.Anything, mock.Anything).Return(nil)
+			}
+
+			it.CreateTaskOnWorker(1, cluster)
+			s.Equal(tc.expectedState, indexpb.JobState(it.IndexState))
+		})
+	}
+}
+
+func (s *indexTaskSuite) TestCreateTaskOnWorkerVectorArrayEstimateFailureMarksFailed() {
+	const (
+		dim        = 128
+		enoughRows = int64(10000)
+	)
+
+	catalog := catalogmocks.NewDataCoordCatalog(s.T())
+	catalog.EXPECT().AlterSegmentIndexes(mock.Anything, mock.Anything).Return(nil)
+
+	mt := &meta{
+		segments: &SegmentsInfo{
+			segments: map[int64]*SegmentInfo{
+				s.segID: {
+					SegmentInfo: &datapb.SegmentInfo{
+						ID:            s.segID,
+						CollectionID:  s.collID,
+						PartitionID:   s.partID,
+						InsertChannel: "ch1",
+						NumOfRows:     enoughRows,
+						State:         commonpb.SegmentState_Flushed,
+						MaxRowNum:     enoughRows,
+						Level:         datapb.SegmentLevel_L2,
+					},
+				},
+			},
+		},
+		indexMeta: createIndexMetaWithSegment(catalog, s.collID, s.partID, s.segID, s.indexID, s.fieldID, s.taskID),
+	}
+	mt.indexMeta.indexes[s.collID][s.indexID].IndexParams = []*commonpb.KeyValuePair{
+		{Key: common.IndexTypeKey, Value: "HNSW"},
+		{Key: common.MetricTypeKey, Value: "MAX_SIM_COSINE"},
+	}
+	segIndex, ok := mt.indexMeta.segmentBuildInfo.Get(s.taskID)
+	s.True(ok)
+	segIndex.NumRows = enoughRows
+
+	handler := NewNMockHandler(s.T())
+	handler.EXPECT().GetCollection(mock.Anything, s.collID).Return(&collectionInfo{
+		ID: s.collID,
+		Schema: &schemapb.CollectionSchema{
+			Fields: []*schemapb.FieldSchema{
+				{
+					Name:        "array_vector",
+					FieldID:     s.fieldID,
+					DataType:    schemapb.DataType_ArrayOfVector,
+					ElementType: schemapb.DataType_FloatVector,
+					TypeParams: []*commonpb.KeyValuePair{
+						{Key: common.DimKey, Value: fmt.Sprintf("%d", dim)},
+					},
+				},
+			},
+		},
+		Partitions: []int64{s.partID},
+	}, nil)
+
+	it := newIndexBuildTask(segIndex, 1, mt, handler, nil, newIndexEngineVersionManager())
+	it.CreateTaskOnWorker(1, session.NewMockCluster(s.T()))
+
+	s.Equal(indexpb.JobState_JobStateFailed, indexpb.JobState(it.IndexState))
+	s.Contains(it.FailReason, "failed to estimate vector array element count")
+	s.Contains(it.FailReason, "count is unknown")
+	s.Contains(it.FailReason, "vector array field binlog not found")
+}
+
+func (s *indexTaskSuite) TestCreateTaskOnWorkerVectorArrayMissingBinlogOnStaleSchemaFakeFinishes() {
+	const (
+		dim        = 128
+		enoughRows = int64(10000)
+	)
+
+	catalog := catalogmocks.NewDataCoordCatalog(s.T())
+	catalog.EXPECT().AlterSegmentIndexes(mock.Anything, mock.Anything).Return(nil)
+
+	mt := &meta{
+		segments: &SegmentsInfo{
+			segments: map[int64]*SegmentInfo{
+				s.segID: {
+					SegmentInfo: &datapb.SegmentInfo{
+						ID:            s.segID,
+						CollectionID:  s.collID,
+						PartitionID:   s.partID,
+						InsertChannel: "ch1",
+						NumOfRows:     enoughRows,
+						State:         commonpb.SegmentState_Flushed,
+						MaxRowNum:     enoughRows,
+						Level:         datapb.SegmentLevel_L2,
+						SchemaVersion: 0,
+					},
+				},
+			},
+		},
+		indexMeta: createIndexMetaWithSegment(catalog, s.collID, s.partID, s.segID, s.indexID, s.fieldID, s.taskID),
+	}
+	mt.indexMeta.indexes[s.collID][s.indexID].IndexParams = []*commonpb.KeyValuePair{
+		{Key: common.IndexTypeKey, Value: "HNSW"},
+		{Key: common.MetricTypeKey, Value: "MAX_SIM_COSINE"},
+	}
+	segIndex, ok := mt.indexMeta.segmentBuildInfo.Get(s.taskID)
+	s.True(ok)
+	segIndex.NumRows = enoughRows
+
+	handler := NewNMockHandler(s.T())
+	handler.EXPECT().GetCollection(mock.Anything, s.collID).Return(&collectionInfo{
+		ID: s.collID,
+		Schema: &schemapb.CollectionSchema{
+			Version: 1,
+			Fields: []*schemapb.FieldSchema{
+				{
+					Name:        "array_vector",
+					FieldID:     s.fieldID,
+					DataType:    schemapb.DataType_ArrayOfVector,
+					ElementType: schemapb.DataType_FloatVector,
+					Nullable:    true,
+					TypeParams: []*commonpb.KeyValuePair{
+						{Key: common.DimKey, Value: fmt.Sprintf("%d", dim)},
+					},
+				},
+			},
+		},
+		Partitions: []int64{s.partID},
+	}, nil)
+
+	it := newIndexBuildTask(segIndex, 1, mt, handler, nil, newIndexEngineVersionManager())
+	it.CreateTaskOnWorker(1, session.NewMockCluster(s.T()))
+
+	s.Equal(indexpb.JobState_JobStateFinished, indexpb.JobState(it.IndexState))
+	s.Equal("fake finished index success", it.FailReason)
+}
+
+func (s *indexTaskSuite) TestEstimateVectorArrayElementCountErrors() {
+	field := &schemapb.FieldSchema{
+		FieldID:     s.fieldID,
+		DataType:    schemapb.DataType_ArrayOfVector,
+		ElementType: schemapb.DataType_FloatVector,
+		TypeParams: []*commonpb.KeyValuePair{
+			{Key: common.DimKey, Value: "128"},
+		},
+	}
+
+	_, err := estimateVectorArrayElementCount(&datapb.SegmentInfo{}, field)
+	s.Error(err)
+	s.Contains(err.Error(), "vector array field binlog not found")
+
+	fieldWithoutDim := &schemapb.FieldSchema{
+		FieldID:     s.fieldID,
+		DataType:    schemapb.DataType_ArrayOfVector,
+		ElementType: schemapb.DataType_FloatVector,
+	}
+	_, err = estimateVectorArrayElementCount(&datapb.SegmentInfo{
+		Binlogs: []*datapb.FieldBinlog{{FieldID: s.fieldID}},
+	}, fieldWithoutDim)
+	s.Error(err)
+	s.Contains(err.Error(), "invalid vector array dim")
+
+	fieldWithUnsupportedElement := &schemapb.FieldSchema{
+		FieldID:     s.fieldID,
+		DataType:    schemapb.DataType_ArrayOfVector,
+		ElementType: schemapb.DataType_SparseFloatVector,
+		TypeParams: []*commonpb.KeyValuePair{
+			{Key: common.DimKey, Value: "128"},
+		},
+	}
+	_, err = estimateVectorArrayElementCount(&datapb.SegmentInfo{
+		Binlogs: []*datapb.FieldBinlog{{FieldID: s.fieldID}},
+	}, fieldWithUnsupportedElement)
+	s.Error(err)
+	s.Contains(err.Error(), "unsupported vector array element type")
+}
+
 func (s *indexTaskSuite) TestQueryTaskOnWorker() {
 	t := &model.SegmentIndex{
 		CollectionID: s.collID,