Sync the sealed and flushed segments to datanode

Signed-off-by: Cai Zhang <cai.zhang@zilliz.com>
milvus-io · Jul 1, 2024 · 2bac2f5 · 2bac2f5
1 parent ef3ced8
commit 2bac2f5
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 1 deletion.
diff --git a/internal/datacoord/compaction_trigger.go b/internal/datacoord/compaction_trigger.go
@@ -863,6 +863,10 @@ 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-- {

diff --git a/internal/datacoord/sync_segments_scheduler.go b/internal/datacoord/sync_segments_scheduler.go
@@ -115,7 +115,7 @@ func (sss *SyncSegmentsScheduler) SyncFlushedSegments(collectionID, partitionID
 	log := log.With(zap.Int64("collectionID", collectionID), zap.Int64("partitionID", partitionID),
 		zap.String("channelName", channelName), zap.Int64("nodeID", nodeID))
 	segments := sss.meta.SelectSegments(WithChannel(channelName), SegmentFilterFunc(func(info *SegmentInfo) bool {
-		return info.GetPartitionID() == partitionID && isFlush(info)
+		return info.GetPartitionID() == partitionID && needSync(info)
 	}))
 	req := &datapb.SyncSegmentsRequest{
 		ChannelName:  channelName,