fix: CAS the schema-bump in-place manifest commit to stop dropping a concurrent index (#51376) (#51377)

issue: #51376

## What this fixes

The in-place partial-materialization path of
`bump_schema_version_compaction` commits onto a live segment's manifest,
and DataCoord adopted the result on a version-newer check only — it
never verified the result was built on the segment's *current* manifest.
A bump built on a plan-pinned base could therefore be adopted after a
concurrent stats/index task advanced the manifest pointer, silently
dropping that index (correct version number, wrong content lineage). The
loss is permanent and non-self-healing, because the segment meta stats
placeholder stops the stats inspector from re-triggering.

## Fix — application-layer adoption CAS (mirrors the stats path)

- The schema-bump result carries the plan-pinned base manifest
(`CompactionSegment.base_manifest`, like `StatsResult.base_manifest`);
the DataNode fills it with the manifest it materialized on.
- DataCoord adopts the in-place result only if `result.base_manifest ==
segment.ManifestPath` (the current pointer); a missing or stale base is
rejected (`ErrIllegalCompactionPlan`) so the task re-triggers and
rebuilds on the current manifest instead of overwriting the concurrent
commit. This mirrors the stats path's `errStatsResultStale`.
- The storage commit stays on `OverwriteResolver` (unchanged). Because
the CAS is at the adoption layer (against the meta pointer, not
storage-latest), a result lost after a DataNode crash self-heals on
retry: the pointer stays put, so the retry re-pins the same base and is
adopted — no wedge.
- The bump policy is restricted to internal collections (matching every
sibling compaction policy), so a rejected attempt's orphan is reclaimed
by internal dropped-segment GC and bump never churns on external
collections.

## Verification
Both interleavings converge with no data loss: stats-first → bump's base
no longer matches the advanced pointer → reject + retrigger on the
current manifest; bump-first → stats' existing base-check rejects and
retries. Added meta-level adoption tests (matching/stale base) and
updated the in-place adoption tests to carry the base.

Signed-off-by: MrPresent-Han <chun.han@gmail.com>
Co-authored-by: MrPresent-Han <chun.han@gmail.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

internal/datacoord/compaction_policy_bump_schema_version.go

@@ -92,6 +92,10 @@ func (policy *bumpSchemaVersionPolicy) Trigger(ctx context.Context) (map[Compact
 		if collection.Schema == nil {
 			continue
 		}
+		if collection.IsExternal() {
+			mlog.Info(ctx, "skip schema bump compaction trigger for external collection", mlog.FieldCollectionID(collection.ID))
+			continue
+		}
 		if policy.meta.isCollectionCompactionBlocked(collection.ID) {
 			mlog.Info(ctx, "skip schema bump compaction for collection due to snapshot compaction block",
 				mlog.FieldCollectionID(collection.ID))

internal/datacoord/compaction_policy_bump_schema_version_test.go

@@ -369,6 +369,20 @@ func (s *BumpSchemaVersionPolicySuite) TestTriggerSkipsSnapshotProtectedSegment(
 	s.Empty(events[TriggerTypeBumpSchemaVersion])
 }
 
+func (s *BumpSchemaVersionPolicySuite) TestTriggerSkipsExternalCollection() {
+	ctx := context.Background()
+	collID := int64(100)
+	coll := newBumpSchemaVersionTestCollection(collID, 2)
+	// IsExternalCollection is true when any field carries an external field mapping.
+	coll.Schema.Fields[1].ExternalField = "src_text"
+	s.bumpSchemaVersionPolicy.meta.collections.Insert(collID, coll)
+	s.bumpSchemaVersionPolicy.meta.segments.SetSegment(101, newBumpSchemaVersionTestSegment(collID, 101, 1, storage.StorageV3, "manifest"))
+
+	events, err := s.bumpSchemaVersionPolicy.Trigger(ctx)
+	s.NoError(err)
+	s.Empty(events[TriggerTypeBumpSchemaVersion])
+}
+
 func (s *BumpSchemaVersionPolicySuite) TestName() {
 	// Test Name method
 	s.Equal("BumpSchemaVersion", s.bumpSchemaVersionPolicy.Name())

internal/datacoord/compaction_task_bump_schema_version_test.go

@@ -504,6 +504,7 @@ func (s *BumpSchemaVersionCompactionTaskSuite) TestQueryTaskOnWorker() {
 				SegmentID:      segmentID,
 				InsertLogs:     []*datapb.FieldBinlog{},
 				Manifest:       manifest,
+				BaseManifest:   manifest,
 				StorageVersion: storage.StorageV3,
 			}},
 		}, nil).Once()
@@ -757,6 +758,7 @@ func (s *BumpSchemaVersionCompactionTaskSuite) TestSaveSegmentMeta() {
 					SegmentID:      segmentID,
 					StorageVersion: storage.StorageV3,
 					Manifest:       resultManifest,
+					BaseManifest:   currentManifest,
 					InsertLogs: []*datapb.FieldBinlog{
 						{FieldID: 101, Binlogs: []*datapb.Binlog{{LogID: 1000, EntriesNum: 1000}}},
 						{FieldID: 102, Binlogs: []*datapb.Binlog{{LogID: 2000, EntriesNum: 1000}}},

internal/datacoord/meta.go

@@ -3328,13 +3328,32 @@ func (m *meta) completeBumpSchemaVersionCompactionMutation(
 	if currentManifest == "" {
 		return nil, nil, merr.WrapErrIllegalCompactionPlan("schema bump compaction input segment should contain a StorageV3 manifest")
 	}
-	if currentManifest != resultManifest {
-		manifestCompare, err := packed.CompareManifestPath(resultManifest, currentManifest)
-		if err != nil {
-			return nil, nil, merr.WrapErrIllegalCompactionPlanMsg("schema bump compaction result manifest is not comparable with current manifest: %v", err)
+	// Optimistic-concurrency CAS on the manifest pointer. Adopt the in-place
+	// result only when it is a valid successor of the current pointer:
+	//   - result == current: idempotent replay of an adoption whose task state
+	//     was lost to a crash after AlterSegments but before meta_saved.
+	//   - base == current AND result strictly newer on the same base path: a
+	//     fresh forward commit built on the current pointer.
+	// Reject anything else — a stale base (a concurrent stats/index/bump commit
+	// advanced the pointer), a rollback, a different base path, or an unparsable
+	// result — so the task re-triggers and rebuilds on the current manifest
+	// instead of overwriting the concurrent commit (mirrors the stats path's
+	// errStatsResultStale). The check also self-heals a lost result: the pointer
+	// stays put, so the retry re-pins the same base.
+	baseManifest := resultSegment.GetBaseManifest()
+	if baseManifest == "" {
+		return nil, nil, merr.WrapErrIllegalCompactionPlan("schema bump result missing base manifest")
+	}
+	if resultManifest != currentManifest {
+		if baseManifest != currentManifest {
+			return nil, nil, merr.WrapErrIllegalCompactionPlanMsg("schema bump result base manifest %s no longer matches current %s", baseManifest, currentManifest)
 		}
-		if manifestCompare <= 0 {
-			return nil, nil, merr.WrapErrIllegalCompactionPlan("schema bump compaction result manifest is not newer than current manifest")
+		cmp, err := packed.CompareManifestPath(resultManifest, currentManifest)
+		if err != nil {
+			return nil, nil, merr.WrapErrIllegalCompactionPlanMsg("schema bump result manifest %s not comparable with current %s: %v", resultManifest, currentManifest, err)
+		}
+		if cmp <= 0 {
+			return nil, nil, merr.WrapErrIllegalCompactionPlanMsg("schema bump result manifest %s does not advance current %s", resultManifest, currentManifest)
 		}
 	}
 

internal/datacoord/meta_test.go

@@ -1928,6 +1928,161 @@ func (suite *MetaBasicSuite) TestCompleteBumpSchemaVersionCompactionMutation() {
 		suite.Nil(mutation)
 	})
 
+	suite.Run("in-place result with matching base manifest is adopted", func() {
+		currentManifest := packed.MarshalManifestPath("/data/segments/1", 10)
+		resultManifest := packed.MarshalManifestPath("/data/segments/1", 12)
+		segs := makeSegments(1, commonpb.SegmentState_Flushed)
+		old := segs.GetSegment(1)
+		old.StorageVersion = storage.StorageV3
+		old.ManifestPath = currentManifest
+		m := &meta{
+			catalog:  &datacoord.Catalog{MetaKv: NewMetaMemoryKV()},
+			segments: segs,
+		}
+		task := &datapb.CompactionTask{
+			InputSegments: []int64{1},
+			Type:          datapb.CompactionType_BumpSchemaVersionCompaction,
+			Schema:        &schemapb.CollectionSchema{Version: 3},
+		}
+		result := &datapb.CompactionPlanResult{
+			Segments: []*datapb.CompactionSegment{
+				{
+					SegmentID:      1,
+					NumOfRows:      5,
+					InsertLogs:     []*datapb.FieldBinlog{getFieldBinlogIDs(0, 10001)},
+					Manifest:       resultManifest,
+					BaseManifest:   currentManifest, // == current pointer and advances it, adopt
+					StorageVersion: storage.StorageV3,
+				},
+			},
+		}
+		infos, mutation, err := m.completeBumpSchemaVersionCompactionMutation(task, result)
+		suite.NoError(err)
+		suite.NotNil(mutation)
+		suite.Require().Len(infos, 1)
+		suite.EqualValues(1, infos[0].GetID())
+		suite.Equal(resultManifest, infos[0].GetManifestPath())
+	})
+
+	suite.Run("in-place result with stale base manifest is rejected", func() {
+		segs := makeSegments(1, commonpb.SegmentState_Flushed)
+		old := segs.GetSegment(1)
+		old.StorageVersion = storage.StorageV3
+		// pointer advanced past the result's base
+		old.ManifestPath = packed.MarshalManifestPath("/data/segments/1", 11)
+		m := &meta{
+			catalog:  &datacoord.Catalog{MetaKv: NewMetaMemoryKV()},
+			segments: segs,
+		}
+		task := &datapb.CompactionTask{
+			InputSegments: []int64{1},
+			Type:          datapb.CompactionType_BumpSchemaVersionCompaction,
+			Schema:        &schemapb.CollectionSchema{Version: 3},
+		}
+		result := &datapb.CompactionPlanResult{
+			Segments: []*datapb.CompactionSegment{
+				{
+					SegmentID:      1,
+					NumOfRows:      5,
+					Manifest:       packed.MarshalManifestPath("/data/segments/1", 12),
+					BaseManifest:   packed.MarshalManifestPath("/data/segments/1", 10), // stale, != current v11
+					StorageVersion: storage.StorageV3,
+				},
+			},
+		}
+		infos, mutation, err := m.completeBumpSchemaVersionCompactionMutation(task, result)
+		suite.Error(err)
+		suite.ErrorIs(err, merr.ErrIllegalCompactionPlan)
+		suite.Nil(infos)
+		suite.Nil(mutation)
+	})
+
+	// Defensive: base == current only claims the pointer has not drifted; it does
+	// not independently validate the worker's result. A buggy worker that reports
+	// the matching base but produces a result that does not advance the current
+	// manifest on the same base path (rollback / equal version / different base
+	// path / unparsable) must be rejected, not adopted.
+	suite.Run("in-place result on current base that does not advance is rejected", func() {
+		currentManifest := packed.MarshalManifestPath("/data/segments/1", 10)
+		cases := map[string]string{
+			"rollback":       packed.MarshalManifestPath("/data/segments/1", 9),
+			"different base": packed.MarshalManifestPath("/data/segments/2", 12),
+			"unparsable":     "not-a-manifest",
+		}
+		for name, resultManifest := range cases {
+			suite.Run(name, func() {
+				segs := makeSegments(1, commonpb.SegmentState_Flushed)
+				old := segs.GetSegment(1)
+				old.StorageVersion = storage.StorageV3
+				old.ManifestPath = currentManifest
+				m := &meta{
+					catalog:  &datacoord.Catalog{MetaKv: NewMetaMemoryKV()},
+					segments: segs,
+				}
+				task := &datapb.CompactionTask{
+					InputSegments: []int64{1},
+					Type:          datapb.CompactionType_BumpSchemaVersionCompaction,
+					Schema:        &schemapb.CollectionSchema{Version: 3},
+				}
+				result := &datapb.CompactionPlanResult{
+					Segments: []*datapb.CompactionSegment{
+						{
+							SegmentID:      1,
+							NumOfRows:      5,
+							Manifest:       resultManifest,
+							BaseManifest:   currentManifest, // == current pointer, but result does not advance it
+							StorageVersion: storage.StorageV3,
+						},
+					},
+				}
+				infos, mutation, err := m.completeBumpSchemaVersionCompactionMutation(task, result)
+				suite.Error(err)
+				suite.ErrorIs(err, merr.ErrIllegalCompactionPlan)
+				suite.Nil(infos)
+				suite.Nil(mutation)
+			})
+		}
+	})
+
+	suite.Run("in-place result already at current manifest is idempotently adopted", func() {
+		// Crash-replay: an adoption already succeeded (segment is at the
+		// materialized manifest) but the task state was lost before meta_saved.
+		// Replaying the same result — stale base, but manifest already == current
+		// — must be accepted, not rejected, so the already-succeeded task is not
+		// spuriously failed.
+		segs := makeSegments(1, commonpb.SegmentState_Flushed)
+		old := segs.GetSegment(1)
+		old.StorageVersion = storage.StorageV3
+		old.ManifestPath = "base/manifest-12" // already adopted the materialized manifest
+		old.SchemaVersion = 3
+		m := &meta{
+			catalog:  &datacoord.Catalog{MetaKv: NewMetaMemoryKV()},
+			segments: segs,
+		}
+		task := &datapb.CompactionTask{
+			InputSegments: []int64{1},
+			Type:          datapb.CompactionType_BumpSchemaVersionCompaction,
+			Schema:        &schemapb.CollectionSchema{Version: 3},
+		}
+		result := &datapb.CompactionPlanResult{
+			Segments: []*datapb.CompactionSegment{
+				{
+					SegmentID:      1,
+					NumOfRows:      5,
+					InsertLogs:     []*datapb.FieldBinlog{getFieldBinlogIDs(0, 10000)},
+					Manifest:       "base/manifest-12", // == current pointer
+					BaseManifest:   "base/manifest-10", // stale base, but result already current
+					StorageVersion: storage.StorageV3,
+				},
+			},
+		}
+		infos, mutation, err := m.completeBumpSchemaVersionCompactionMutation(task, result)
+		suite.NoError(err)
+		suite.NotNil(mutation)
+		suite.Require().Len(infos, 1)
+		suite.Equal("base/manifest-12", infos[0].GetManifestPath())
+	})
+
 	suite.Run("replacement result without preallocated ID rejected", func() {
 		m := &meta{
 			catalog:  &datacoord.Catalog{MetaKv: NewMetaMemoryKV()},
@@ -2362,6 +2517,7 @@ func (suite *MetaBasicSuite) TestCompleteBumpSchemaVersionCompactionMutation() {
 					SegmentID:      1,
 					InsertLogs:     []*datapb.FieldBinlog{getFieldBinlogIDs(0, 10001)},
 					Manifest:       resultManifest,
+					BaseManifest:   currentManifest,
 					StorageVersion: storage.StorageV3,
 				},
 			},
@@ -2408,6 +2564,7 @@ func (suite *MetaBasicSuite) TestCompleteBumpSchemaVersionCompactionMutation() {
 					SegmentID:      1,
 					InsertLogs:     segment.GetBinlogs(),
 					Manifest:       manifestPath,
+					BaseManifest:   manifestPath,
 					StorageVersion: storage.StorageV3,
 				},
 			},
@@ -2443,6 +2600,7 @@ func (suite *MetaBasicSuite) TestCompleteBumpSchemaVersionCompactionMutation() {
 					SegmentID:      1,
 					InsertLogs:     segment.GetBinlogs(),
 					Manifest:       manifestPath,
+					BaseManifest:   manifestPath,
 					StorageVersion: storage.StorageV3,
 				},
 			},
@@ -2544,6 +2702,7 @@ func (suite *MetaBasicSuite) TestCompleteBumpSchemaVersionCompactionMutation() {
 					SegmentID:      1,
 					InsertLogs:     []*datapb.FieldBinlog{getFieldBinlogIDs(0, 10001)},
 					Manifest:       resultManifest,
+					BaseManifest:   currentManifest,
 					StorageVersion: storage.StorageV3,
 				},
 			},
@@ -2796,6 +2955,7 @@ func (suite *MetaBasicSuite) TestCompleteCompactionMutation_DispatchesBumpSchema
 				SegmentID:      1,
 				InsertLogs:     []*datapb.FieldBinlog{getFieldBinlogIDs(0, 10001)},
 				Manifest:       manifestPath,
+				BaseManifest:   manifestPath,
 				StorageVersion: storage.StorageV3,
 			},
 		},

internal/datanode/compactor/bump_schema_version_compactor.go

@@ -410,6 +410,7 @@ func (t *bumpSchemaVersionCompactionTask) runSchemaVersionBumpOnly() *datapb.Com
 				Channel:             segment.GetInsertChannel(),
 				StorageVersion:      segment.GetStorageVersion(),
 				Manifest:            segment.GetManifest(),
+				BaseManifest:        segment.GetManifest(),
 				ExpirQuantiles:      segment.GetExpirQuantiles(),
 			},
 		},
@@ -1124,6 +1125,7 @@ func (t *bumpSchemaVersionCompactionTask) runMissingFunctionMaterialization(ctx
 				Channel:             segment.GetInsertChannel(),
 				StorageVersion:      writerResult.storageVersion,
 				Manifest:            manifestPath,
+				BaseManifest:        segment.GetManifest(),
 			},
 		},
 		Type: t.plan.GetType(),

pkg/proto/data_coord.proto

@@ -788,6 +788,11 @@ message CompactionSegment {
   string manifest = 12;
   repeated int64 expirQuantiles = 13;
   bool is_sorted_by_namespace = 14;
+  // base_manifest is the manifest the in-place schema-bump result was built
+  // on (the plan-pinned base). DataCoord adopts the result only if it still
+  // equals the segment's current manifest (optimistic-concurrency CAS),
+  // mirroring StatsResult.base_manifest.
+  string base_manifest = 15;
 }
 
 message CompactionPlanResult {