fix: support dropping and re-adding BM25 function field (#50735)

issue: #50729

### What
Dropping a BM25 function field and re-adding it with the same function
name and output field left the new sparse index stuck `InProgress`. The
flow was blocked on two layers, fixed here:

- **DataCoord**: accept expanded pre-allocated segment ID ranges when
completing the schema-bump replacement compaction (previously only a
single-ID range was allowed, causing `BumpSchemaVersionCompaction failed
... compaction plan illegal`). Still require the replacement segment to
use the range begin, matching DataNode full-rewrite behavior.
- **QueryNode**: allow a BM25 function field to be dropped and re-added
with a new output field ID (reject only incompatible in-place changes to
a shared output field), and prune dropped BM25 stats from
current/growing/sealed state and local disk so reclaimed disk size stays
accurate.

### Tests
- Cover expanded-range completion in the schema-bump metadata regression
test.
- Cover drop/re-add validation and dropped-stats pruning
(current/growing/sealed + on-disk) in the delegator and idf_oracle
tests.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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/meta.go

@@ -3329,8 +3329,8 @@ func (m *meta) completeBumpSchemaVersionReplacementMutation(
 	schemaVersion int32,
 ) ([]*SegmentInfo, *segMetricMutation, error) {
 	idRange := t.GetPreAllocatedSegmentIDs()
-	if idRange == nil || idRange.GetBegin()+1 != idRange.GetEnd() || resultSegment.GetSegmentID() != idRange.GetBegin() {
-		return nil, nil, merr.WrapErrIllegalCompactionPlanMsg("schema bump replacement result segment ID %d does not match the pre-allocated segment ID", resultSegment.GetSegmentID())
+	if idRange == nil || idRange.GetBegin() >= idRange.GetEnd() || resultSegment.GetSegmentID() != idRange.GetBegin() {
+		return nil, nil, merr.WrapErrIllegalCompactionPlanMsg("schema bump replacement result segment ID %d does not match the pre-allocated segment ID range", resultSegment.GetSegmentID())
 	}
 
 	dropped := oldSegment.Clone()

internal/datacoord/meta_test.go

@@ -1953,7 +1953,7 @@ func (suite *MetaBasicSuite) TestCompleteBumpSchemaVersionCompactionMutation() {
 		suite.Nil(mutation)
 	})
 
-	suite.Run("replacement result drops old segment and creates flushed new segment", func() {
+	suite.Run("replacement result accepts expanded preallocated segment ID range", func() {
 		m := &meta{
 			catalog:  &datacoord.Catalog{MetaKv: NewMetaMemoryKV()},
 			segments: makeSegments(1, commonpb.SegmentState_Flushed),
@@ -1964,7 +1964,7 @@ func (suite *MetaBasicSuite) TestCompleteBumpSchemaVersionCompactionMutation() {
 			Schema: &schemapb.CollectionSchema{
 				Version: 3,
 			},
-			PreAllocatedSegmentIDs: &datapb.IDRange{Begin: 2, End: 3},
+			PreAllocatedSegmentIDs: &datapb.IDRange{Begin: 2, End: 4},
 		}
 		oldBefore := m.segments.GetSegment(1)
 		oldBefore.InsertChannel = "test-channel"

internal/querynodev2/delegator/delegator.go

@@ -1241,9 +1241,9 @@ func (sd *shardDelegator) UpdateSchema(ctx context.Context, schema *schemapb.Col
 	oldSet := newBM25FunctionSet(sd.collection.Schema())
 	newSet := newBM25FunctionSet(schema)
 	idfOracle := sd.getIDFOracle()
-	if idfOracle != nil && !newSet.IsSupersetOf(oldSet) {
+	if idfOracle != nil && newSet.HasIncompatibleCommonFunction(oldSet) {
 		newFunctionState.Close()
-		return merr.WrapErrServiceInternal("unsupported non-additive BM25 function schema change on loaded collection")
+		return merr.WrapErrServiceInternal("unsupported incompatible BM25 function schema change on loaded collection")
 	}
 
 	// Keep the load barrier monotonic. A higher logical schema version can be

internal/querynodev2/delegator/delegator_test.go

@@ -2217,8 +2217,19 @@ func TestBM25FunctionSetAdditiveValidation(t *testing.T) {
 
 	changed := proto.Clone(newBM25FunctionSchema()).(*schemapb.FunctionSchema)
 	changed.InputFieldIds = []int64{103}
-	assert.False(t, newBM25FunctionSet(newFunctionRuntimeTestSchema(changed)).IsSupersetOf(base))
-	assert.False(t, newBM25FunctionSet(newFunctionRuntimeTestSchema()).IsSupersetOf(base))
+	changedSet := newBM25FunctionSet(newFunctionRuntimeTestSchema(changed))
+	assert.False(t, changedSet.IsSupersetOf(base))
+	assert.True(t, changedSet.HasIncompatibleCommonFunction(base))
+
+	droppedSet := newBM25FunctionSet(newFunctionRuntimeTestSchema())
+	assert.False(t, droppedSet.IsSupersetOf(base))
+	assert.False(t, droppedSet.HasIncompatibleCommonFunction(base))
+
+	readded := proto.Clone(newBM25FunctionSchema()).(*schemapb.FunctionSchema)
+	readded.OutputFieldIds = []int64{104}
+	readdedSet := newBM25FunctionSet(newFunctionRuntimeTestSchema(readded))
+	assert.False(t, readdedSet.IsSupersetOf(base))
+	assert.False(t, readdedSet.HasIncompatibleCommonFunction(base))
 }
 
 func TestBM25FunctionSetIgnoresAnalyzerParams(t *testing.T) {
@@ -2252,7 +2263,7 @@ func TestBM25FunctionSetNormalizesFunctionParamOrder(t *testing.T) {
 	assert.True(t, changed.IsSupersetOf(base))
 }
 
-func TestUpdateSchemaRejectsNonAdditiveBM25FunctionChange(t *testing.T) {
+func TestUpdateSchemaRejectsIncompatibleBM25FunctionChange(t *testing.T) {
 	paramtable.Init()
 	paramtable.SetNodeID(1)
 	workerManager := cluster.NewMockManager(t)
@@ -2279,7 +2290,7 @@ func TestUpdateSchemaRejectsNonAdditiveBM25FunctionChange(t *testing.T) {
 	changed.InputFieldIds = []int64{103}
 	err := sd.UpdateSchema(context.Background(), newFunctionRuntimeTestSchemaWithVersion(1, changed), 100)
 	require.Error(t, err)
-	assert.Contains(t, err.Error(), "unsupported non-additive BM25 function schema change")
+	assert.Contains(t, err.Error(), "unsupported incompatible BM25 function schema change")
 	assert.True(t, sd.functionState.hasFunctionType(102, schemapb.FunctionType_BM25))
 	assert.Same(t, oldOracle, sd.getIDFOracle())
 }
@@ -2432,7 +2443,7 @@ func TestUpdateSchemaRefreshesCollectionBaselineForSequentialBM25Validation(t *t
 	changed.InputFieldIds = []int64{103}
 	err = sd.UpdateSchema(context.Background(), newFunctionRuntimeTestSchemaWithVersion(2, changed), 200)
 	require.Error(t, err)
-	assert.Contains(t, err.Error(), "unsupported non-additive BM25 function schema change")
+	assert.Contains(t, err.Error(), "unsupported incompatible BM25 function schema change")
 	require.Equal(t, uint64(1), sd.collection.SchemaVersion())
 }
 

internal/querynodev2/delegator/idf_oracle.go

@@ -117,6 +117,19 @@ func (s bm25FunctionSet) IsSupersetOf(old bm25FunctionSet) bool {
 	return true
 }
 
+func (s bm25FunctionSet) HasIncompatibleCommonFunction(old bm25FunctionSet) bool {
+	// Do not reuse IsSupersetOf here: loaded collections must allow BM25
+	// function fields to be dropped and re-added with new output field IDs,
+	// while still rejecting in-place changes to an existing output field.
+	for outputFieldID, newFunction := range s {
+		oldFunction, ok := old[outputFieldID]
+		if ok && !sameBM25Function(newFunction, oldFunction) {
+			return true
+		}
+	}
+	return false
+}
+
 func (s bm25FunctionSet) Equal(other bm25FunctionSet) bool {
 	return len(s) == len(other) && s.IsSupersetOf(other)
 }
@@ -208,6 +221,40 @@ func (s *sealedBm25Stats) addFieldsLocked(fieldIDs []int64) {
 	}
 }
 
+func (s *sealedBm25Stats) RetainFields(fieldIDs map[int64]struct{}) int64 {
+	s.Lock()
+	defer s.Unlock()
+
+	kept := s.fieldList[:0]
+	removedDiskSize := int64(0)
+	for _, fieldID := range s.fieldList {
+		if _, ok := fieldIDs[fieldID]; ok {
+			kept = append(kept, fieldID)
+			continue
+		}
+		if s.localDir == "" {
+			continue
+		}
+		fieldDir := path.Join(s.localDir, fmt.Sprintf("%d", fieldID))
+		fieldDiskSize := bm25FieldDirDiskSize(fieldDir)
+		if err := os.RemoveAll(fieldDir); err != nil {
+			// Removal failed: the files remain on disk, so keep tracking the
+			// field and do not count its bytes as freed — tracked disk usage
+			// must match what is physically present.
+			mlog.Warn(context.TODO(), "remove dropped bm25 stats field failed", mlog.Err(err), mlog.String("path", fieldDir))
+			kept = append(kept, fieldID)
+			continue
+		}
+		removedDiskSize += fieldDiskSize
+	}
+	s.fieldList = kept
+	if removedDiskSize > s.diskSize {
+		removedDiskSize = s.diskSize
+	}
+	s.diskSize -= removedDiskSize
+	return removedDiskSize
+}
+
 func (s *sealedBm25Stats) FieldList() []int64 {
 	s.RLock()
 	defer s.RUnlock()
@@ -285,16 +332,34 @@ func newBm25Stats(functions []*schemapb.FunctionSchema) bm25Stats {
 	return stats
 }
 
-func (s bm25Stats) AddMissingFunctions(functions []*schemapb.FunctionSchema) {
+func bm25FunctionFieldIDs(functions []*schemapb.FunctionSchema) map[int64]struct{} {
+	fieldIDs := make(map[int64]struct{})
 	for _, function := range functions {
 		if function.GetType() != schemapb.FunctionType_BM25 || len(function.GetOutputFieldIds()) == 0 {
 			continue
 		}
-		fieldID := function.GetOutputFieldIds()[0]
+		fieldIDs[function.GetOutputFieldIds()[0]] = struct{}{}
+	}
+	return fieldIDs
+}
+
+func (s bm25Stats) SyncFunctions(functions []*schemapb.FunctionSchema) map[int64]struct{} {
+	fieldIDs := bm25FunctionFieldIDs(functions)
+	s.RetainFields(fieldIDs)
+	for fieldID := range fieldIDs {
 		if _, ok := s[fieldID]; !ok {
 			s[fieldID] = storage.NewBM25Stats()
 		}
 	}
+	return fieldIDs
+}
+
+func (s bm25Stats) RetainFields(fieldIDs map[int64]struct{}) {
+	for fieldID := range s {
+		if _, ok := fieldIDs[fieldID]; !ok {
+			delete(s, fieldID)
+		}
+	}
 }
 
 type idfTarget struct {
@@ -411,7 +476,18 @@ func (o *idfOracle) RegisterGrowing(segmentID int64, stats bm25Stats) {
 
 func (o *idfOracle) SyncFunctions(functions []*schemapb.FunctionSchema) error {
 	o.Lock()
-	o.current.AddMissingFunctions(functions)
+	fieldIDs := o.current.SyncFunctions(functions)
+	for _, stats := range o.growing {
+		stats.RetainFields(fieldIDs)
+	}
+	removedDiskSize := int64(0)
+	o.sealed.Range(func(_ int64, stats *sealedBm25Stats) bool {
+		removedDiskSize += stats.RetainFields(fieldIDs)
+		return true
+	})
+	if removedDiskSize > 0 {
+		o.sealedDiskSize.Add(-removedDiskSize)
+	}
 	o.Unlock()
 	o.syncResource()
 	return nil
@@ -611,6 +687,27 @@ type streamLoadResult struct {
 	diskSize  int64
 }
 
+func bm25FieldDirDiskSize(fieldDir string) int64 {
+	entries, err := os.ReadDir(fieldDir)
+	if err != nil {
+		return 0
+	}
+
+	size := int64(0)
+	for _, entry := range entries {
+		if entry.IsDir() {
+			continue
+		}
+		info, err := entry.Info()
+		if err != nil {
+			mlog.Warn(context.TODO(), "stat bm25 stats field file failed", mlog.Err(err), mlog.String("path", path.Join(fieldDir, entry.Name())))
+			continue
+		}
+		size += info.Size()
+	}
+	return size
+}
+
 // streamLoad downloads BM25 stats from remote storage to local disk.
 // When needParse is true, also parses stats using TeeReader.
 func (o *idfOracle) streamLoad(ctx context.Context, segmentID int64, binlogPaths map[int64][]string, cm storage.ChunkManager, needParse bool) (streamLoadResult, error) {

internal/querynodev2/delegator/idf_oracle_test.go

@@ -1004,3 +1004,90 @@ func TestBuildIDFNewFieldAfterSyncFunctions(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, float64(0), avgdl)
 }
+
+func TestIDFSyncFunctionsPrunesDroppedBM25Field(t *testing.T) {
+	idfOracle := NewIDFOracle("test-channel", []*schemapb.FunctionSchema{
+		{Type: schemapb.FunctionType_BM25, InputFieldIds: []int64{101}, OutputFieldIds: []int64{102}},
+		{Type: schemapb.FunctionType_BM25, InputFieldIds: []int64{103}, OutputFieldIds: []int64{104}},
+	}).(*idfOracle)
+	idfOracle.dirPath = t.TempDir()
+	idfOracle.Start()
+	defer idfOracle.Close()
+
+	droppedStats := genBM25StatsForField(102, 1, 2)[102]
+	keptStats := genBM25StatsForField(104, 3, 4)[104]
+	idfOracle.current[102] = droppedStats
+	idfOracle.current[104] = keptStats
+	idfOracle.growing[1] = &growingBm25Stats{
+		bm25Stats: bm25Stats{
+			102: droppedStats.Clone(),
+			104: keptStats.Clone(),
+		},
+		activate: true,
+	}
+
+	segDir := path.Join(idfOracle.dirPath, "10")
+	droppedDir := path.Join(segDir, "102")
+	keptDir := path.Join(segDir, "104")
+	require.NoError(t, os.MkdirAll(droppedDir, os.ModePerm))
+	require.NoError(t, os.MkdirAll(keptDir, os.ModePerm))
+	droppedBytes := []byte("dropped")
+	keptBytes := []byte("kept")
+	require.NoError(t, os.WriteFile(path.Join(droppedDir, "0.data"), droppedBytes, 0o600))
+	require.NoError(t, os.WriteFile(path.Join(keptDir, "0.data"), keptBytes, 0o600))
+	idfOracle.sealed.Insert(10, &sealedBm25Stats{
+		activate:  atomic.NewBool(true),
+		segmentID: 10,
+		localDir:  segDir,
+		fieldList: []int64{102, 104},
+		diskSize:  int64(len(droppedBytes) + len(keptBytes)),
+	})
+	idfOracle.sealedDiskSize.Store(int64(len(droppedBytes) + len(keptBytes)))
+
+	err := idfOracle.SyncFunctions([]*schemapb.FunctionSchema{
+		{Type: schemapb.FunctionType_BM25, InputFieldIds: []int64{103}, OutputFieldIds: []int64{104}},
+	})
+	require.NoError(t, err)
+
+	_, err = idfOracle.current.GetStats(102)
+	require.Error(t, err)
+	existing, err := idfOracle.current.GetStats(104)
+	require.NoError(t, err)
+	assert.Same(t, keptStats, existing)
+	_, ok := idfOracle.growing[1].bm25Stats[102]
+	assert.False(t, ok)
+	_, ok = idfOracle.growing[1].bm25Stats[104]
+	assert.True(t, ok)
+
+	sealedStats, ok := idfOracle.sealed.Get(10)
+	require.True(t, ok)
+	assert.ElementsMatch(t, []int64{104}, sealedStats.FieldList())
+	_, err = os.Stat(droppedDir)
+	assert.True(t, os.IsNotExist(err))
+	_, err = os.Stat(keptDir)
+	assert.NoError(t, err)
+	assert.Equal(t, int64(len(keptBytes)), sealedStats.diskSize)
+	assert.Equal(t, int64(len(keptBytes)), idfOracle.sealedDiskSize.Load())
+}
+
+func TestIDFSyncFunctionsPrunesAllDroppedBM25Fields(t *testing.T) {
+	idfOracle := NewIDFOracle("test-channel", []*schemapb.FunctionSchema{{
+		Type:           schemapb.FunctionType_BM25,
+		InputFieldIds:  []int64{101},
+		OutputFieldIds: []int64{102},
+	}}).(*idfOracle)
+	idfOracle.dirPath = t.TempDir()
+	idfOracle.Start()
+	defer idfOracle.Close()
+
+	stats := genBM25StatsForField(102, 1, 2)[102]
+	idfOracle.current[102] = stats
+	idfOracle.growing[1] = &growingBm25Stats{bm25Stats: bm25Stats{102: stats.Clone()}, activate: true}
+
+	err := idfOracle.SyncFunctions(nil)
+	require.NoError(t, err)
+
+	_, err = idfOracle.current.GetStats(102)
+	require.Error(t, err)
+	assert.Empty(t, idfOracle.growing[1].bm25Stats)
+}