feat: impl StructArray -- create schema, insert, and retrieve data

cmd/tools/migration/meta/meta220.go

@@ -50,6 +50,7 @@ func (meta *TtCollectionsMeta220) GenerateSaves(sourceVersion semver.Version) (m
 	if sourceVersion.LT(versions.Version220) {
 		opts = append(opts, model.WithFields())
 		opts = append(opts, model.WithPartitions())
+		opts = append(opts, model.WithStructArrayFields())
 	}
 
 	for collectionID := range *meta {
@@ -98,6 +99,7 @@ func (meta *CollectionsMeta220) GenerateSaves(sourceVersion semver.Version) (map
 	if sourceVersion.LT(versions.Version220) {
 		opts = append(opts, model.WithFields())
 		opts = append(opts, model.WithPartitions())
+		opts = append(opts, model.WithStructArrayFields())
 	}
 
 	for collectionID := range *meta {

internal/core/unittest/test_utils/DataGen.h

@@ -357,7 +357,8 @@ GenerateRandomSparseFloatVector(size_t rows,
     return tensor;
 }
 
-inline SchemaPtr CreateTestSchema() {
+inline SchemaPtr
+CreateTestSchema() {
     auto schema = std::make_shared<milvus::Schema>();
     auto bool_field =
         schema->AddDebugField("bool", milvus::DataType::BOOL, true);

internal/datacoord/compaction_trigger.go

@@ -786,10 +786,6 @@ func isFlush(segment *SegmentInfo) bool {
 	return segment.GetState() == commonpb.SegmentState_Flushed || segment.GetState() == commonpb.SegmentState_Flushing
 }
 
-func needSync(segment *SegmentInfo) bool {
-	return segment.GetState() == commonpb.SegmentState_Flushed || segment.GetState() == commonpb.SegmentState_Flushing || segment.GetState() == commonpb.SegmentState_Sealed
-}
-
 // buckets will be updated inplace
 func (t *compactionTrigger) squeezeSmallSegmentsToBuckets(small []*SegmentInfo, buckets [][]*SegmentInfo, expectedSize int64) (remaining []*SegmentInfo) {
 	for i := len(small) - 1; i >= 0; i-- {

internal/datacoord/services.go

@@ -135,7 +135,7 @@ func (s *Server) Flush(ctx context.Context, req *datapb.FlushRequest) (*datapb.F
 	flushSegmentIDs := make([]UniqueID, 0, len(segments))
 	for _, segment := range segments {
 		if segment != nil &&
-			(isFlushState(segment.GetState())) &&
+			isFlushState(segment.GetState()) &&
 			segment.GetLevel() != datapb.SegmentLevel_L0 && // SegmentLevel_Legacy, SegmentLevel_L1, SegmentLevel_L2
 			!sealedSegmentsIDDict[segment.GetID()] {
 			flushSegmentIDs = append(flushSegmentIDs, segment.GetID())

internal/datanode/compactor/clustering_compactor.go

@@ -605,7 +605,7 @@ func (t *clusteringCompactionTask) mappingSegment(
 	}
 
 	reader := storage.NewDeserializeReader(rr, func(r storage.Record, v []*storage.Value) error {
-		return storage.ValueDeserializer(r, v, t.plan.Schema.Fields)
+		return storage.ValueDeserializer(r, v, t.plan.Schema)
 	})
 	defer reader.Close()
 
@@ -884,7 +884,7 @@ func (t *clusteringCompactionTask) scalarAnalyzeSegment(
 	}
 
 	pkIter := storage.NewDeserializeReader(rr, func(r storage.Record, v []*storage.Value) error {
-		return storage.ValueDeserializer(r, v, t.plan.Schema.Fields)
+		return storage.ValueDeserializer(r, v, t.plan.Schema)
 	})
 	defer pkIter.Close()
 	analyzeResult, remained, err := t.iterAndGetScalarAnalyzeResult(pkIter, expiredFilter)

internal/datanode/compactor/segment_writer.go

@@ -554,7 +554,7 @@ func NewSegmentWriter(sch *schemapb.CollectionSchema, maxCount int64, batchSize
 
 func newBinlogWriter(collID, partID, segID int64, schema *schemapb.CollectionSchema, batchSize int,
 ) (writer *storage.BinlogSerializeWriter, closers []func() (*storage.Blob, error), err error) {
-	fieldWriters := storage.NewBinlogStreamWriters(collID, partID, segID, schema.Fields)
+	fieldWriters := storage.NewBinlogStreamWriters(collID, partID, segID, schema)
 	closers = make([]func() (*storage.Blob, error), 0, len(fieldWriters))
 	for _, w := range fieldWriters {
 		closers = append(closers, w.Finalize)

internal/datanode/importv2/util.go

@@ -435,6 +435,12 @@
 	fields := lo.KeyBy(schema.GetFields(), func(field *schemapb.FieldSchema) int64 {
 		return field.GetFieldID()
 	})
+	for _, structField := range schema.GetStructArrayFields() {
+		for _, subField := range structField.GetFields() {
+			fields[subField.GetFieldID()] = subField
+		}
+	}
+
 	for fieldID, fd := range data.Data {
 		if fd.RowNum() == 0 && CanBeZeroRowField(fields[fieldID]) {
 			continue

internal/flushcommon/syncmgr/pack_writer_v2.go

@@ -210,15 +210,15 @@ func (bw *BulkPackWriterV2) serializeBinlog(ctx context.Context, pack *SyncPack)
 	if len(pack.insertData) == 0 {
 		return nil, nil
 	}
-	arrowSchema, err := storage.ConvertToArrowSchema(bw.schema.Fields)
+	arrowSchema, err := storage.ConvertToArrowSchema(bw.schema)
 	if err != nil {
 		return nil, err
 	}
 	builder := array.NewRecordBuilder(memory.DefaultAllocator, arrowSchema)
 	defer builder.Release()
 
 	for _, chunk := range pack.insertData {
-		if err := storage.BuildRecord(builder, chunk, bw.schema.GetFields()); err != nil {
+		if err := storage.BuildRecord(builder, chunk, bw.schema); err != nil {
 			return nil, err
 		}
 	}

internal/metastore/kv/rootcoord/kv_catalog.go

@@ -80,6 +80,14 @@
 	return fmt.Sprintf("%s/%d", BuildFunctionPrefix(collectionID), functionID)
 }
 
+func BuildStructArrayFieldPrefix(collectionID typeutil.UniqueID) string {
+	return fmt.Sprintf("%s/%d", StructArrayFieldMetaPrefix, collectionID)
+}
+
+func BuildStructArrayFieldKey(collectionId typeutil.UniqueID, fieldId int64) string {
+	return fmt.Sprintf("%s/%d", BuildStructArrayFieldPrefix(collectionId), fieldId)
+}
+
 func BuildAliasKey210(alias string) string {
 	return fmt.Sprintf("%s/%s", CollectionAliasMetaPrefix210, alias)
 }
@@ -209,6 +217,17 @@
 		kvs[k] = string(v)
 	}
 
+	// save struct array fields to new path
+	for _, structArrayField := range coll.StructArrayFields {
+		k := BuildStructArrayFieldKey(coll.CollectionID, structArrayField.FieldID)
+		structArrayFieldInfo := model.MarshalStructArrayFieldModel(structArrayField)
+		v, err := proto.Marshal(structArrayFieldInfo)
+		if err != nil {
+			return err
+		}
+		kvs[k] = string(v)
+	}
+
 	// save functions info to new path.
 	for _, function := range coll.Functions {
 		k := BuildFunctionKey(coll.CollectionID, function.ID)
@@ -389,7 +408,7 @@
 }
 
 func fieldVersionAfter210(collMeta *pb.CollectionInfo) bool {
-	return len(collMeta.GetSchema().GetFields()) <= 0
+	return len(collMeta.GetSchema().GetFields()) <= 0 && len(collMeta.GetSchema().GetStructArrayFields()) <= 0
 }
 
 func (kc *Catalog) listFieldsAfter210(ctx context.Context, collectionID typeutil.UniqueID, ts typeutil.Timestamp) ([]*model.Field, error) {
@@ -436,6 +455,24 @@
 	return ret, nil
 }
 
+func (kc *Catalog) listStructArrayFieldsAfter210(ctx context.Context, collectionID typeutil.UniqueID, ts typeutil.Timestamp) ([]*model.StructArrayField, error) {
+	prefix := BuildStructArrayFieldPrefix(collectionID)
+	_, values, err := kc.Snapshot.LoadWithPrefix(ctx, prefix, ts)
+	if err != nil {
+		return nil, err
+	}
+	structFields := make([]*model.StructArrayField, 0, len(values))
+	for _, v := range values {
+		partitionMeta := &schemapb.StructArrayFieldSchema{}
+		err := proto.Unmarshal([]byte(v), partitionMeta)
+		if err != nil {
+			return nil, err
+		}
+		structFields = append(structFields, model.UnmarshalStructArrayFieldModel(partitionMeta))
+	}
+	return structFields, nil
+}
+
 func (kc *Catalog) listFunctions(ctx context.Context, collectionID typeutil.UniqueID, ts typeutil.Timestamp) ([]*model.Function, error) {
 	prefix := BuildFunctionPrefix(collectionID)
 	_, values, err := kc.Snapshot.LoadWithPrefix(ctx, prefix, ts)
@@ -499,6 +536,12 @@
 	}
 	collection.Fields = fields
 
+	structArrayFields, err := kc.listStructArrayFieldsAfter210(ctx, collection.CollectionID, ts)
+	if err != nil {
+		return nil, err
+	}
+	collection.StructArrayFields = structArrayFields
+
 	functions, err := kc.listFunctions(ctx, collection.CollectionID, ts)
 	if err != nil {
 		return nil, err
@@ -610,6 +653,9 @@
 	for _, field := range collectionInfo.Fields {
 		delMetakeysSnap = append(delMetakeysSnap, BuildFieldKey(collectionInfo.CollectionID, field.FieldID))
 	}
+	for _, structArrayField := range collectionInfo.StructArrayFields {
+		delMetakeysSnap = append(delMetakeysSnap, BuildStructArrayFieldKey(collectionInfo.CollectionID, structArrayField.FieldID))
+	}
 	for _, function := range collectionInfo.Functions {
 		delMetakeysSnap = append(delMetakeysSnap, BuildFunctionKey(collectionInfo.CollectionID, function.ID))
 	}
@@ -650,6 +696,7 @@
 	oldCollClone.State = newColl.State
 	oldCollClone.Properties = newColl.Properties
 	oldCollClone.Fields = newColl.Fields
+	oldCollClone.StructArrayFields = newColl.StructArrayFields
 	oldCollClone.UpdateTimestamp = newColl.UpdateTimestamp
 
 	newKey := BuildCollectionKey(newColl.DBID, oldColl.CollectionID)
@@ -670,7 +717,18 @@
 			}
 			saves[k] = string(v)
 		}
+
+		for _, structArrayField := range newColl.StructArrayFields {
+			k := BuildStructArrayFieldKey(newColl.CollectionID, structArrayField.FieldID)
+			structArrayFieldInfo := model.MarshalStructArrayFieldModel(structArrayField)
+			v, err := proto.Marshal(structArrayFieldInfo)
+			if err != nil {
+				return err
+			}
+			saves[k] = string(v)
+		}
 	}
+
 	return etcd.SaveByBatchWithLimit(saves, util.MaxEtcdTxnNum/2, func(partialKvs map[string]string) error {
 		return kc.Snapshot.MultiSave(ctx, partialKvs, ts)
 	})

internal/metastore/kv/rootcoord/kv_catalog_test.go

@@ -214,6 +214,12 @@ func TestCatalog_ListCollections(t *testing.T) {
 			}), ts).
 			Return([]string{"rootcoord/functions/1/1"}, []string{string(fcm)}, nil)
 
+		kv.On("LoadWithPrefix", mock.Anything, mock.MatchedBy(
+			func(prefix string) bool {
+				return strings.HasPrefix(prefix, StructArrayFieldMetaPrefix)
+			}), ts).
+			Return([]string{}, []string{}, nil)
+
 		kc := NewCatalog(nil, kv)
 		ret, err := kc.ListCollections(ctx, testDb, ts)
 		assert.NoError(t, err)
@@ -264,6 +270,12 @@ func TestCatalog_ListCollections(t *testing.T) {
 			}), ts).
 			Return([]string{"rootcoord/functions/1/1"}, []string{string(fcm)}, nil)
 
+		kv.On("LoadWithPrefix", mock.Anything, mock.MatchedBy(
+			func(prefix string) bool {
+				return strings.HasPrefix(prefix, StructArrayFieldMetaPrefix)
+			}), ts).
+			Return([]string{}, []string{}, nil)
+
 		kv.On("MultiSaveAndRemove", mock.Anything, mock.Anything, mock.Anything, ts).Return(nil)
 		kc := NewCatalog(nil, kv)
 
@@ -1287,7 +1299,7 @@ func TestCatalog_CreateCollection(t *testing.T) {
 		assert.NoError(t, err)
 	})
 
-	t.Run("create collection with function", func(t *testing.T) {
+	t.Run("create collection with function and struct array field", func(t *testing.T) {
 		mockSnapshot := newMockSnapshot(t, withMockSave(nil), withMockMultiSave(nil))
 		kc := NewCatalog(nil, mockSnapshot)
 		ctx := context.Background()
@@ -1311,6 +1323,23 @@ func TestCatalog_CreateCollection(t *testing.T) {
 					DataType: schemapb.DataType_SparseFloatVector,
 				},
 			},
+			StructArrayFields: []*model.StructArrayField{
+				{
+					Name: "test_struct",
+					Fields: []*model.Field{
+						{
+							Name:        "sub_text",
+							DataType:    schemapb.DataType_Array,
+							ElementType: schemapb.DataType_VarChar,
+						},
+						{
+							Name:        "sub_sparse",
+							DataType:    schemapb.DataType_ArrayOfVector,
+							ElementType: schemapb.DataType_SparseFloatVector,
+						},
+					},
+				},
+			},
 			Functions: []*model.Function{
 				{
 					Name:             "test",
@@ -1417,6 +1446,23 @@ func TestCatalog_DropCollection(t *testing.T) {
 					DataType: schemapb.DataType_SparseFloatVector,
 				},
 			},
+			StructArrayFields: []*model.StructArrayField{
+				{
+					Name: "test_struct",
+					Fields: []*model.Field{
+						{
+							Name:        "sub_text",
+							DataType:    schemapb.DataType_Array,
+							ElementType: schemapb.DataType_VarChar,
+						},
+						{
+							Name:        "sub_sparse",
+							DataType:    schemapb.DataType_ArrayOfVector,
+							ElementType: schemapb.DataType_SparseFloatVector,
+						},
+					},
+				},
+			},
 			Functions: []*model.Function{
 				{
 					Name:             "test",

internal/metastore/kv/rootcoord/rootcoord_constant.go

@@ -17,10 +17,11 @@ const (
 	// CollectionMetaPrefix prefix for collection meta
 	CollectionMetaPrefix = ComponentPrefix + "/collection"
 
-	PartitionMetaPrefix = ComponentPrefix + "/partitions"
-	AliasMetaPrefix     = ComponentPrefix + "/aliases"
-	FieldMetaPrefix     = ComponentPrefix + "/fields"
-	FunctionMetaPrefix  = ComponentPrefix + "/functions"
+	PartitionMetaPrefix        = ComponentPrefix + "/partitions"
+	AliasMetaPrefix            = ComponentPrefix + "/aliases"
+	FieldMetaPrefix            = ComponentPrefix + "/fields"
+	StructArrayFieldMetaPrefix = ComponentPrefix + "/struct-array-fields"
+	FunctionMetaPrefix         = ComponentPrefix + "/functions"
 
 	// CollectionAliasMetaPrefix210 prefix for collection alias meta
 	CollectionAliasMetaPrefix210 = ComponentPrefix + "/collection-alias"