fix: use binlog memory size for ArrayOfVector field data size estimation in index build (#48843)

issue: https://github.com/milvus-io/milvus/issues/48743,
https://github.com/milvus-io/milvus/issues/49014

---------

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

internal/datanode/index/task_index.go

@@ -235,13 +235,22 @@ func (it *indexBuildTask) Execute(ctx context.Context) error {
 
 	indexType := it.newIndexParams[common.IndexTypeKey]
 	var fieldDataSize uint64
-	var err error
 
-	// Ignore the error here, this param will only be used for diskann and aisaq
-	fieldDataSize, _ = estimateFieldDataSize(it.req.GetDim(), it.req.GetNumRows(), it.req.GetField().GetDataType())
+	// Estimate field data size, only used for diskann and aisaq
+	dim := uint64(it.req.GetDim())
+	numRows := uint64(it.req.GetNumRows())
+	switch it.req.GetField().GetDataType() {
+	case schemapb.DataType_BinaryVector:
+		fieldDataSize = dim / 8 * numRows
+	case schemapb.DataType_FloatVector:
+		fieldDataSize = dim * numRows * 4
+	case schemapb.DataType_Float16Vector, schemapb.DataType_BFloat16Vector:
+		fieldDataSize = dim * numRows * 2
+	case schemapb.DataType_ArrayOfVector:
+		fieldDataSize = getFieldDataSizeFromBinlogs(it.req.GetInsertLogs(), it.req.GetField().GetFieldID())
+	}
 	if vecindexmgr.GetVecIndexMgrInstance().IsDiskANN(indexType) {
-		err = indexparams.SetDiskIndexBuildParams(it.newIndexParams, int64(fieldDataSize))
-		if err != nil {
+		if err := indexparams.SetDiskIndexBuildParams(it.newIndexParams, int64(fieldDataSize)); err != nil {
 			log.Warn("failed to fill disk index params", zap.Error(err))
 			return err
 		}
@@ -324,6 +333,7 @@ func (it *indexBuildTask) Execute(ctx context.Context) error {
 		buildIndexParams.StoragePluginContext = it.pluginContext
 	}
 
+	var err error
 	it.index, err = indexcgowrapper.CreateIndex(ctx, buildIndexParams)
 	if err != nil {
 		if it.index != nil && it.index.CleanLocalData() != nil {

internal/datanode/index/util.go

@@ -28,11 +28,9 @@ package index
 import "C"
 
 import (
-	"github.com/cockroachdb/errors"
-
 	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
-	"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
 	"github.com/milvus-io/milvus/pkg/v2/common"
+	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v2/util/hardware"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
@@ -46,19 +44,17 @@ func getCurrentIndexVersion(v int32) int32 {
 	return v
 }
 
-func estimateFieldDataSize(dim int64, numRows int64, dataType schemapb.DataType) (uint64, error) {
-	switch dataType {
-	case schemapb.DataType_BinaryVector:
-		return uint64(dim) / 8 * uint64(numRows), nil
-	case schemapb.DataType_FloatVector:
-		return uint64(dim) * uint64(numRows) * 4, nil
-	case schemapb.DataType_Float16Vector, schemapb.DataType_BFloat16Vector:
-		return uint64(dim) * uint64(numRows) * 2, nil
-	case schemapb.DataType_SparseFloatVector:
-		return 0, errors.New("could not estimate field data size of SparseFloatVector")
-	default:
-		return 0, nil
+func getFieldDataSizeFromBinlogs(insertLogs []*datapb.FieldBinlog, fieldID int64) uint64 {
+	var totalSize uint64
+	for _, fieldBinlog := range insertLogs {
+		if fieldBinlog.GetFieldID() == fieldID {
+			for _, binlog := range fieldBinlog.GetBinlogs() {
+				totalSize += uint64(binlog.GetMemorySize())
+			}
+			break
+		}
 	}
+	return totalSize
 }
 
 func mapToKVPairs(m map[string]string) []*commonpb.KeyValuePair {

internal/datanode/index/util_test.go

@@ -21,6 +21,8 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/suite"
+
+	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 )
 
 type utilSuite struct {
@@ -37,6 +39,45 @@ func (s *utilSuite) Test_mapToKVPairs() {
 	s.Equal(3, len(mapToKVPairs(indexParams)))
 }
 
+func (s *utilSuite) Test_getFieldDataSizeFromBinlogs() {
+	// Storage V2/V3 splits vector fields into their own column group,
+	// where GroupID (used as FieldBinlog.FieldID) equals the vector field's ID.
+	// Non-vector fields may be grouped together under a different GroupID.
+	vectorFieldID := int64(101)
+	insertLogs := []*datapb.FieldBinlog{
+		{
+			// Non-vector column group: GroupID=100, contains scalar fields 100 and 102
+			FieldID:     100,
+			ChildFields: []int64{100, 102},
+			Binlogs: []*datapb.Binlog{
+				{MemorySize: 1024},
+				{MemorySize: 2048},
+			},
+		},
+		{
+			// Vector field in its own column group: GroupID = FieldID = 101
+			FieldID:     vectorFieldID,
+			ChildFields: []int64{vectorFieldID},
+			Binlogs: []*datapb.Binlog{
+				{MemorySize: 4096},
+				{MemorySize: 8192},
+			},
+		},
+	}
+
+	// Vector field matches by FieldID (== GroupID)
+	s.Equal(uint64(12288), getFieldDataSizeFromBinlogs(insertLogs, vectorFieldID))
+
+	// Scalar field grouped under GroupID=100
+	s.Equal(uint64(3072), getFieldDataSizeFromBinlogs(insertLogs, 100))
+
+	// No match
+	s.Equal(uint64(0), getFieldDataSizeFromBinlogs(insertLogs, 999))
+
+	// Nil insert logs
+	s.Equal(uint64(0), getFieldDataSizeFromBinlogs(nil, vectorFieldID))
+}
+
 func Test_utilSuite(t *testing.T) {
 	suite.Run(t, new(utilSuite))
 }