fix: allow mixed import compaction fallback positions (#50693)

## Summary - Allow mixed normal/import compaction results to use output binlog timestamp ranges even when the fallback start position is earlier than an import commit timestamp. - Normalize fallback-only start/dml positions to the max input commit timestamp when output timestamps are unavailable. - Add regression coverage for manual mix compaction with normal rows preceding committed import rows. Fixes #50464 ## Test Plan - [x] `go test -tags dynamic,test -gcflags="all=-N -l" -count=1 ./internal/datacoord -run '^TestMeta$' -ldflags="-r ${RPATH}"` - [x] `make static-check` Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
2026-07-21 10:15:43 +00:00 · 2026-06-23 10:22:29 +08:00
parent 1515cdc5ab
commit 360b19504d
2 changed files with 91 additions and 48 deletions
@@ -2284,21 +2284,29 @@ func normalizePositionTimestamp(pos *msgpb.MsgPosition, commitTs uint64) *msgpb.
 	}
 }

-func validateCompactionFallbackStartPosition(compactFromSegInfos []*SegmentInfo, fallbackStart *msgpb.MsgPosition) error {
-	if fallbackStart == nil {
-		return nil
-	}
+func maxCommitTimestamp(compactFromSegInfos []*SegmentInfo) uint64 {
 	var maxCommitTs uint64
 	for _, info := range compactFromSegInfos {
 		maxCommitTs = max(maxCommitTs, info.GetCommitTimestamp())
 	}
-	if maxCommitTs == 0 || fallbackStart.GetTimestamp() >= maxCommitTs {
-		return nil
-	}
-	return merr.WrapErrServiceInternalMsg(
-		"compaction fallback start position timestamp %d is earlier than max input commit timestamp %d",
-		fallbackStart.GetTimestamp(),
-		maxCommitTs)
+	return maxCommitTs
+}
+
+func getCompactionFallbackPositions(compactFromSegInfos []*SegmentInfo) (fallbackStart, fallbackDml *msgpb.MsgPosition) {
+	maxCommitTs := maxCommitTimestamp(compactFromSegInfos)
+
+	// Fallback positions are used only when output binlog timestamps are
+	// unavailable. Keep the raw minimum start timestamp so normal rows in mixed
+	// compaction still receive deletes after their original start. Normalize the
+	// fallback DML timestamp to commit_ts so temporal cleanup does not treat the
+	// compacted output as complete before committed import rows become visible.
+	fallbackStart = getMinPosition(lo.Map(compactFromSegInfos, func(info *SegmentInfo, _ int) *msgpb.MsgPosition {
+		return info.GetStartPosition()
+	}))
+	fallbackDml = normalizePositionTimestamp(getMaxPosition(lo.Map(compactFromSegInfos, func(info *SegmentInfo, _ int) *msgpb.MsgPosition {
+		return info.GetDmlPosition()
+	})), maxCommitTs)
+	return fallbackStart, fallbackDml
 }

 func (m *meta) completeClusterCompactionMutation(t *datapb.CompactionTask, result *datapb.CompactionPlanResult) ([]*SegmentInfo, *segMetricMutation, error) {
@@ -2336,22 +2344,7 @@ func (m *meta) completeClusterCompactionMutation(t *datapb.CompactionTask, resul
 		compactFromSegIDs = append(compactFromSegIDs, cloned.GetID())
 	}

-	// Compaction normalizes import segments: row timestamps in the output
-	// binlogs are already rewritten to commit_ts by the compactor, so
-	// recalculateSegmentPosition will pick up commit_ts from output binlogs.
-	// The fallback start position must not be earlier than any input import
-	// segment's commit_ts. The check below pins that invariant so a future
-	// change to sort-compaction position handling cannot silently expose
-	// compacted import rows too early.
-	fallbackStart := getMinPosition(lo.Map(compactFromSegInfos, func(info *SegmentInfo, _ int) *msgpb.MsgPosition {
-		return info.GetStartPosition()
-	}))
-	if err := validateCompactionFallbackStartPosition(compactFromSegInfos, fallbackStart); err != nil {
-		return nil, nil, err
-	}
-	fallbackDml := getMaxPosition(lo.Map(compactFromSegInfos, func(info *SegmentInfo, _ int) *msgpb.MsgPosition {
-		return info.GetDmlPosition()
-	}))
+	fallbackStart, fallbackDml := getCompactionFallbackPositions(compactFromSegInfos)

 	for _, seg := range result.GetSegments() {
 		startPos, dmlPos := recalculateSegmentPosition(seg.GetInsertLogs(), t.GetChannel(), fallbackStart, fallbackDml)
@@ -2463,23 +2456,7 @@ func (m *meta) completeMixCompactionMutation(
 	}
 	outputSchemaVersion := t.GetSchema().GetVersion()

-	// Compaction normalizes import segments: row timestamps in the output
-	// binlogs are already rewritten to commit_ts by the compactor, so the
-	// output segment is a normal segment with CommitTimestamp = 0.
-	// recalculateSegmentPosition will pick up commit_ts from output binlogs.
-	// The fallback start position must not be earlier than any input import
-	// segment's commit_ts. The check below pins that invariant so a future
-	// change to sort-compaction position handling cannot silently expose
-	// compacted import rows too early.
-	fallbackStart := getMinPosition(lo.Map(compactFromSegInfos, func(info *SegmentInfo, _ int) *msgpb.MsgPosition {
-		return info.GetStartPosition()
-	}))
-	if err := validateCompactionFallbackStartPosition(compactFromSegInfos, fallbackStart); err != nil {
-		return nil, nil, err
-	}
-	fallbackDml := getMaxPosition(lo.Map(compactFromSegInfos, func(info *SegmentInfo, _ int) *msgpb.MsgPosition {
-		return info.GetDmlPosition()
-	}))
+	fallbackStart, fallbackDml := getCompactionFallbackPositions(compactFromSegInfos)

 	compactToSegments := make([]*SegmentInfo, 0)
 	for _, compactToSegment := range result.GetSegments() {
@@ -1023,19 +1023,21 @@ func (suite *MetaBasicSuite) TestCompleteCompactionMutation() {
 		suite.EqualValues(0, infos[0].GetCommitTimestamp(), "sort compaction normalizes commit_timestamp after rewriting row timestamps")
 	})

-	suite.Run("mix compaction rejects stale import fallback start position", func() {
+	suite.Run("mix compaction preserves fallback start while normalizing fallback dml", func() {
 		latestSegments := NewSegmentsInfo()
 		latestSegments.SetSegment(1, &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
 			ID: 1, CollectionID: 100, PartitionID: 10,
 			State: commonpb.SegmentState_Flushed, Level: datapb.SegmentLevel_L1,
 			NumOfRows: 2, CommitTimestamp: 5000,
 			StartPosition: &msgpb.MsgPosition{ChannelName: "ch-1", Timestamp: 1000},
+			DmlPosition:   &msgpb.MsgPosition{ChannelName: "ch-1", Timestamp: 1500},
 		}})
 		latestSegments.SetSegment(2, &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
 			ID: 2, CollectionID: 100, PartitionID: 10,
 			State: commonpb.SegmentState_Flushed, Level: datapb.SegmentLevel_L1,
 			NumOfRows: 3, CommitTimestamp: 0,
 			StartPosition: &msgpb.MsgPosition{ChannelName: "ch-1", Timestamp: 2000},
+			DmlPosition:   &msgpb.MsgPosition{ChannelName: "ch-1", Timestamp: 3000},
 		}})

 		result := &datapb.CompactionPlanResult{
@@ -1053,9 +1055,11 @@ func (suite *MetaBasicSuite) TestCompleteCompactionMutation() {
 			chunkManager: mockChMgr,
 		}
 		infos, _, err := m.CompleteCompactionMutation(context.TODO(), task, result)
-		suite.Error(err)
-		suite.Contains(err.Error(), "earlier than max input commit timestamp")
-		suite.Nil(infos)
+		suite.NoError(err)
+		suite.Require().Equal(1, len(infos))
+		suite.EqualValues(1000, infos[0].GetStartPosition().GetTimestamp())
+		suite.EqualValues(5000, infos[0].GetDmlPosition().GetTimestamp())
+		suite.EqualValues(0, infos[0].GetCommitTimestamp())
 	})
 }

@@ -1404,6 +1408,68 @@ func (suite *MetaBasicSuite) TestCompleteCompactionMutation_RecalculatePositions
 		suite.Equal("ch-1", infos[0].GetStartPosition().GetChannelName())
 	})

+	suite.Run("mix_compaction_uses_output_timestamps_when_normal_segment_precedes_import_commit", func() {
+		latestSegments := NewSegmentsInfo()
+		for segID, segment := range map[UniqueID]*SegmentInfo{
+			1: {SegmentInfo: &datapb.SegmentInfo{
+				ID:              1,
+				CollectionID:    100,
+				PartitionID:     10,
+				State:           commonpb.SegmentState_Flushed,
+				Level:           datapb.SegmentLevel_L1,
+				Binlogs:         []*datapb.FieldBinlog{getFieldBinlogIDs(0, 10000)},
+				Statslogs:       []*datapb.FieldBinlog{getFieldBinlogIDs(0, 20000)},
+				NumOfRows:       2,
+				CommitTimestamp: 0,
+				StartPosition:   &msgpb.MsgPosition{ChannelName: "ch-1", Timestamp: 1000},
+				DmlPosition:     &msgpb.MsgPosition{ChannelName: "ch-1", Timestamp: 1500},
+			}},
+			2: {SegmentInfo: &datapb.SegmentInfo{
+				ID:              2,
+				CollectionID:    100,
+				PartitionID:     10,
+				State:           commonpb.SegmentState_Flushed,
+				Level:           datapb.SegmentLevel_L1,
+				Binlogs:         []*datapb.FieldBinlog{getFieldBinlogIDs(0, 11000)},
+				Statslogs:       []*datapb.FieldBinlog{getFieldBinlogIDs(0, 21000)},
+				NumOfRows:       3,
+				CommitTimestamp: 5000,
+				StartPosition:   &msgpb.MsgPosition{ChannelName: "ch-1", Timestamp: 2000},
+				DmlPosition:     &msgpb.MsgPosition{ChannelName: "ch-1", Timestamp: 2500},
+			}},
+		} {
+			latestSegments.SetSegment(segID, segment)
+		}
+
+		result := &datapb.CompactionPlanResult{
+			Segments: []*datapb.CompactionSegment{{
+				SegmentID:           3,
+				InsertLogs:          []*datapb.FieldBinlog{fieldBinlogWithTimestamps(0, 50000, 1000, 5000)},
+				Field2StatslogPaths: []*datapb.FieldBinlog{getFieldBinlogIDs(0, 50001)},
+				NumOfRows:           5,
+			}},
+		}
+		task := &datapb.CompactionTask{
+			InputSegments: []UniqueID{1, 2},
+			Type:          datapb.CompactionType_MixCompaction,
+			Channel:       "ch-1",
+			Schema:        &schemapb.CollectionSchema{Version: 1},
+		}
+		m := &meta{
+			catalog:      &datacoord.Catalog{MetaKv: NewMetaMemoryKV()},
+			segments:     latestSegments,
+			chunkManager: mockChMgr,
+		}
+
+		infos, _, err := m.CompleteCompactionMutation(context.TODO(), task, result)
+		suite.NoError(err)
+		suite.Require().Equal(1, len(infos))
+
+		suite.EqualValues(0, infos[0].GetCommitTimestamp())
+		suite.Equal(uint64(1000), infos[0].GetStartPosition().GetTimestamp())
+		suite.Equal(uint64(5000), infos[0].GetDmlPosition().GetTimestamp())
+	})
+
 	suite.Run("mix_compaction_fallback_when_no_timestamps", func() {
 		latestSegments := NewSegmentsInfo()
 		for segID, segment := range map[UniqueID]*SegmentInfo{