enhance: support AckSyncUp for broadcaster, and enable it in truncate api (milvus-io#46313)

issue: milvus-io#43897
also for issue: milvus-io#46166

add ack_sync_up flag into broadcast message header, which indicates that
whether the broadcast operation is need to be synced up between the
streaming node and the coordinator.
If the ack_sync_up is false, the broadcast operation will be acked once
the recovery storage see the message at current vchannel, the fast ack
operation can be applied to speed up the broadcast operation.
If the ack_sync_up is true, the broadcast operation will be acked after
the checkpoint of current vchannel reach current message.
The fast ack operation can not be applied to speed up the broadcast
operation, because the ack operation need to be synced up with streaming
node.
e.g. if truncate collection operation want to call ack once callback
after the all segment are flushed at current vchannel, it should set the
ack_sync_up to be true.

TODO: current implementation doesn't promise the ack sync up semantic,
it only promise FastAck operation will not be applied, wait for 3.0 to
implement the ack sync up semantic. only for truncate api now.

---------

Signed-off-by: chyezh <chyezh@outlook.com>

Author: chyezh

internal/datacoord/handler.go

@@ -565,6 +565,7 @@ func (h *ServerHandler) GetCollection(ctx context.Context, collectionID UniqueID
 		return nil, err
 	}
 
+	// TODO: the cache should be removed in next step.
 	return h.s.meta.GetCollection(collectionID), nil
 }
 

internal/datacoord/server_test.go

@@ -2582,7 +2582,6 @@ func TestServer_InitMessageCallback(t *testing.T) {
 	server.initMessageCallback()
 
 	// Test Import message check callback
-	resourceKey := message.NewImportJobIDResourceKey(1)
 	msg, err := message.NewImportMessageBuilderV1().
 		WithHeader(&message.ImportMessageHeader{}).
 		WithBody(&msgpb.ImportMsg{
@@ -2591,7 +2590,7 @@ func TestServer_InitMessageCallback(t *testing.T) {
 			},
 			Schema: &schemapb.CollectionSchema{},
 		}).
-		WithBroadcast([]string{"ch-0"}, resourceKey).
+		WithBroadcast([]string{"ch-0"}).
 		BuildBroadcast()
 	err = registry.CallMessageCheckCallback(ctx, msg)
 	assert.NoError(t, err)
@@ -2605,7 +2604,7 @@ func TestServer_InitMessageCallback(t *testing.T) {
 			},
 			Schema: &schemapb.CollectionSchema{},
 		}).
-		WithBroadcast([]string{"test_channel"}, resourceKey).
+		WithBroadcast([]string{"test_channel"}).
 		MustBuildBroadcast()
 	err = registry.CallMessageAckCallback(ctx, importMsg, map[string]*message.AppendResult{
 		"test_channel": {

internal/datacoord/util.go

@@ -197,6 +197,10 @@ func getCompactedSegmentSize(s *datapb.CompactionSegment) int64 {
 // getCollectionAutoCompactionEnabled returns whether auto compaction for collection is enabled.
 // if not set, returns global auto compaction config.
 func getCollectionAutoCompactionEnabled(properties map[string]string) (bool, error) {
+	// when collection is on truncating, disable auto compaction.
+	if _, ok := properties[common.CollectionOnTruncatingKey]; ok {
+		return false, nil
+	}
 	v, ok := properties[common.CollectionAutoCompactionKey]
 	if ok {
 		enabled, err := strconv.ParseBool(v)

internal/distributed/streaming/streaming_test.go

@@ -147,7 +147,7 @@ func TestStreamingBroadcast(t *testing.T) {
 			CollectionID:   1,
 			CollectionName: collectionName,
 		}).
-		WithBroadcast(vChannels, message.NewExclusiveCollectionNameResourceKey("db", collectionName)).
+		WithBroadcast(vChannels).
 		BuildBroadcast()
 
 	resp, err := streaming.WAL().Broadcast().Append(context.Background(), msg)

internal/metastore/kv/rootcoord/kv_catalog.go

@@ -698,6 +698,7 @@ func (kc *Catalog) alterModifyCollection(ctx context.Context, oldColl *model.Col
 	oldCollClone.UpdateTimestamp = newColl.UpdateTimestamp
 	oldCollClone.EnableDynamicField = newColl.EnableDynamicField
 	oldCollClone.SchemaVersion = newColl.SchemaVersion
+	oldCollClone.ShardInfos = newColl.ShardInfos
 
 	newKey := BuildCollectionKey(newColl.DBID, oldColl.CollectionID)
 	value, err := proto.Marshal(model.MarshalCollectionModel(oldCollClone))

internal/metastore/model/collection.go

@@ -28,30 +28,36 @@ import (
 
 // TODO: These collection is dirty implementation and easy to be broken, we should drop it in the future.
 type Collection struct {
-	TenantID               string
-	DBID                   int64
-	CollectionID           int64
-	Partitions             []*Partition
-	Name                   string
-	DBName                 string
-	Description            string
-	AutoID                 bool
-	Fields                 []*Field
-	StructArrayFields      []*StructArrayField
-	Functions              []*Function
-	VirtualChannelNames    []string
-	PhysicalChannelNames   []string
-	ShardsNum              int32
-	StartPositions         []*commonpb.KeyDataPair
-	CreateTime             uint64
-	ConsistencyLevel       commonpb.ConsistencyLevel
-	Aliases                []string // TODO: deprecate this.
-	Properties             []*commonpb.KeyValuePair
-	State                  pb.CollectionState
-	EnableDynamicField     bool
-	UpdateTimestamp        uint64
-	SchemaVersion          int32
-	LastTruncateTimestamps map[string]uint64
+	TenantID             string
+	DBID                 int64
+	CollectionID         int64
+	Partitions           []*Partition
+	Name                 string
+	DBName               string
+	Description          string
+	AutoID               bool
+	Fields               []*Field
+	StructArrayFields    []*StructArrayField
+	Functions            []*Function
+	VirtualChannelNames  []string
+	PhysicalChannelNames []string
+	ShardsNum            int32
+	StartPositions       []*commonpb.KeyDataPair
+	CreateTime           uint64
+	ConsistencyLevel     commonpb.ConsistencyLevel
+	Aliases              []string // TODO: deprecate this.
+	Properties           []*commonpb.KeyValuePair
+	State                pb.CollectionState
+	EnableDynamicField   bool
+	UpdateTimestamp      uint64
+	SchemaVersion        int32
+	ShardInfos           map[string]*ShardInfo
+}
+
+type ShardInfo struct {
+	PChannelName         string // the pchannel name of the shard, it is the same with the physical channel name.
+	VChannelName         string // the vchannel name of the shard, it is the same with the virtual channel name.
+	LastTruncateTimeTick uint64 // the last truncate time tick of the shard, if the shard is not truncated, the value is 0.
 }
 
 func (c *Collection) Available() bool {
@@ -60,59 +66,67 @@ func (c *Collection) Available() bool {
 
 func (c *Collection) ShallowClone() *Collection {
 	return &Collection{
-		TenantID:               c.TenantID,
-		DBID:                   c.DBID,
-		CollectionID:           c.CollectionID,
-		Name:                   c.Name,
-		DBName:                 c.DBName,
-		Description:            c.Description,
-		AutoID:                 c.AutoID,
-		Fields:                 c.Fields,
-		StructArrayFields:      c.StructArrayFields,
-		Partitions:             c.Partitions,
-		VirtualChannelNames:    c.VirtualChannelNames,
-		PhysicalChannelNames:   c.PhysicalChannelNames,
-		ShardsNum:              c.ShardsNum,
-		ConsistencyLevel:       c.ConsistencyLevel,
-		CreateTime:             c.CreateTime,
-		StartPositions:         c.StartPositions,
-		Aliases:                c.Aliases,
-		Properties:             c.Properties,
-		State:                  c.State,
-		EnableDynamicField:     c.EnableDynamicField,
-		Functions:              c.Functions,
-		UpdateTimestamp:        c.UpdateTimestamp,
-		SchemaVersion:          c.SchemaVersion,
-		LastTruncateTimestamps: c.LastTruncateTimestamps,
+		TenantID:             c.TenantID,
+		DBID:                 c.DBID,
+		CollectionID:         c.CollectionID,
+		Name:                 c.Name,
+		DBName:               c.DBName,
+		Description:          c.Description,
+		AutoID:               c.AutoID,
+		Fields:               c.Fields,
+		StructArrayFields:    c.StructArrayFields,
+		Partitions:           c.Partitions,
+		VirtualChannelNames:  c.VirtualChannelNames,
+		PhysicalChannelNames: c.PhysicalChannelNames,
+		ShardsNum:            c.ShardsNum,
+		ConsistencyLevel:     c.ConsistencyLevel,
+		CreateTime:           c.CreateTime,
+		StartPositions:       c.StartPositions,
+		Aliases:              c.Aliases,
+		Properties:           c.Properties,
+		State:                c.State,
+		EnableDynamicField:   c.EnableDynamicField,
+		Functions:            c.Functions,
+		UpdateTimestamp:      c.UpdateTimestamp,
+		SchemaVersion:        c.SchemaVersion,
+		ShardInfos:           c.ShardInfos,
 	}
 }
 
 func (c *Collection) Clone() *Collection {
+	shardInfos := make(map[string]*ShardInfo, len(c.ShardInfos))
+	for channelName, shardInfo := range c.ShardInfos {
+		shardInfos[channelName] = &ShardInfo{
+			VChannelName:         channelName,
+			PChannelName:         shardInfo.PChannelName,
+			LastTruncateTimeTick: shardInfo.LastTruncateTimeTick,
+		}
+	}
 	return &Collection{
-		TenantID:               c.TenantID,
-		DBID:                   c.DBID,
-		CollectionID:           c.CollectionID,
-		Name:                   c.Name,
-		DBName:                 c.DBName,
-		Description:            c.Description,
-		AutoID:                 c.AutoID,
-		Fields:                 CloneFields(c.Fields),
-		StructArrayFields:      CloneStructArrayFields(c.StructArrayFields),
-		Partitions:             ClonePartitions(c.Partitions),
-		VirtualChannelNames:    common.CloneStringList(c.VirtualChannelNames),
-		PhysicalChannelNames:   common.CloneStringList(c.PhysicalChannelNames),
-		ShardsNum:              c.ShardsNum,
-		ConsistencyLevel:       c.ConsistencyLevel,
-		CreateTime:             c.CreateTime,
-		StartPositions:         common.CloneKeyDataPairs(c.StartPositions),
-		Aliases:                common.CloneStringList(c.Aliases),
-		Properties:             common.CloneKeyValuePairs(c.Properties),
-		State:                  c.State,
-		EnableDynamicField:     c.EnableDynamicField,
-		Functions:              CloneFunctions(c.Functions),
-		UpdateTimestamp:        c.UpdateTimestamp,
-		SchemaVersion:          c.SchemaVersion,
-		LastTruncateTimestamps: common.CloneMap(c.LastTruncateTimestamps),
+		TenantID:             c.TenantID,
+		DBID:                 c.DBID,
+		CollectionID:         c.CollectionID,
+		Name:                 c.Name,
+		DBName:               c.DBName,
+		Description:          c.Description,
+		AutoID:               c.AutoID,
+		Fields:               CloneFields(c.Fields),
+		StructArrayFields:    CloneStructArrayFields(c.StructArrayFields),
+		Partitions:           ClonePartitions(c.Partitions),
+		VirtualChannelNames:  common.CloneStringList(c.VirtualChannelNames),
+		PhysicalChannelNames: common.CloneStringList(c.PhysicalChannelNames),
+		ShardsNum:            c.ShardsNum,
+		ConsistencyLevel:     c.ConsistencyLevel,
+		CreateTime:           c.CreateTime,
+		StartPositions:       common.CloneKeyDataPairs(c.StartPositions),
+		Aliases:              common.CloneStringList(c.Aliases),
+		Properties:           common.CloneKeyValuePairs(c.Properties),
+		State:                c.State,
+		EnableDynamicField:   c.EnableDynamicField,
+		Functions:            CloneFunctions(c.Functions),
+		UpdateTimestamp:      c.UpdateTimestamp,
+		SchemaVersion:        c.SchemaVersion,
+		ShardInfos:           shardInfos,
 	}
 }
 
@@ -179,29 +193,45 @@ func UnmarshalCollectionModel(coll *pb.CollectionInfo) *Collection {
 			PartitionCreatedTimestamp: coll.PartitionCreatedTimestamps[idx],
 		}
 	}
+	shardInfos := make(map[string]*ShardInfo, len(coll.VirtualChannelNames))
+	for idx, channelName := range coll.VirtualChannelNames {
+		if len(coll.ShardInfos) == 0 {
+			shardInfos[channelName] = &ShardInfo{
+				VChannelName:         channelName,
+				PChannelName:         coll.PhysicalChannelNames[idx],
+				LastTruncateTimeTick: 0,
+			}
+		} else {
+			shardInfos[channelName] = &ShardInfo{
+				VChannelName:         channelName,
+				PChannelName:         coll.PhysicalChannelNames[idx],
+				LastTruncateTimeTick: coll.ShardInfos[idx].LastTruncateTimeTick,
+			}
+		}
+	}
 
 	return &Collection{
-		CollectionID:           coll.ID,
-		DBID:                   coll.DbId,
-		Name:                   coll.Schema.Name,
-		DBName:                 coll.Schema.DbName,
-		Description:            coll.Schema.Description,
-		AutoID:                 coll.Schema.AutoID,
-		Fields:                 UnmarshalFieldModels(coll.GetSchema().GetFields()),
-		StructArrayFields:      UnmarshalStructArrayFieldModels(coll.GetSchema().GetStructArrayFields()),
-		Partitions:             partitions,
-		VirtualChannelNames:    coll.VirtualChannelNames,
-		PhysicalChannelNames:   coll.PhysicalChannelNames,
-		ShardsNum:              coll.ShardsNum,
-		ConsistencyLevel:       coll.ConsistencyLevel,
-		CreateTime:             coll.CreateTime,
-		StartPositions:         coll.StartPositions,
-		State:                  coll.State,
-		Properties:             coll.Properties,
-		EnableDynamicField:     coll.Schema.EnableDynamicField,
-		UpdateTimestamp:        coll.UpdateTimestamp,
-		SchemaVersion:          coll.Schema.Version,
-		LastTruncateTimestamps: coll.LastTruncateTimestamps,
+		CollectionID:         coll.ID,
+		DBID:                 coll.DbId,
+		Name:                 coll.Schema.Name,
+		DBName:               coll.Schema.DbName,
+		Description:          coll.Schema.Description,
+		AutoID:               coll.Schema.AutoID,
+		Fields:               UnmarshalFieldModels(coll.GetSchema().GetFields()),
+		StructArrayFields:    UnmarshalStructArrayFieldModels(coll.GetSchema().GetStructArrayFields()),
+		Partitions:           partitions,
+		VirtualChannelNames:  coll.VirtualChannelNames,
+		PhysicalChannelNames: coll.PhysicalChannelNames,
+		ShardsNum:            coll.ShardsNum,
+		ConsistencyLevel:     coll.ConsistencyLevel,
+		CreateTime:           coll.CreateTime,
+		StartPositions:       coll.StartPositions,
+		State:                coll.State,
+		Properties:           coll.Properties,
+		EnableDynamicField:   coll.Schema.EnableDynamicField,
+		UpdateTimestamp:      coll.UpdateTimestamp,
+		SchemaVersion:        coll.Schema.Version,
+		ShardInfos:           shardInfos,
 	}
 }
 
@@ -265,20 +295,32 @@ func marshalCollectionModelWithConfig(coll *Collection, c *config) *pb.Collectio
 		collSchema.StructArrayFields = structArrayFields
 	}
 
+	shardInfos := make([]*pb.CollectionShardInfo, len(coll.ShardInfos))
+	for idx, channelName := range coll.VirtualChannelNames {
+		if shard, ok := coll.ShardInfos[channelName]; ok {
+			shardInfos[idx] = &pb.CollectionShardInfo{
+				LastTruncateTimeTick: shard.LastTruncateTimeTick,
+			}
+		} else {
+			shardInfos[idx] = &pb.CollectionShardInfo{
+				LastTruncateTimeTick: 0,
+			}
+		}
+	}
 	collectionPb := &pb.CollectionInfo{
-		ID:                     coll.CollectionID,
-		DbId:                   coll.DBID,
-		Schema:                 collSchema,
-		CreateTime:             coll.CreateTime,
-		VirtualChannelNames:    coll.VirtualChannelNames,
-		PhysicalChannelNames:   coll.PhysicalChannelNames,
-		ShardsNum:              coll.ShardsNum,
-		ConsistencyLevel:       coll.ConsistencyLevel,
-		StartPositions:         coll.StartPositions,
-		State:                  coll.State,
-		Properties:             coll.Properties,
-		UpdateTimestamp:        coll.UpdateTimestamp,
-		LastTruncateTimestamps: coll.LastTruncateTimestamps,
+		ID:                   coll.CollectionID,
+		DbId:                 coll.DBID,
+		Schema:               collSchema,
+		CreateTime:           coll.CreateTime,
+		VirtualChannelNames:  coll.VirtualChannelNames,
+		PhysicalChannelNames: coll.PhysicalChannelNames,
+		ShardsNum:            coll.ShardsNum,
+		ConsistencyLevel:     coll.ConsistencyLevel,
+		StartPositions:       coll.StartPositions,
+		State:                coll.State,
+		Properties:           coll.Properties,
+		UpdateTimestamp:      coll.UpdateTimestamp,
+		ShardInfos:           shardInfos,
 	}
 
 	if c.withPartitions {

internal/metastore/model/collection_test.go

@@ -77,6 +77,13 @@ var (
 				Value: "v",
 			},
 		},
+		ShardInfos: map[string]*ShardInfo{
+			"vch": {
+				PChannelName:         "pch",
+				VChannelName:         "vch",
+				LastTruncateTimeTick: 0,
+			},
+		},
 	}
 
 	deprecatedColPb = &pb.CollectionInfo{
@@ -588,6 +595,18 @@ func TestClone(t *testing.T) {
 		Properties:           []*commonpb.KeyValuePair{{Key: "32", Value: "33"}},
 		State:                pb.CollectionState_CollectionCreated,
 		EnableDynamicField:   true,
+		ShardInfos: map[string]*ShardInfo{
+			"c1": {
+				PChannelName:         "c3",
+				VChannelName:         "c1",
+				LastTruncateTimeTick: 0,
+			},
+			"c2": {
+				PChannelName:         "c4",
+				VChannelName:         "c2",
+				LastTruncateTimeTick: 0,
+			},
+		},
 	}
 
 	clone1 := collection.Clone()