enhance: isolate function runner lifecycle keys (#50604)

relate: #49716
pr: https://github.com/milvus-io/milvus/pull/49717

## Summary
Split FunctionRunnerManager lifecycle references between WAL and
delegator keys while still sharing runners by function signature. Add
latest-version lookup support for schemaVersion 0 or nil schema in the
manager; master call sites still pass the current schema version
explicitly.

Signed-off-by: aoiasd <zhicheng.yue@zilliz.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

internal/flushcommon/pipeline/flow_graph_write_node.go

@@ -17,7 +17,6 @@ import (
 	"github.com/milvus-io/milvus/internal/util/function"
 	"github.com/milvus-io/milvus/internal/util/streamingutil"
 	"github.com/milvus-io/milvus/pkg/v3/mlog"
-	"github.com/milvus-io/milvus/pkg/v3/mq/msgstream"
 	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
 )
@@ -32,8 +31,7 @@ type writeNode struct {
 	metacache    metacache.MetaCache
 	pkField      *schemapb.FieldSchema
 
-	functionRunners        map[int32][]function.FunctionRunner
-	functionOutputFieldIDs map[int32][]int64
+	functionStore *function.FunctionRunnerLocalStore
 }
 
 // Name returns node name, implementing flowgraph.Node
@@ -46,55 +44,7 @@ func (wNode *writeNode) Free() {
 }
 
 func (wNode *writeNode) releaseFunctionRunners() {
-	for _, runners := range wNode.functionRunners {
-		function.CloseRunners(runners)
-	}
-	wNode.functionRunners = make(map[int32][]function.FunctionRunner)
-	wNode.functionOutputFieldIDs = make(map[int32][]int64)
-}
-
-func (wNode *writeNode) getEmbeddingOutputFieldIDs(schema *schemapb.CollectionSchema) ([]int64, error) {
-	schemaVersion := schema.GetVersion()
-	if outputFieldIDs, ok := wNode.functionOutputFieldIDs[schemaVersion]; ok {
-		return outputFieldIDs, nil
-	}
-
-	if !function.HasEmbeddingFunctions(schema) {
-		wNode.functionOutputFieldIDs[schemaVersion] = nil
-		return nil, nil
-	}
-	outputFieldIDs, err := function.EmbeddingOutputFieldIDs(schema)
-	if err != nil {
-		return nil, err
-	}
-	wNode.functionOutputFieldIDs[schemaVersion] = outputFieldIDs
-	return outputFieldIDs, nil
-}
-
-// fillEmbeddingData is only used to handle old insert messages that were not embedded before WAL append.
-func (wNode *writeNode) fillEmbeddingData(schema *schemapb.CollectionSchema, msg *msgstream.InsertMsg) error {
-	if !function.HasEmbeddingFunctions(schema) {
-		return nil
-	}
-	schemaVersion := schema.GetVersion()
-	_, ok, err := function.TryMaterialize(wNode.collectionID, schemaVersion, msg.InsertRequest)
-	if err != nil {
-		return err
-	}
-	if ok {
-		return nil
-	}
-
-	runners, ok := wNode.functionRunners[schemaVersion]
-	if !ok {
-		runners, err = function.BuildEmbeddingRunners(schema)
-		if err != nil {
-			return err
-		}
-		wNode.functionRunners[schemaVersion] = runners
-	}
-	_, err = function.FillFunctionFields(runners, msg.InsertRequest)
-	return err
+	wNode.functionStore.Close()
 }
 
 func (wNode *writeNode) Operate(in []Msg) []Msg {
@@ -143,7 +93,7 @@ func (wNode *writeNode) Operate(in []Msg) []Msg {
 	start, end := fgMsg.StartPositions[0], fgMsg.EndPositions[0]
 	currentSchema := wNode.metacache.GetSchema(fgMsg.TimeTick())
 	schemaVersion := currentSchema.GetVersion()
-	functionOutputFieldIDs, err := wNode.getEmbeddingOutputFieldIDs(currentSchema)
+	functionOutputFieldIDs, err := wNode.functionStore.OutputFieldIDs(currentSchema)
 	if err != nil {
 		mlog.Error(context.TODO(), "failed to get embedding output fields", mlog.Err(err))
 		panic(err)
@@ -155,7 +105,7 @@ func (wNode *writeNode) Operate(in []Msg) []Msg {
 			if len(functionOutputFieldIDs) == 0 || function.HasAllFieldDataByID(msg.GetFieldsData(), functionOutputFieldIDs) {
 				continue
 			}
-			if err := wNode.fillEmbeddingData(currentSchema, msg); err != nil {
+			if err := wNode.functionStore.FillEmbeddingData(wNode.collectionID, currentSchema, msg.InsertRequest); err != nil {
 				mlog.Error(context.TODO(), "failed to fill embedding data", mlog.Err(err))
 				panic(err)
 			}
@@ -240,10 +190,9 @@ func newWriteNode(
 		metacache:    config.metacache,
 		pkField:      pkField,
 
-		functionRunners:        make(map[int32][]function.FunctionRunner),
-		functionOutputFieldIDs: make(map[int32][]int64),
+		functionStore: function.NewFunctionRunnerLocalStore(),
 	}
-	if _, err := wNode.getEmbeddingOutputFieldIDs(collSchema); err != nil {
+	if _, err := wNode.functionStore.OutputFieldIDs(collSchema); err != nil {
 		return nil, err
 	}
 	return wNode, nil

internal/flushcommon/writebuffer/write_buffer_test.go

@@ -872,10 +872,7 @@ func TestPrepareInsertMaterializesLegacyBM25Output(t *testing.T) {
 		},
 	}
 
-	errCh := function.AllocFunctionRunners(1, "v1", collSchema)
-	if errCh != nil {
-		assert.NoError(t, <-errCh)
-	}
+	assert.NoError(t, function.AllocFunctionRunners(1, "v1", collSchema))
 	_, err := function.FillFunctionData(context.Background(), 1, collSchema, insertMsg.InsertRequest)
 	assert.NoError(t, err)
 	defer function.ReleaseFunctionRunners(1, "v1")

internal/querynodev2/delegator/delegator.go

@@ -1302,6 +1302,7 @@ func (sd *shardDelegator) UpdateSchema(ctx context.Context, schema *schemapb.Col
 		newFunctionState.Close()
 		return err
 	}
+	sd.updateFunctionRunners(schema)
 	if idfOracle == nil && len(newSet) > 0 {
 		// StreamingNode schema changes flush and fence old growings before UpdateSchema and new-schema inserts.
 		// Old growings cannot receive stats for newly added BM25 fields; ProcessInsert registers only new growings.
@@ -1361,6 +1362,7 @@ func (sd *shardDelegator) Close() {
 	}
 
 	sd.functionState.Close()
+	sd.releaseFunctionRunners()
 
 	// clean up l0 segment in delete buffer
 	start := time.Now()
@@ -1374,6 +1376,39 @@ func (sd *shardDelegator) Close() {
 	}
 }
 
+func (sd *shardDelegator) allocFunctionRunners(schema *schemapb.CollectionSchema) {
+	if err := function.AllocFunctionRunners(sd.collectionID, delegatorFunctionRunnerKey(sd.vchannelName), schema); err != nil {
+		sd.warnFunctionRunnerInit(err, "allocate", schema)
+	}
+}
+
+func (sd *shardDelegator) updateFunctionRunners(schema *schemapb.CollectionSchema) {
+	if err := function.UpdateFunctionRunners(sd.collectionID, delegatorFunctionRunnerKey(sd.vchannelName), schema); err != nil {
+		sd.warnFunctionRunnerInit(err, "update", schema)
+	}
+}
+
+func (sd *shardDelegator) warnFunctionRunnerInit(err error, operation string, schema *schemapb.CollectionSchema) {
+	schemaVersion := function.LatestFunctionRunnerVersion
+	if schema != nil {
+		schemaVersion = schema.GetVersion()
+	}
+	mlog.Warn(context.TODO(), "failed to initialize delegator function runners",
+		mlog.Int64("collectionID", sd.collectionID),
+		mlog.String("vchannel", sd.vchannelName),
+		mlog.String("operation", operation),
+		mlog.Int32("schemaVersion", schemaVersion),
+		mlog.Err(err))
+}
+
+func (sd *shardDelegator) releaseFunctionRunners() {
+	function.ReleaseFunctionRunners(sd.collectionID, delegatorFunctionRunnerKey(sd.vchannelName))
+}
+
+func delegatorFunctionRunnerKey(vchannel string) string {
+	return "DELEGATOR-" + vchannel
+}
+
 // As partition stats is an optimization for search/query which is not mandatory for milvus instance,
 // loading partitionStats will be a try-best process and will skip+logError when running across errors rather than
 // return an error status
@@ -1485,6 +1520,7 @@ func NewShardDelegator(ctx context.Context, collectionID UniqueID, replicaID Uni
 		return nil, err
 	}
 	sd.functionState = functionState
+	sd.allocFunctionRunners(collection.Schema())
 
 	hasBM25Field := lo.ContainsBy(collection.Schema().GetFunctions(), func(tf *schemapb.FunctionSchema) bool {
 		return tf.GetType() == schemapb.FunctionType_BM25

internal/querynodev2/delegator/delegator_data_test.go

@@ -301,15 +301,12 @@ func (s *DelegatorDataSuite) enableGrowingSourceFlush() {
 }
 
 func (s *DelegatorDataSuite) TearDownTest() {
-	function.ReleaseFunctionRunners(s.collectionID, s.vchannelName)
+	function.ReleaseFunctionRunners(s.collectionID, "WAL-"+s.vchannelName)
+	function.ReleaseFunctionRunners(s.collectionID, delegatorFunctionRunnerKey(s.vchannelName))
 }
 
 func (s *DelegatorDataSuite) allocFunctionRunnersForTest() {
-	function.ReleaseFunctionRunners(s.collectionID, s.vchannelName)
-	errCh := function.AllocFunctionRunners(s.collectionID, s.vchannelName, s.delegator.collection.Schema())
-	if errCh != nil {
-		s.Require().NoError(<-errCh)
-	}
+	s.Require().NoError(function.UpdateFunctionRunners(s.collectionID, delegatorFunctionRunnerKey(s.vchannelName), s.delegator.collection.Schema()))
 }
 
 func (s *DelegatorDataSuite) TestProcessInsert() {

internal/querynodev2/delegator/delegator_test.go

@@ -189,8 +189,10 @@ func (s *DelegatorSuite) SetupTest() {
 }
 
 func (s *DelegatorSuite) TearDownTest() {
-	function.ReleaseFunctionRunners(s.collectionID, s.vchannelName)
-	s.delegator.Close()
+	if s.delegator != nil {
+		s.delegator.Close()
+	}
+	function.ReleaseFunctionRunners(s.collectionID, "WAL-"+s.vchannelName)
 	s.delegator = nil
 }
 
@@ -259,8 +261,9 @@ func (s *DelegatorSuite) TestCreateDelegatorWithFunction() {
 			}},
 		}, nil, &querypb.LoadMetaInfo{SchemaBarrierTs: tsoutil.ComposeTSByTime(time.Now(), 0)})
 
-		_, err := NewShardDelegator(context.Background(), s.collectionID, s.replicaID, s.vchannelName, s.version, s.workerManager, manager, s.loader, 10000, nil, s.chunkManager, NewChannelQueryView(nil, nil, nil, initialTargetVersion), nil)
+		delegator, err := NewShardDelegator(context.Background(), s.collectionID, s.replicaID, s.vchannelName, s.version, s.workerManager, manager, s.loader, 10000, nil, s.chunkManager, NewChannelQueryView(nil, nil, nil, initialTargetVersion), nil)
 		s.NoError(err)
+		defer delegator.Close()
 	})
 }
 
@@ -1506,12 +1509,8 @@ func (s *DelegatorSuite) ResetDelegator() {
 }
 
 func (s *DelegatorSuite) allocFunctionRunnersForTest() {
-	function.ReleaseFunctionRunners(s.collectionID, s.vchannelName)
 	sd := s.delegator.(*shardDelegator)
-	errCh := function.AllocFunctionRunners(s.collectionID, s.vchannelName, sd.collection.Schema())
-	if errCh != nil {
-		s.Require().NoError(<-errCh)
-	}
+	s.Require().NoError(function.UpdateFunctionRunners(s.collectionID, delegatorFunctionRunnerKey(s.vchannelName), sd.collection.Schema()))
 }
 
 func (s *DelegatorSuite) nextSchemaBarrierLoadMeta() *querypb.LoadMetaInfo {
@@ -2884,10 +2883,8 @@ func (s *DelegatorSuite) TestDelegatorSearchWithMinHashFunction() {
 		s.Require().NoError(err)
 		defer delegator.Close()
 
-		function.ReleaseFunctionRunners(s.collectionID, s.vchannelName)
-		errCh := function.AllocFunctionRunners(s.collectionID, s.vchannelName, schema1)
-		s.Require().NotNil(errCh)
-		s.NoError(<-errCh)
+		function.ReleaseFunctionRunners(s.collectionID, "WAL-"+s.vchannelName)
+		s.NoError(function.AllocFunctionRunners(s.collectionID, "WAL-"+s.vchannelName, schema1))
 
 		changed, err := function.FillFunctionData(context.Background(), s.collectionID, schema1, &msgpb.InsertRequest{
 			FieldsData: []*schemapb.FieldData{{
@@ -2912,7 +2909,8 @@ func (s *DelegatorSuite) TestDelegatorSearchWithMinHashFunction() {
 		manager := segments.NewManager()
 		manager.Collection.PutOrRef(s.collectionID, schema1, nil, &querypb.LoadMetaInfo{SchemaBarrierTs: tsoutil.ComposeTSByTime(time.Now(), 0)})
 
-		_, err := NewShardDelegator(context.Background(), s.collectionID, s.replicaID, s.vchannelName, s.version, s.workerManager, manager, s.loader, 10000, nil, s.chunkManager, NewChannelQueryView(nil, nil, nil, initialTargetVersion), nil)
+		delegator, err := NewShardDelegator(context.Background(), s.collectionID, s.replicaID, s.vchannelName, s.version, s.workerManager, manager, s.loader, 10000, nil, s.chunkManager, NewChannelQueryView(nil, nil, nil, initialTargetVersion), nil)
 		s.NoError(err)
+		defer delegator.Close()
 	})
 }

internal/querynodev2/pipeline/insert_node.go

@@ -42,8 +42,7 @@ type insertNode struct {
 	manager      *DataManager
 	delegator    delegator.ShardDelegator
 
-	functionRunners        map[int32][]function.FunctionRunner
-	functionOutputFieldIDs map[int32][]int64
+	functionStore *function.FunctionRunnerLocalStore
 }
 
 func (iNode *insertNode) addInsertData(insertDatas map[UniqueID]*delegator.InsertData, msg *InsertMsg, collection *Collection) {
@@ -95,56 +94,8 @@ func (iNode *insertNode) addInsertData(insertDatas map[UniqueID]*delegator.Inser
 		mlog.Uint64("timestampMax", msg.EndTimestamp))
 }
 
-// fillEmbeddingData is only used to handle old insert messages that were not embedded before WAL append.
-func (iNode *insertNode) fillEmbeddingData(schema *schemapb.CollectionSchema, msg *InsertMsg) error {
-	if !function.HasEmbeddingFunctions(schema) {
-		return nil
-	}
-	schemaVersion := schema.GetVersion()
-	_, ok, err := function.TryMaterialize(iNode.collectionID, schemaVersion, msg.InsertRequest)
-	if err != nil {
-		return err
-	}
-	if ok {
-		return nil
-	}
-
-	runners, ok := iNode.functionRunners[schemaVersion]
-	if !ok {
-		runners, err = function.BuildEmbeddingRunners(schema)
-		if err != nil {
-			return err
-		}
-		iNode.functionRunners[schemaVersion] = runners
-	}
-	_, err = function.FillFunctionFields(runners, msg.InsertRequest)
-	return err
-}
-
-func (iNode *insertNode) getEmbeddingOutputFieldIDs(schema *schemapb.CollectionSchema) ([]int64, error) {
-	schemaVersion := schema.GetVersion()
-	if outputFieldIDs, ok := iNode.functionOutputFieldIDs[schemaVersion]; ok {
-		return outputFieldIDs, nil
-	}
-
-	if !function.HasEmbeddingFunctions(schema) {
-		iNode.functionOutputFieldIDs[schemaVersion] = nil
-		return nil, nil
-	}
-	outputFieldIDs, err := function.EmbeddingOutputFieldIDs(schema)
-	if err != nil {
-		return nil, err
-	}
-	iNode.functionOutputFieldIDs[schemaVersion] = outputFieldIDs
-	return outputFieldIDs, nil
-}
-
 func (iNode *insertNode) Close() {
-	for _, runners := range iNode.functionRunners {
-		function.CloseRunners(runners)
-	}
-	iNode.functionRunners = make(map[int32][]function.FunctionRunner)
-	iNode.functionOutputFieldIDs = make(map[int32][]int64)
+	iNode.functionStore.Close()
 }
 
 // Insert task
@@ -163,7 +114,7 @@ func (iNode *insertNode) Operate(in Msg) Msg {
 			panic("insertNode with collection not exist")
 		}
 		schema := collection.Schema()
-		functionOutputFieldIDs, err := iNode.getEmbeddingOutputFieldIDs(schema)
+		functionOutputFieldIDs, err := iNode.functionStore.OutputFieldIDs(schema)
 		if err != nil {
 			mlog.Error(context.TODO(), "failed to get embedding output fields", mlog.String("channel", iNode.channel), mlog.Err(err))
 			panic(err)
@@ -172,7 +123,7 @@ func (iNode *insertNode) Operate(in Msg) Msg {
 		insertDatas := make(map[UniqueID]*delegator.InsertData)
 		for _, msg := range nodeMsg.insertMsgs {
 			if len(functionOutputFieldIDs) > 0 && !function.HasAllFieldDataByID(msg.GetFieldsData(), functionOutputFieldIDs) {
-				if err := iNode.fillEmbeddingData(schema, msg); err != nil {
+				if err := iNode.functionStore.FillEmbeddingData(iNode.collectionID, schema, msg.InsertRequest); err != nil {
 					mlog.Error(context.TODO(), "failed to fill embedding data for insert message", mlog.String("channel", iNode.channel), mlog.Err(err))
 					panic(err)
 				}
@@ -201,15 +152,14 @@ func newInsertNode(
 	maxQueueLength int32,
 ) (*insertNode, error) {
 	iNode := &insertNode{
-		BaseNode:               base.NewBaseNode(fmt.Sprintf("InsertNode-%s", channel), maxQueueLength),
-		collectionID:           collectionID,
-		channel:                channel,
-		manager:                manager,
-		delegator:              delegator,
-		functionRunners:        make(map[int32][]function.FunctionRunner),
-		functionOutputFieldIDs: make(map[int32][]int64),
+		BaseNode:      base.NewBaseNode(fmt.Sprintf("InsertNode-%s", channel), maxQueueLength),
+		collectionID:  collectionID,
+		channel:       channel,
+		manager:       manager,
+		delegator:     delegator,
+		functionStore: function.NewFunctionRunnerLocalStore(),
 	}
-	if _, err := iNode.getEmbeddingOutputFieldIDs(schema); err != nil {
+	if _, err := iNode.functionStore.OutputFieldIDs(schema); err != nil {
 		return nil, err
 	}
 	return iNode, nil

internal/streamingnode/server/wal/interceptors/shard/function_materializer.go

@@ -12,51 +12,40 @@ import (
 	"github.com/milvus-io/milvus/pkg/v3/streaming/util/message"
 )
 
+func walFunctionRunnerKey(vchannel string) string {
+	return "WAL-" + vchannel
+}
+
 func (impl *shardInterceptor) allocFunctionRunners(collectionID int64, vchannel string, schema *schemapb.CollectionSchema) {
-	if !function.HasEmbeddingFunctions(schema) {
-		return
-	}
-	errCh := function.AllocFunctionRunners(collectionID, vchannel, schema)
-	if errCh == nil {
-		return
-	}
-	schemaVersion := int32(0)
+	key := walFunctionRunnerKey(vchannel)
+	schemaVersion := function.LatestFunctionRunnerVersion
 	if schema != nil {
 		schemaVersion = schema.GetVersion()
 	}
-	go func() {
-		if err := <-errCh; err != nil {
-			impl.shardManager.Logger().Warn(context.TODO(), "failed to allocate function runners",
-				mlog.Int64("collectionID", collectionID),
-				mlog.String("vchannel", vchannel),
-				mlog.Int32("schemaVersion", schemaVersion),
-				mlog.Err(err))
-		}
-	}()
+	if err := function.AllocFunctionRunners(collectionID, key, schema); err != nil {
+		impl.shardManager.Logger().Warn(context.TODO(), "failed to allocate function runners",
+			mlog.Int64("collectionID", collectionID),
+			mlog.String("vchannel", vchannel),
+			mlog.String("key", key),
+			mlog.Int32("schemaVersion", schemaVersion),
+			mlog.Err(err))
+	}
 }
 
 func (impl *shardInterceptor) updateFunctionRunners(collectionID int64, vchannel string, schema *schemapb.CollectionSchema) {
-	if !function.HasEmbeddingFunctions(schema) {
-		function.ReleaseFunctionRunners(collectionID, vchannel)
-		return
-	}
-	errCh := function.UpdateFunctionRunners(collectionID, vchannel, schema)
-	if errCh == nil {
-		return
-	}
-	schemaVersion := int32(0)
+	key := walFunctionRunnerKey(vchannel)
+	schemaVersion := function.LatestFunctionRunnerVersion
 	if schema != nil {
 		schemaVersion = schema.GetVersion()
 	}
-	go func() {
-		if err := <-errCh; err != nil {
-			impl.shardManager.Logger().Warn(context.TODO(), "failed to update function runners",
-				mlog.Int64("collectionID", collectionID),
-				mlog.String("vchannel", vchannel),
-				mlog.Int32("schemaVersion", schemaVersion),
-				mlog.Err(err))
-		}
-	}()
+	if err := function.UpdateFunctionRunners(collectionID, key, schema); err != nil {
+		impl.shardManager.Logger().Warn(context.TODO(), "failed to update function runners",
+			mlog.Int64("collectionID", collectionID),
+			mlog.String("vchannel", vchannel),
+			mlog.String("key", key),
+			mlog.Int32("schemaVersion", schemaVersion),
+			mlog.Err(err))
+	}
 }
 
 type collectionSchemaProvider interface {

internal/streamingnode/server/wal/interceptors/shard/shard_interceptor.go

@@ -102,7 +102,7 @@ func (impl *shardInterceptor) handleDropCollection(ctx context.Context, msg mess
 		return msgID, err
 	}
 	impl.shardManager.DropCollection(message.MustAsImmutableDropCollectionMessageV1(msg.IntoImmutableMessage(msgID)))
-	function.ReleaseFunctionRunners(dropCollectionMessage.Header().GetCollectionId(), dropCollectionMessage.VChannel())
+	function.ReleaseFunctionRunners(dropCollectionMessage.Header().GetCollectionId(), walFunctionRunnerKey(dropCollectionMessage.VChannel()))
 	return msgID, nil
 }
 
@@ -151,7 +151,8 @@ func (impl *shardInterceptor) handleInsertMessage(ctx context.Context, msg messa
 	header := insertMsg.Header()
 	collectionID := header.GetCollectionId()
 	schemaVersion := header.GetSchemaVersion()
-	if correctSchemaVersion, err := impl.shardManager.CheckIfCollectionSchemaVersionMatch(header); err != nil {
+	correctSchemaVersion, err := impl.shardManager.CheckIfCollectionSchemaVersionMatch(header)
+	if err != nil {
 		if errors.Is(err, shards.ErrCollectionNotFound) {
 			return nil, status.NewUnrecoverableError("collection %d not found", collectionID)
 		}
@@ -175,6 +176,7 @@ func (impl *shardInterceptor) handleInsertMessage(ctx context.Context, msg messa
 			mlog.Err(err))
 		return nil, errors.Wrap(err, "CheckIfCollectionSchemaVersionMatch")
 	}
+	schemaVersion = correctSchemaVersion
 	if err := impl.materializeFunctionFields(ctx, insertMsg, header.GetCollectionId(), schemaVersion); err != nil {
 		impl.shardManager.Logger().Warn(ctx, "failed to materialize function fields before WAL append",
 			mlog.Int64("collectionID", header.GetCollectionId()),
@@ -375,7 +377,7 @@ func (impl *shardInterceptor) handleTruncateCollectionMessage(ctx context.Contex
 func (impl *shardInterceptor) Close() {
 	if schemaProvider, ok := impl.shardManager.(collectionSchemaProvider); ok {
 		for collectionID, schemaInfo := range schemaProvider.GetAllCollectionSchemaInfos() {
-			function.ReleaseFunctionRunners(collectionID, schemaInfo.VChannel)
+			function.ReleaseFunctionRunners(collectionID, walFunctionRunnerKey(schemaInfo.VChannel))
 		}
 	}
 }

internal/streamingnode/server/wal/interceptors/shard/shard_interceptor_test.go

@@ -127,11 +127,8 @@ func TestShardInterceptorUpdateFunctionRunnersReleasesWhenFunctionsDropped(t *te
 			},
 		},
 	}
-	errCh := function.AllocFunctionRunners(collectionID, vchannel, schema)
-	if errCh != nil {
-		assert.NoError(t, <-errCh)
-	}
-	defer function.ReleaseFunctionRunners(collectionID, vchannel)
+	assert.NoError(t, function.AllocFunctionRunners(collectionID, walFunctionRunnerKey(vchannel), schema))
+	defer function.ReleaseFunctionRunners(collectionID, walFunctionRunnerKey(vchannel))
 
 	ok, err := function.RunWithAnalyzer(context.Background(), collectionID, schema.GetVersion(), 101, func(function.Analyzer) error {
 		return nil

internal/streamingnode/server/wal/interceptors/shard/shards/shard_manager.go

@@ -22,6 +22,10 @@ import (
 	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
 )
 
+// latestCollectionSchemaVersion asks GetCollectionSchema to use the latest
+// schema snapshot. Zero is a valid schema version and must stay explicit.
+const latestCollectionSchemaVersion int32 = -1
+
 var (
 	ErrCollectionExists                = errors.New("collection exists")
 	ErrCollectionNotFound              = errors.New("collection not found")

internal/streamingnode/server/wal/interceptors/shard/shards/shard_manager_collection.go

@@ -208,9 +208,10 @@ func (m *shardManagerImpl) checkIfCollectionSchemaVersionMatch(header *message.I
 		m.Logger().Warn(context.TODO(), "collection not found", mlog.Int64("collectionID", collectionID))
 		return -1, ErrCollectionNotFound
 	}
-	// Input schemaVersion 0 means the proxy did not set it (old proxy or old SDK).
-	// Skip the schema presence and version checks for backward compatibility during rolling
-	// upgrades, where a legacy collection may still have Schema == nil when an old proxy writes.
+	// Missing schemaVersion means the proxy did not set it (old proxy or old SDK).
+	// Skip the schema presence and version checks for backward compatibility during
+	// rolling upgrades, where a legacy collection may still have Schema == nil when
+	// an old proxy writes.
 	if header.SchemaVersion == nil {
 		return collectionInfo.SchemaVersion(), nil
 	}
@@ -243,7 +244,7 @@ func (m *shardManagerImpl) GetCollectionSchema(collectionID int64, schemaVersion
 		return nil, ErrCollectionSchemaNotFound
 	}
 	collectionSchemaVersion := collectionInfo.SchemaVersion()
-	if schemaVersion != 0 && collectionSchemaVersion != schemaVersion {
+	if schemaVersion != latestCollectionSchemaVersion && collectionSchemaVersion != schemaVersion {
 		return nil, ErrCollectionSchemaVersionNotMatch
 	}
 

internal/streamingnode/server/wal/interceptors/shard/shards/shard_manager_test.go

@@ -446,6 +446,11 @@ func TestShardManagerSchemaVersionCheck(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, int32(1), ver)
 
+	_, err = m.GetCollectionSchema(100, latestCollectionSchemaVersion)
+	assert.NoError(t, err)
+	_, err = m.GetCollectionSchema(100, 0)
+	assert.ErrorIs(t, err, ErrCollectionSchemaVersionNotMatch)
+
 	// version mismatch should fail
 	ver, err = m.CheckIfCollectionSchemaVersionMatch(&message.InsertMessageHeader{
 		CollectionId:  100,
@@ -544,6 +549,8 @@ func TestShardManagerSchemaVersionCheck(t *testing.T) {
 	})
 	assert.NoError(t, err)
 	assert.Equal(t, int32(0), ver)
+	_, err = m.GetCollectionSchema(103, 0)
+	assert.NoError(t, err)
 
 	// Test 4: AlterCollection updates schema version
 	updatedSchema := &schemapb.CollectionSchema{

internal/util/function/local_store.go

@@ -0,0 +1,104 @@
+// Licensed to the LF AI & Data foundation under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package function
+
+import (
+	"github.com/milvus-io/milvus-proto/go-api/v3/msgpb"
+	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
+)
+
+// FunctionRunnerLocalStore owns short-lived runners used when a managed runner
+// is unavailable. It is intended for consumer-side old insert message handling.
+type FunctionRunnerLocalStore struct {
+	runners        map[int32][]FunctionRunner
+	outputFieldIDs map[int32][]int64
+}
+
+func NewFunctionRunnerLocalStore() *FunctionRunnerLocalStore {
+	return &FunctionRunnerLocalStore{
+		runners:        make(map[int32][]FunctionRunner),
+		outputFieldIDs: make(map[int32][]int64),
+	}
+}
+
+func (s *FunctionRunnerLocalStore) Close() {
+	if s == nil {
+		return
+	}
+	for _, runners := range s.runners {
+		CloseRunners(runners)
+	}
+	s.runners = make(map[int32][]FunctionRunner)
+	s.outputFieldIDs = make(map[int32][]int64)
+}
+
+func (s *FunctionRunnerLocalStore) OutputFieldIDs(schema *schemapb.CollectionSchema) ([]int64, error) {
+	if s == nil {
+		return EmbeddingOutputFieldIDs(schema)
+	}
+	schemaVersion := schema.GetVersion()
+	if outputFieldIDs, ok := s.outputFieldIDs[schemaVersion]; ok {
+		return outputFieldIDs, nil
+	}
+
+	if !HasEmbeddingFunctions(schema) {
+		s.outputFieldIDs[schemaVersion] = nil
+		return nil, nil
+	}
+	outputFieldIDs, err := EmbeddingOutputFieldIDs(schema)
+	if err != nil {
+		return nil, err
+	}
+	s.outputFieldIDs[schemaVersion] = outputFieldIDs
+	return outputFieldIDs, nil
+}
+
+// FillEmbeddingData is only used to handle old insert messages that were not
+// embedded before WAL append.
+func (s *FunctionRunnerLocalStore) FillEmbeddingData(collectionID int64, schema *schemapb.CollectionSchema, body *msgpb.InsertRequest) error {
+	if !HasEmbeddingFunctions(schema) {
+		return nil
+	}
+	if s == nil {
+		runners, err := BuildEmbeddingRunners(schema)
+		if err != nil {
+			return err
+		}
+		defer CloseRunners(runners)
+		_, err = FillFunctionFields(runners, body)
+		return err
+	}
+	schemaVersion := schema.GetVersion()
+	_, ok, err := TryMaterialize(collectionID, schemaVersion, body)
+	if err != nil {
+		return err
+	}
+	if ok {
+		return nil
+	}
+
+	runners, ok := s.runners[schemaVersion]
+	if !ok {
+		runners, err = BuildEmbeddingRunners(schema)
+		if err != nil {
+			return err
+		}
+		s.runners[schemaVersion] = runners
+	}
+	_, err = FillFunctionFields(runners, body)
+	return err
+}

internal/util/function/manager.go

@@ -30,6 +30,11 @@ var (
 
 var errFunctionRunnerEntryRemoved = errors.New("function runner manager entry was removed")
 
+// LatestFunctionRunnerVersion tells the manager to use the latest version
+// snapshot already allocated for the collection. It must not be zero because
+// schema version 0 is valid for compatibility paths.
+const LatestFunctionRunnerVersion int32 = -1
+
 type functionRunnerState int
 
 const (
@@ -39,39 +44,49 @@ const (
 )
 
 type FunctionRunnerManager interface {
-	// Alloc records that a vchannel of the collection is using the schema version
+	// Alloc records that a lifecycle key is using the schema version
 	// and asynchronously tries to initialize function runners for that version.
+	// The key identifies an independent lifecycle scope; for example, WAL uses
+	// "WAL-"+vchannel while delegator uses "DELEGATOR-"+vchannel.
 	// Initialization failures are logged and retried by later Alloc, Update, or
 	// Materialize calls instead of failing collection recovery.
-	Alloc(collectionID int64, vchannel string, schema *schemapb.CollectionSchema) <-chan error
+	Alloc(collectionID int64, key string, schema *schemapb.CollectionSchema) error
 
-	// Update moves a vchannel to a newer schema version and asynchronously
-	// initializes any missing function runners required by that version.
-	// Runners for older versions are kept until no vchannel references them.
-	Update(collectionID int64, vchannel string, schema *schemapb.CollectionSchema) <-chan error
+	// Update moves a lifecycle key to a newer schema version and asynchronously
+	// initializes any missing function runners required by that version. If the
+	// schema no longer has runner-backed functions, Update releases the key.
+	// Runners for older versions are kept until no key uses those versions.
+	Update(collectionID int64, key string, schema *schemapb.CollectionSchema) error
 
-	// Release removes one vchannel reference from a collection. The collection
-	// entry and its runners are closed only after all vchannels are released.
-	Release(collectionID int64, vchannel string)
+	// Release removes one lifecycle key. The collection entry and its runners
+	// are closed only after all keys are released.
+	Release(collectionID int64, key string)
 
 	// Materialize fills missing function output fields for an insert request.
-	// This is the write-before path and will initialize missing runners on demand;
-	// initialization or execution failures are returned to the caller.
+	// Passing a nil schema uses the latest version snapshot already allocated by
+	// Create, recovery, or update; any foreground runner initialization uses the
+	// schema saved in that entry. Passing a non-nil schema can initialize the
+	// matching version. Initialization or execution failures are returned to the
+	// caller.
 	Materialize(ctx context.Context, collectionID int64, schema *schemapb.CollectionSchema, body *msgpb.InsertRequest) (bool, error)
 
 	// TryMaterialize is used by compatibility paths for old insert messages. It
-	// only uses already cached runners for the given schema version and returns
-	// ok=false when the caller should build short-lived runners instead.
+	// only uses already cached runners for the given schema version, or the
+	// latest allocated version when schemaVersion is LatestFunctionRunnerVersion.
+	// It returns ok=false when the caller should build short-lived runners instead.
 	TryMaterialize(collectionID int64, schemaVersion int32, body *msgpb.InsertRequest) (bool, bool, error)
 
 	// RunWithRunner runs the callback with the runner that owns the output field.
-	// The callback is executed synchronously while the manager protects the runner
-	// from concurrent close; callers must not retain the runner after the callback.
+	// schemaVersion can be LatestFunctionRunnerVersion to use the latest allocated
+	// version. The callback is executed synchronously while the manager protects
+	// the runner from concurrent close; callers must not retain the runner after
+	// the callback.
 	RunWithRunner(ctx context.Context, collectionID int64, schemaVersion int32, outputFieldID int64, run func(schemapb.FunctionType, FunctionRunner) error) (bool, error)
 
 	// RunWithAnalyzer runs the callback with the analyzer service associated with
-	// a function input field. BM25 runners can serve analyzer requests. The
-	// callback is protected from concurrent close and must not retain the analyzer.
+	// a function input field. schemaVersion can be LatestFunctionRunnerVersion to
+	// use the latest allocated version. BM25 runners can serve analyzer requests.
+	// The callback is protected from concurrent close and must not retain the analyzer.
 	RunWithAnalyzer(ctx context.Context, collectionID int64, schemaVersion int32, fieldID int64, run func(Analyzer) error) (bool, error)
 
 	// Close releases all cached runners managed by this manager.
@@ -83,12 +98,18 @@ type functionRunnerManager struct {
 	entries map[int64]*functionRunnerCollectionEntry
 }
 
+// Lock order is functionRunnerManager.mu -> functionRunnerCollectionEntry.mu ->
+// functionRunnerEntry.mu. The manager lock serializes get-or-create with
+// zero-key removal so a newly allocated lifecycle key cannot reuse an entry that
+// is already being removed. Runner Close calls are always done after releasing
+// the manager/collection locks, so slow runner teardown does not block collection
+// map operations.
 type functionRunnerCollectionEntry struct {
-	mu               sync.RWMutex
-	collectionID     int64
-	vchannelVersions map[string]int32
-	versionRunners   map[int32]*functionRunnerVersion
-	runners          map[string]*functionRunnerEntry
+	mu             sync.RWMutex
+	collectionID   int64
+	keyVersions    map[string]int32
+	versionRunners map[int32]*functionRunnerVersion
+	runners        map[string]*functionRunnerEntry
 }
 
 type functionRunnerVersion struct {
@@ -113,23 +134,27 @@ type functionRunnerEntry struct {
 
 func newFunctionRunnerCollectionEntry(collectionID int64) *functionRunnerCollectionEntry {
 	return &functionRunnerCollectionEntry{
-		collectionID:     collectionID,
-		vchannelVersions: make(map[string]int32),
-		versionRunners:   make(map[int32]*functionRunnerVersion),
-		runners:          make(map[string]*functionRunnerEntry),
+		collectionID:   collectionID,
+		keyVersions:    make(map[string]int32),
+		versionRunners: make(map[int32]*functionRunnerVersion),
+		runners:        make(map[string]*functionRunnerEntry),
 	}
 }
 
-func (e *functionRunnerCollectionEntry) Alloc(vchannel string, schema *schemapb.CollectionSchema) <-chan error {
-	return e.allocOrUpdate(vchannel, schema, "initialize")
+func (e *functionRunnerCollectionEntry) Alloc(key string, schema *schemapb.CollectionSchema) error {
+	return e.allocOrUpdate(key, schema, "initialize")
 }
 
-func (e *functionRunnerCollectionEntry) Update(vchannel string, schema *schemapb.CollectionSchema) <-chan error {
-	return e.allocOrUpdate(vchannel, schema, "update")
+func (e *functionRunnerCollectionEntry) Update(key string, schema *schemapb.CollectionSchema) error {
+	return e.allocOrUpdate(key, schema, "update")
 }
 
-func (e *functionRunnerCollectionEntry) allocOrUpdate(vchannel string, schema *schemapb.CollectionSchema, operation string) <-chan error {
-	schemaVersion := int32(0)
+func (e *functionRunnerCollectionEntry) allocOrUpdate(
+	key string,
+	schema *schemapb.CollectionSchema,
+	operation string,
+) error {
+	schemaVersion := LatestFunctionRunnerVersion
 	if schema != nil {
 		schemaVersion = schema.GetVersion()
 	}
@@ -137,19 +162,17 @@ func (e *functionRunnerCollectionEntry) allocOrUpdate(vchannel string, schema *s
 		mlog.Warn(context.TODO(), "failed to initialize function runners, will retry on next request",
 			mlog.String("operation", operation),
 			mlog.Int64("collectionID", e.collectionID),
-			mlog.String("vchannel", vchannel),
+			mlog.String("key", key),
 			mlog.Int32("schemaVersion", schemaVersion),
 			mlog.Err(err))
 	}
-	runnerEntries, err := e.ensureVersion(vchannel, schema)
+	runnerEntries, err := e.ensureVersion(key, schema)
 	if err != nil {
-		warnInitFailure(err)
-		return nil
+		return err
 	}
 	if len(runnerEntries) == 0 {
 		return nil
 	}
-	errCh := make(chan error, 1)
 	go func() {
 		for _, runnerEntry := range runnerEntries {
 			if err := runnerEntry.ensureReady(context.Background()); err != nil {
@@ -159,25 +182,24 @@ func (e *functionRunnerCollectionEntry) allocOrUpdate(vchannel string, schema *s
 				break
 			}
 		}
-		errCh <- nil
 	}()
-	return errCh
+	return nil
 }
 
-func (e *functionRunnerCollectionEntry) Release(vchannel string) bool {
+func (e *functionRunnerCollectionEntry) Release(key string) bool {
 	e.mu.Lock()
 	defer e.mu.Unlock()
 
-	if _, ok := e.vchannelVersions[vchannel]; !ok {
+	if _, ok := e.keyVersions[key]; !ok {
 		return false
 	}
-	delete(e.vchannelVersions, vchannel)
-	return len(e.vchannelVersions) == 0
+	delete(e.keyVersions, key)
+	return len(e.keyVersions) == 0
 }
 
 func (e *functionRunnerCollectionEntry) Close() {
 	e.mu.Lock()
-	e.vchannelVersions = make(map[string]int32)
+	e.keyVersions = make(map[string]int32)
 	runnerEntries := e.gcLocked()
 	e.mu.Unlock()
 
@@ -193,27 +215,23 @@ func (e *functionRunnerCollectionEntry) GC() {
 }
 
 func (e *functionRunnerCollectionEntry) gcLocked() []*functionRunnerEntry {
-	if len(e.vchannelVersions) == 0 {
+	if len(e.keyVersions) == 0 {
 		runnerEntries := make([]*functionRunnerEntry, 0, len(e.runners))
 		for _, runnerEntry := range e.runners {
 			runnerEntries = append(runnerEntries, runnerEntry)
 		}
-		e.vchannelVersions = make(map[string]int32)
+		e.keyVersions = make(map[string]int32)
 		e.versionRunners = make(map[int32]*functionRunnerVersion)
 		e.runners = make(map[string]*functionRunnerEntry)
 		return runnerEntries
 	}
 
-	var minVersion int32
-	first := true
-	for _, version := range e.vchannelVersions {
-		if first || version < minVersion {
-			minVersion = version
-			first = false
-		}
+	activeVersions := make(map[int32]struct{}, len(e.keyVersions))
+	for _, version := range e.keyVersions {
+		activeVersions[version] = struct{}{}
 	}
 	for version := range e.versionRunners {
-		if version < minVersion {
+		if _, ok := activeVersions[version]; !ok {
 			delete(e.versionRunners, version)
 		}
 	}
@@ -236,15 +254,35 @@ func (e *functionRunnerCollectionEntry) gcLocked() []*functionRunnerEntry {
 	return runnerEntries
 }
 
-func (e *functionRunnerCollectionEntry) getVersionRunnerEntries(schemaVersion int32) ([]*functionRunnerEntry, []int64, bool) {
-	e.mu.RLock()
-	defer e.mu.RUnlock()
+func useLatestFunctionRunnerVersion(schemaVersion int32) bool {
+	return schemaVersion == LatestFunctionRunnerVersion
+}
 
-	versionRunners, ok := e.versionRunners[schemaVersion]
+func (e *functionRunnerCollectionEntry) getVersionRunnerLocked(schemaVersion int32) (*functionRunnerVersion, bool) {
+	if !useLatestFunctionRunnerVersion(schemaVersion) {
+		versionRunners, ok := e.versionRunners[schemaVersion]
+		return versionRunners, ok
+	}
+
+	if len(e.versionRunners) == 0 {
+		return nil, false
+	}
+	var latestVersion int32
+	first := true
+	for version := range e.versionRunners {
+		if first || version > latestVersion {
+			latestVersion = version
+			first = false
+		}
+	}
+	return e.versionRunners[latestVersion], true
+}
+
+func (e *functionRunnerCollectionEntry) getVersionRunnerEntriesLocked(schemaVersion int32) ([]*functionRunnerEntry, []int64, bool) {
+	versionRunners, ok := e.getVersionRunnerLocked(schemaVersion)
 	if !ok {
 		return nil, nil, false
 	}
-
 	runnerEntries := make([]*functionRunnerEntry, 0, len(versionRunners.signatures))
 	for _, signature := range versionRunners.signatures {
 		runnerEntry := e.runners[signature]
@@ -256,6 +294,13 @@ func (e *functionRunnerCollectionEntry) getVersionRunnerEntries(schemaVersion in
 	return runnerEntries, append([]int64(nil), versionRunners.outputFieldIDs...), true
 }
 
+func (e *functionRunnerCollectionEntry) getVersionRunnerEntries(schemaVersion int32) ([]*functionRunnerEntry, []int64, bool) {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
+
+	return e.getVersionRunnerEntriesLocked(schemaVersion)
+}
+
 func runWithRunnerEntries(
 	ctx context.Context,
 	runnerEntries []*functionRunnerEntry,
@@ -273,6 +318,9 @@ func runWithRunnerEntries(
 		}
 	}
 
+	// Hold runner entry read locks while the callback runs to prevent concurrent
+	// close. The read locks are shared, so concurrent materialization is still
+	// allowed; concrete runners protect their own mutable state.
 	runners := make([]FunctionRunner, 0, len(runnerEntries))
 	for _, runnerEntry := range runnerEntries {
 		runner, unlock, ok, err := runnerEntry.lockRunner(initRunner)
@@ -445,34 +493,43 @@ func newFunctionRunnerManager() *functionRunnerManager {
 	}
 }
 
-func (m *functionRunnerManager) Alloc(collectionID int64, vchannel string, schema *schemapb.CollectionSchema) <-chan error {
-	if vchannel == "" {
-		return functionRunnerErrorResult(errors.New("vchannel is empty"))
+func (m *functionRunnerManager) Alloc(
+	collectionID int64,
+	key string,
+	schema *schemapb.CollectionSchema,
+) error {
+	if key == "" {
+		return merr.WrapErrFunctionFailedMsg("function runner key is empty")
 	}
 	if schema == nil {
-		return functionRunnerErrorResult(errors.New("collection schema is nil"))
+		return merr.WrapErrFunctionFailedMsg("collection schema is nil")
 	}
 	if !HasEmbeddingFunctions(schema) {
 		return nil
 	}
-	return m.getOrCreateEntry(collectionID).Alloc(vchannel, schema)
+	return m.getOrCreateEntry(collectionID).Alloc(key, schema)
 }
 
-func (m *functionRunnerManager) Update(collectionID int64, vchannel string, schema *schemapb.CollectionSchema) <-chan error {
-	if vchannel == "" {
-		return functionRunnerErrorResult(errors.New("vchannel is empty"))
+func (m *functionRunnerManager) Update(
+	collectionID int64,
+	key string,
+	schema *schemapb.CollectionSchema,
+) error {
+	if key == "" {
+		return merr.WrapErrFunctionFailedMsg("function runner key is empty")
 	}
 	if schema == nil {
-		return functionRunnerErrorResult(errors.New("collection schema is nil"))
+		return merr.WrapErrFunctionFailedMsg("collection schema is nil")
 	}
 	if !HasEmbeddingFunctions(schema) {
+		m.Release(collectionID, key)
 		return nil
 	}
-	return m.getOrCreateEntry(collectionID).Update(vchannel, schema)
+	return m.getOrCreateEntry(collectionID).Update(key, schema)
 }
 
 func (e *functionRunnerCollectionEntry) ensureVersion(
-	vchannel string,
+	key string,
 	schema *schemapb.CollectionSchema,
 ) ([]*functionRunnerEntry, error) {
 	if schema == nil {
@@ -481,11 +538,7 @@ func (e *functionRunnerCollectionEntry) ensureVersion(
 
 	schemaVersion := schema.GetVersion()
 	e.mu.Lock()
-	if vchannel != "" {
-		if currentVersion, ok := e.vchannelVersions[vchannel]; !ok || currentVersion <= schemaVersion {
-			e.vchannelVersions[vchannel] = schemaVersion
-		}
-	}
+	e.updateKeyVersionLocked(key, schemaVersion)
 
 	versionRunners, ok := e.versionRunners[schemaVersion]
 	if ok {
@@ -513,7 +566,6 @@ func (e *functionRunnerCollectionEntry) ensureVersion(
 
 	functions := embeddingFunctions(schema)
 	functionsBySignature := make(map[string]*schemapb.FunctionSchema, len(functions))
-	outputFieldIDsBySignature := make(map[string][]int64, len(functions))
 	signatures := make([]string, 0, len(functions))
 	outputFieldIDs := make([]int64, 0, len(functions))
 	outputFieldSignatures := make(map[int64]string)
@@ -535,7 +587,6 @@ func (e *functionRunnerCollectionEntry) ensureVersion(
 		signatures = append(signatures, signature)
 		outputFieldIDs = append(outputFieldIDs, functionOutputFieldIDs...)
 		functionsBySignature[signature] = fn
-		outputFieldIDsBySignature[signature] = append([]int64(nil), functionOutputFieldIDs...)
 		for _, outputFieldID := range functionOutputFieldIDs {
 			outputFieldSignatures[outputFieldID] = signature
 			outputFieldTypes[outputFieldID] = fn.GetType()
@@ -548,11 +599,7 @@ func (e *functionRunnerCollectionEntry) ensureVersion(
 	}
 
 	e.mu.Lock()
-	if vchannel != "" {
-		if currentVersion, ok := e.vchannelVersions[vchannel]; !ok || currentVersion <= schemaVersion {
-			e.vchannelVersions[vchannel] = schemaVersion
-		}
-	}
+	e.updateKeyVersionLocked(key, schemaVersion)
 
 	versionRunners, ok = e.versionRunners[schemaVersion]
 	if !ok {
@@ -576,8 +623,7 @@ func (e *functionRunnerCollectionEntry) ensureVersion(
 		e.versionRunners[schemaVersion] = versionRunners
 	} else {
 		for _, signature := range versionRunners.signatures {
-			_, outputOK := outputFieldIDsBySignature[signature]
-			if _, functionOK := functionsBySignature[signature]; !functionOK || !outputOK {
+			if _, functionOK := functionsBySignature[signature]; !functionOK {
 				e.mu.Unlock()
 				return nil, merr.WrapErrFunctionFailedMsg("function runner signature %s not found in schema version %d", signature, schemaVersion)
 			}
@@ -607,11 +653,30 @@ func (e *functionRunnerCollectionEntry) ensureVersion(
 	return initRunnerEntries, nil
 }
 
+func (e *functionRunnerCollectionEntry) updateKeyVersionLocked(
+	key string,
+	schemaVersion int32,
+) {
+	if key == "" {
+		return
+	}
+
+	version, ok := e.keyVersions[key]
+	if !ok {
+		e.keyVersions[key] = schemaVersion
+		return
+	}
+
+	if version <= schemaVersion {
+		e.keyVersions[key] = schemaVersion
+	}
+}
+
 func (e *functionRunnerCollectionEntry) getRunnerEntryByOutputField(schemaVersion int32, outputFieldID int64) (*functionRunnerEntry, schemapb.FunctionType, bool) {
 	e.mu.RLock()
 	defer e.mu.RUnlock()
 
-	versionRunners, ok := e.versionRunners[schemaVersion]
+	versionRunners, ok := e.getVersionRunnerLocked(schemaVersion)
 	if !ok {
 		return nil, schemapb.FunctionType_Unknown, false
 	}
@@ -630,7 +695,7 @@ func (e *functionRunnerCollectionEntry) getRunnerEntryByAnalyzerField(schemaVers
 	e.mu.RLock()
 	defer e.mu.RUnlock()
 
-	versionRunners, ok := e.versionRunners[schemaVersion]
+	versionRunners, ok := e.getVersionRunnerLocked(schemaVersion)
 	if !ok {
 		return nil, false
 	}
@@ -645,51 +710,35 @@ func (e *functionRunnerCollectionEntry) getRunnerEntryByAnalyzerField(schemaVers
 	return runnerEntry, true
 }
 
-func (e *functionRunnerCollectionEntry) RunWithRunnerEntries(
-	ctx context.Context,
-	runnerEntries []*functionRunnerEntry,
-	initRunner bool,
-	run func([]FunctionRunner) error,
-) (bool, error) {
-	return runWithRunnerEntries(ctx, runnerEntries, initRunner, run)
-}
-
-func (e *functionRunnerCollectionEntry) runWithVersionRunners(
-	ctx context.Context,
-	schemaVersion int32,
-	initRunner bool,
-	run func([]FunctionRunner, []int64) error,
-) (bool, error) {
-	runnerEntries, outputFieldIDs, ok := e.getVersionRunnerEntries(schemaVersion)
-	if !ok {
-		return false, nil
-	}
-	return runWithRunnerEntries(ctx, runnerEntries, initRunner, func(runners []FunctionRunner) error {
-		return run(runners, outputFieldIDs)
-	})
-}
-
 func (e *functionRunnerCollectionEntry) Materialize(
 	ctx context.Context,
 	schema *schemapb.CollectionSchema,
 	body *msgpb.InsertRequest,
 ) (bool, error) {
-	if schema == nil {
-		return false, merr.WrapErrFunctionFailedMsg("collection schema is nil")
-	}
-	changed, ok, err := e.TryMaterialize(schema.GetVersion(), body)
-	if err != nil {
-		return false, err
-	}
-	if ok {
-		return changed, nil
+	if body == nil {
+		return false, merr.WrapErrFunctionFailedMsg("insert request is nil")
 	}
 
-	if _, err := e.ensureVersion("", schema); err != nil {
-		return false, err
+	var runnerEntries []*functionRunnerEntry
+	var outputFieldIDs []int64
+	var ok bool
+	if schema != nil {
+		if _, err := e.ensureVersion("", schema); err != nil {
+			return false, err
+		}
+		runnerEntries, outputFieldIDs, ok = e.getVersionRunnerEntries(schema.GetVersion())
+	} else {
+		runnerEntries, outputFieldIDs, ok = e.getVersionRunnerEntries(LatestFunctionRunnerVersion)
+	}
+	if !ok {
+		return false, nil
+	}
+	if len(outputFieldIDs) == 0 || HasAllFieldDataByID(body.GetFieldsData(), outputFieldIDs) {
+		return false, nil
 	}
 
-	ok, err = e.runWithVersionRunners(ctx, schema.GetVersion(), true, func(runners []FunctionRunner, _ []int64) error {
+	changed := false
+	ok, err := runWithRunnerEntries(ctx, runnerEntries, true, func(runners []FunctionRunner) error {
 		var runErr error
 		changed, runErr = FillFunctionFields(runners, body)
 		return runErr
@@ -722,7 +771,7 @@ func (e *functionRunnerCollectionEntry) TryMaterialize(
 	}
 
 	changed := false
-	ok, err := e.RunWithRunnerEntries(context.Background(), runnerEntries, false, func(runners []FunctionRunner) error {
+	ok, err := runWithRunnerEntries(context.Background(), runnerEntries, false, func(runners []FunctionRunner) error {
 		var runErr error
 		changed, runErr = FillFunctionFields(runners, body)
 		return runErr
@@ -789,8 +838,8 @@ func (m *functionRunnerManager) getEntry(collectionID int64) *functionRunnerColl
 	return m.entries[collectionID]
 }
 
-func (m *functionRunnerManager) Release(collectionID int64, vchannel string) {
-	if vchannel == "" {
+func (m *functionRunnerManager) Release(collectionID int64, key string) {
+	if key == "" {
 		return
 	}
 
@@ -800,7 +849,7 @@ func (m *functionRunnerManager) Release(collectionID int64, vchannel string) {
 		m.mu.Unlock()
 		return
 	}
-	remove := entry.Release(vchannel)
+	remove := entry.Release(key)
 	if remove {
 		delete(m.entries, collectionID)
 	}
@@ -819,14 +868,14 @@ func (m *functionRunnerManager) Materialize(
 	schema *schemapb.CollectionSchema,
 	body *msgpb.InsertRequest,
 ) (bool, error) {
-	if schema == nil {
-		return false, merr.WrapErrFunctionFailedMsg("collection schema is nil")
-	}
-	if !HasEmbeddingFunctions(schema) {
+	if schema != nil && !HasEmbeddingFunctions(schema) {
 		return false, nil
 	}
 	entry := m.getEntry(collectionID)
 	if entry == nil {
+		if schema == nil {
+			return false, nil
+		}
 		return false, merr.WrapErrFunctionFailedMsg("function runners for collection %d are not allocated", collectionID)
 	}
 	return entry.Materialize(ctx, schema, body)
@@ -886,12 +935,6 @@ func (m *functionRunnerManager) Close() {
 	}
 }
 
-func functionRunnerErrorResult(err error) <-chan error {
-	errCh := make(chan error, 1)
-	errCh <- err
-	return errCh
-}
-
 func BuildEmbeddingRunner(schema *schemapb.CollectionSchema, fn *schemapb.FunctionSchema) (FunctionRunner, error) {
 	if schema == nil {
 		return nil, merr.WrapErrFunctionFailedMsg("collection schema is nil")
@@ -1146,7 +1189,8 @@ func cloneMap[K comparable, V any](values map[K]V) map[K]V {
 }
 
 // FillFunctionData fills function output fields before appending an insert message to WAL.
-// The function runners are expected to have been allocated by collection create or recovery.
+// Passing nil schema uses the latest runner snapshot allocated by collection
+// create, recovery, or update.
 func FillFunctionData(ctx context.Context, collectionID int64, schema *schemapb.CollectionSchema, body *msgpb.InsertRequest) (bool, error) {
 	return defaultFunctionRunnerManager.Materialize(ctx, collectionID, schema, body)
 }
@@ -1158,16 +1202,24 @@ func TryMaterialize(collectionID int64, schemaVersion int32, body *msgpb.InsertR
 	return defaultFunctionRunnerManager.TryMaterialize(collectionID, schemaVersion, body)
 }
 
-func AllocFunctionRunners(collectionID int64, vchannel string, schema *schemapb.CollectionSchema) <-chan error {
-	return defaultFunctionRunnerManager.Alloc(collectionID, vchannel, schema)
+func AllocFunctionRunners(
+	collectionID int64,
+	key string,
+	schema *schemapb.CollectionSchema,
+) error {
+	return defaultFunctionRunnerManager.Alloc(collectionID, key, schema)
 }
 
-func UpdateFunctionRunners(collectionID int64, vchannel string, schema *schemapb.CollectionSchema) <-chan error {
-	return defaultFunctionRunnerManager.Update(collectionID, vchannel, schema)
+func UpdateFunctionRunners(
+	collectionID int64,
+	key string,
+	schema *schemapb.CollectionSchema,
+) error {
+	return defaultFunctionRunnerManager.Update(collectionID, key, schema)
 }
 
-func ReleaseFunctionRunners(collectionID int64, vchannel string) {
-	defaultFunctionRunnerManager.Release(collectionID, vchannel)
+func ReleaseFunctionRunners(collectionID int64, key string) {
+	defaultFunctionRunnerManager.Release(collectionID, key)
 }
 
 func RunWithRunner(

internal/util/function/manager_test.go

@@ -133,9 +133,8 @@ func TestFunctionRunnerManagerAllocRequiresSchema(t *testing.T) {
 	manager, _ := newMockFunctionRunnerManager(t)
 	t.Cleanup(manager.Close)
 
-	errCh := manager.Alloc(1, "v1", nil)
-	require.NotNil(t, errCh)
-	require.ErrorContains(t, <-errCh, "collection schema is nil")
+	err := manager.Alloc(1, "v1", nil)
+	require.ErrorContains(t, err, "collection schema is nil")
 	requireFunctionRunnerEntryRemoved(t, manager, 1)
 }
 
@@ -146,8 +145,7 @@ func TestFunctionRunnerManagerAllocSkipsSchemaWithoutEmbeddingFunctions(t *testi
 	schema := cloneCollectionSchema(newBM25SignatureTestSchema())
 	schema.Functions = nil
 
-	errCh := manager.Alloc(1, "v1", schema)
-	require.Nil(t, errCh)
+	require.NoError(t, manager.Alloc(1, "v1", schema))
 	requireFunctionRunnerEntryRemoved(t, manager, 1)
 }
 
@@ -166,8 +164,7 @@ func TestFunctionRunnerManagerAllocSkipsSchemaWithoutRunnerBackedFunctions(t *te
 		},
 	}
 
-	errCh := manager.Alloc(1, "v1", schema)
-	require.Nil(t, errCh)
+	require.NoError(t, manager.Alloc(1, "v1", schema))
 	requireFunctionRunnerEntryRemoved(t, manager, 1)
 }
 
@@ -177,61 +174,143 @@ func TestFunctionRunnerManagerUpdateCreatesEntryWhenEmbeddingFunctionAppears(t *
 
 	schemaWithoutFunctions := cloneCollectionSchema(newBM25SignatureTestSchema())
 	schemaWithoutFunctions.Functions = nil
-	requireNoErrorFromAsync(t, manager.Alloc(1, "v1", schemaWithoutFunctions))
+	require.NoError(t, manager.Alloc(1, "v1", schemaWithoutFunctions))
 	requireFunctionRunnerEntryRemoved(t, manager, 1)
 
 	schemaWithFunction := newBM25SignatureTestSchema()
 	schemaWithFunction.Version = 2
-	requireNoErrorFromAsync(t, manager.Update(1, "v1", schemaWithFunction))
+	require.NoError(t, manager.Update(1, "v1", schemaWithFunction))
 
-	vchannelVersions, versionRunners, runnerCount := functionRunnerEntrySnapshot(t, manager, 1)
-	require.Equal(t, map[string]int32{"v1": 2}, vchannelVersions)
+	keyVersions, versionRunners, runnerCount := functionRunnerEntrySnapshot(t, manager, 1)
+	require.Equal(t, map[string]int32{"v1": 2}, keyVersions)
 	require.Len(t, versionRunners, 1)
 	require.Equal(t, 1, runnerCount)
+	requireRunnerByOutput(t, manager, 1, schemaWithFunction.GetVersion(), 102)
 	require.Equal(t, int32(1), factory.buildCount.Load())
 }
 
-func TestFunctionRunnerManagerAllocTracksVChannelsBySchemaVersion(t *testing.T) {
+func TestFunctionRunnerManagerUpdateReleasesKeyWhenEmbeddingFunctionsDisappear(t *testing.T) {
+	manager, _ := newMockFunctionRunnerManager(t)
+	t.Cleanup(manager.Close)
+
+	schema := newBM25SignatureTestSchema()
+	require.NoError(t, manager.Alloc(1, "v1", schema))
+	runner := requireRunnerByOutput(t, manager, 1, schema.GetVersion(), 102)
+
+	schemaWithoutFunctions := cloneCollectionSchema(schema)
+	schemaWithoutFunctions.Version = 2
+	schemaWithoutFunctions.Functions = nil
+	require.NoError(t, manager.Update(1, "v1", schemaWithoutFunctions))
+
+	requireFunctionRunnerEntryRemoved(t, manager, 1)
+	require.True(t, runner.isClosed())
+}
+
+func TestFunctionRunnerManagerAllocTracksKeysBySchemaVersion(t *testing.T) {
 	manager, factory := newMockFunctionRunnerManager(t)
 	t.Cleanup(manager.Close)
 
 	schema := newBM25SignatureTestSchema()
-	requireNoErrorFromAsync(t, manager.Alloc(1, "v1", schema))
-	requireNoErrorFromAsync(t, manager.Alloc(1, "v2", schema))
+	require.NoError(t, manager.Alloc(1, "v1", schema))
+	require.NoError(t, manager.Alloc(1, "v2", schema))
 
-	vchannelVersions, versionRunners, runnerCount := functionRunnerEntrySnapshot(t, manager, 1)
-	require.Equal(t, map[string]int32{"v1": 1, "v2": 1}, vchannelVersions)
+	keyVersions, versionRunners, runnerCount := functionRunnerEntrySnapshot(t, manager, 1)
+	require.Equal(t, map[string]int32{"v1": 1, "v2": 1}, keyVersions)
 	require.Len(t, versionRunners, 1)
 	require.Equal(t, 1, runnerCount)
 
+	baseRunner := requireRunnerByOutput(t, manager, 1, schema.GetVersion(), 102)
 	require.Equal(t, int32(1), factory.buildCount.Load())
-	baseRunner := factory.lastRunnerByOutput(t, 102)
 
 	manager.Release(1, "v1")
 	require.False(t, baseRunner.isClosed())
-	vchannelVersions, _, _ = functionRunnerEntrySnapshot(t, manager, 1)
-	require.Len(t, vchannelVersions, 1)
+	keyVersions, _, _ = functionRunnerEntrySnapshot(t, manager, 1)
+	require.Len(t, keyVersions, 1)
 
 	manager.Release(1, "v2")
 	requireFunctionRunnerEntryRemoved(t, manager, 1)
 	require.True(t, baseRunner.isClosed())
 }
 
+func TestFunctionRunnerManagerReleaseWaitsForAllKeys(t *testing.T) {
+	manager, _ := newMockFunctionRunnerManager(t)
+	t.Cleanup(manager.Close)
+
+	schema := newBM25SignatureTestSchema()
+	require.NoError(t, manager.Alloc(1, "WAL-v1", schema))
+	require.NoError(t, manager.Alloc(1, "DELEGATOR-v1", schema))
+
+	walVersion := functionRunnerKeyVersionSnapshot(t, manager, 1, "WAL-v1")
+	require.Equal(t, int32(1), walVersion)
+	delegatorVersion := functionRunnerKeyVersionSnapshot(t, manager, 1, "DELEGATOR-v1")
+	require.Equal(t, int32(1), delegatorVersion)
+
+	baseRunner := requireRunnerByOutput(t, manager, 1, schema.GetVersion(), 102)
+	manager.Release(1, "WAL-v1")
+	require.False(t, baseRunner.isClosed())
+
+	ok, err := manager.RunWithAnalyzer(context.Background(), 1, schema.GetVersion(), 101, func(Analyzer) error {
+		return nil
+	})
+	require.NoError(t, err)
+	require.True(t, ok)
+
+	manager.Release(1, "DELEGATOR-v1")
+	requireFunctionRunnerEntryRemoved(t, manager, 1)
+	require.True(t, baseRunner.isClosed())
+}
+
+func TestFunctionRunnerManagerKeepsOldVersionUntilAllKeysAdvance(t *testing.T) {
+	manager, _ := newMockFunctionRunnerManager(t)
+	t.Cleanup(manager.Close)
+
+	base := newBM25SignatureTestSchema()
+	changedOutput := newSchemaWithChangedOutput(base)
+	require.NoError(t, manager.Alloc(1, "WAL-v1", base))
+	require.NoError(t, manager.Alloc(1, "DELEGATOR-v1", base))
+	baseRunner := requireRunnerByOutput(t, manager, 1, base.GetVersion(), 102)
+
+	require.NoError(t, manager.Update(1, "WAL-v1", changedOutput))
+	changedRunner := requireRunnerByOutput(t, manager, 1, changedOutput.GetVersion(), 104)
+	require.False(t, baseRunner.isClosed())
+	require.False(t, changedRunner.isClosed())
+
+	_, versionRunners, runnerCount := functionRunnerEntrySnapshot(t, manager, 1)
+	require.Contains(t, versionRunners, int32(1))
+	require.Contains(t, versionRunners, int32(2))
+	require.Equal(t, 2, runnerCount)
+
+	require.NoError(t, manager.Update(1, "DELEGATOR-v1", changedOutput))
+	require.True(t, baseRunner.isClosed())
+	require.False(t, changedRunner.isClosed())
+
+	_, versionRunners, runnerCount = functionRunnerEntrySnapshot(t, manager, 1)
+	require.NotContains(t, versionRunners, int32(1))
+	require.Contains(t, versionRunners, int32(2))
+	require.Equal(t, 1, runnerCount)
+
+	manager.Release(1, "WAL-v1")
+	require.False(t, changedRunner.isClosed())
+	manager.Release(1, "DELEGATOR-v1")
+	requireFunctionRunnerEntryRemoved(t, manager, 1)
+	require.True(t, changedRunner.isClosed())
+}
+
 func TestFunctionRunnerManagerUpdateReusesSameSignatureAcrossSchemaVersions(t *testing.T) {
 	manager, factory := newMockFunctionRunnerManager(t)
 	t.Cleanup(manager.Close)
 
 	base := newBM25SignatureTestSchema()
-	requireNoErrorFromAsync(t, manager.Alloc(1, "v1", base))
-	baseRunner := factory.lastRunnerByOutput(t, 102)
+	require.NoError(t, manager.Alloc(1, "v1", base))
+	baseRunner := requireRunnerByOutput(t, manager, 1, base.GetVersion(), 102)
 
 	schemaVersionChanged := cloneCollectionSchema(base)
 	schemaVersionChanged.Version = 2
-	requireNoErrorFromAsync(t, manager.Update(1, "v1", schemaVersionChanged))
+	require.NoError(t, manager.Update(1, "v1", schemaVersionChanged))
 
-	ok, err := manager.getEntry(1).runWithVersionRunners(context.Background(), schemaVersionChanged.GetVersion(), false, func(runners []FunctionRunner, _ []int64) error {
-		require.Len(t, runners, 1)
-		require.True(t, baseRunner == runners[0])
+	ok, err := manager.RunWithRunner(context.Background(), 1, schemaVersionChanged.GetVersion(), 102, func(functionType schemapb.FunctionType, runner FunctionRunner) error {
+		require.Equal(t, schemapb.FunctionType_BM25, functionType)
+		require.True(t, baseRunner == runner)
 		return nil
 	})
 	require.NoError(t, err)
@@ -245,19 +324,19 @@ func TestFunctionRunnerManagerUpdateReusesSameSignatureAcrossSchemaVersions(t *t
 	require.Equal(t, 1, runnerCount)
 }
 
-func TestFunctionRunnerManagerUpdateKeepsOldVersionUntilAllVChannelsAdvance(t *testing.T) {
-	manager, factory := newMockFunctionRunnerManager(t)
+func TestFunctionRunnerManagerUpdateKeepsOldVersionUntilAllKeysAdvance(t *testing.T) {
+	manager, _ := newMockFunctionRunnerManager(t)
 	t.Cleanup(manager.Close)
 
 	base := newBM25SignatureTestSchema()
 	changedOutput := newSchemaWithChangedOutput(base)
 
-	requireNoErrorFromAsync(t, manager.Alloc(1, "v1", base))
-	requireNoErrorFromAsync(t, manager.Alloc(1, "v2", base))
-	baseRunner := factory.lastRunnerByOutput(t, 102)
+	require.NoError(t, manager.Alloc(1, "v1", base))
+	require.NoError(t, manager.Alloc(1, "v2", base))
+	baseRunner := requireRunnerByOutput(t, manager, 1, base.GetVersion(), 102)
 
-	requireNoErrorFromAsync(t, manager.Update(1, "v1", changedOutput))
-	changedRunner := factory.lastRunnerByOutput(t, 104)
+	require.NoError(t, manager.Update(1, "v1", changedOutput))
+	changedRunner := requireRunnerByOutput(t, manager, 1, changedOutput.GetVersion(), 104)
 	require.False(t, baseRunner.isClosed())
 	require.False(t, changedRunner.isClosed())
 
@@ -282,7 +361,7 @@ func TestFunctionRunnerManagerUpdateKeepsOldVersionUntilAllVChannelsAdvance(t *t
 	require.False(t, HasFieldData(newBody.GetFieldsData(), 102))
 	require.True(t, HasFieldData(newBody.GetFieldsData(), 104))
 
-	requireNoErrorFromAsync(t, manager.Update(1, "v2", changedOutput))
+	require.NoError(t, manager.Update(1, "v2", changedOutput))
 	require.True(t, baseRunner.isClosed())
 	require.False(t, changedRunner.isClosed())
 
@@ -302,21 +381,27 @@ func TestFunctionRunnerManagerUpdateBuildsOnlyAddedFunction(t *testing.T) {
 	t.Cleanup(manager.Close)
 
 	base := newBM25SignatureTestSchema()
-	requireNoErrorFromAsync(t, manager.Alloc(1, "v1", base))
-	baseRunner := factory.lastRunnerByOutput(t, 102)
+	require.NoError(t, manager.Alloc(1, "v1", base))
+	baseRunner := requireRunnerByOutput(t, manager, 1, base.GetVersion(), 102)
 
 	addedFunction := newSchemaWithAddedFunction(base)
-	requireNoErrorFromAsync(t, manager.Update(1, "v1", addedFunction))
-	addedRunner := factory.lastRunnerByOutput(t, 104)
+	require.NoError(t, manager.Update(1, "v1", addedFunction))
+	addedRunner := requireRunnerByOutput(t, manager, 1, addedFunction.GetVersion(), 104)
 
 	require.Equal(t, int32(2), factory.buildCount.Load())
 	require.False(t, baseRunner.isClosed())
 	require.False(t, addedRunner.isClosed())
 
-	ok, err := manager.getEntry(1).runWithVersionRunners(context.Background(), addedFunction.GetVersion(), false, func(runners []FunctionRunner, _ []int64) error {
-		require.Len(t, runners, 2)
-		require.Contains(t, runners, baseRunner)
-		require.Contains(t, runners, addedRunner)
+	ok, err := manager.RunWithRunner(context.Background(), 1, addedFunction.GetVersion(), 102, func(functionType schemapb.FunctionType, runner FunctionRunner) error {
+		require.Equal(t, schemapb.FunctionType_BM25, functionType)
+		require.True(t, baseRunner == runner)
+		return nil
+	})
+	require.NoError(t, err)
+	require.True(t, ok)
+	ok, err = manager.RunWithRunner(context.Background(), 1, addedFunction.GetVersion(), 104, func(functionType schemapb.FunctionType, runner FunctionRunner) error {
+		require.Equal(t, schemapb.FunctionType_BM25, functionType)
+		require.True(t, addedRunner == runner)
 		return nil
 	})
 	require.NoError(t, err)
@@ -329,6 +414,14 @@ func TestFunctionRunnerManagerUpdateBuildsOnlyAddedFunction(t *testing.T) {
 	require.True(t, ok)
 	require.True(t, HasFieldData(body.GetFieldsData(), 102))
 	require.True(t, HasFieldData(body.GetFieldsData(), 104))
+
+	latestBody := newBM25InsertRequest("latest message")
+	changed, ok, err = manager.TryMaterialize(1, LatestFunctionRunnerVersion, latestBody)
+	require.NoError(t, err)
+	require.True(t, changed)
+	require.True(t, ok)
+	require.True(t, HasFieldData(latestBody.GetFieldsData(), 102))
+	require.True(t, HasFieldData(latestBody.GetFieldsData(), 104))
 }
 
 func TestFunctionRunnerManagerReleaseRemovesInitializingEntry(t *testing.T) {
@@ -339,7 +432,9 @@ func TestFunctionRunnerManagerReleaseRemovesInitializingEntry(t *testing.T) {
 	started := make(chan struct{})
 	releaseBuild := make(chan struct{})
 	var once sync.Once
+	buildDone := make(chan struct{})
 	patchBuildEmbeddingRunner(t, func(schema *schemapb.CollectionSchema, fn *schemapb.FunctionSchema) (FunctionRunner, error) {
+		defer close(buildDone)
 		once.Do(func() {
 			close(started)
 		})
@@ -347,15 +442,14 @@ func TestFunctionRunnerManagerReleaseRemovesInitializingEntry(t *testing.T) {
 		return newTestFunctionRunner(schema, fn)
 	})
 
-	errCh := manager.Alloc(1, "v1", schema)
-	require.NotNil(t, errCh)
+	require.NoError(t, manager.Alloc(1, "v1", schema))
 	<-started
 
 	manager.Release(1, "v1")
 	requireFunctionRunnerEntryRemoved(t, manager, 1)
 
 	close(releaseBuild)
-	require.NoError(t, <-errCh)
+	<-buildDone
 }
 
 func TestFunctionRunnerManagerReleaseCloseDoesNotBlockManager(t *testing.T) {
@@ -378,7 +472,8 @@ func TestFunctionRunnerManagerReleaseCloseDoesNotBlockManager(t *testing.T) {
 		return runner, nil
 	})
 
-	requireNoErrorFromAsync(t, manager.Alloc(1, "v1", schema))
+	require.NoError(t, manager.Alloc(1, "v1", schema))
+	requireRunnerByOutput(t, manager, 1, schema.GetVersion(), 102)
 
 	releaseDone := make(chan struct{})
 	go func() {
@@ -394,12 +489,7 @@ func TestFunctionRunnerManagerReleaseCloseDoesNotBlockManager(t *testing.T) {
 
 	allocErr := make(chan error, 1)
 	go func() {
-		errCh := manager.Alloc(2, "v2", schema)
-		if errCh == nil {
-			allocErr <- nil
-			return
-		}
-		allocErr <- <-errCh
+		allocErr <- manager.Alloc(2, "v2", schema)
 	}()
 
 	select {
@@ -437,7 +527,7 @@ func TestFunctionRunnerManagerRunWithRunnerProtectsConcurrentClose(t *testing.T)
 		runner.closeStarted = closeStarted
 		return runner, nil
 	})
-	requireNoErrorFromAsync(t, manager.Alloc(1, "v1", schema))
+	require.NoError(t, manager.Alloc(1, "v1", schema))
 
 	runStarted := make(chan struct{})
 	releaseRun := make(chan struct{})
@@ -486,10 +576,10 @@ func TestFunctionRunnerManagerRunWithAnalyzerUsesBM25Runner(t *testing.T) {
 	t.Cleanup(manager.Close)
 
 	schema := newBM25SignatureTestSchema()
-	requireNoErrorFromAsync(t, manager.Alloc(1, "v1", schema))
+	require.NoError(t, manager.Alloc(1, "v1", schema))
 
 	var tokens [][]*milvuspb.AnalyzerToken
-	ok, err := manager.RunWithAnalyzer(context.Background(), 1, schema.GetVersion(), 101, func(analyzer Analyzer) error {
+	ok, err := manager.RunWithAnalyzer(context.Background(), 1, LatestFunctionRunnerVersion, 101, func(analyzer Analyzer) error {
 		var analyzeErr error
 		tokens, analyzeErr = analyzer.BatchAnalyze(false, false, []string{"hello world"})
 		return analyzeErr
@@ -500,6 +590,56 @@ func TestFunctionRunnerManagerRunWithAnalyzerUsesBM25Runner(t *testing.T) {
 	require.Equal(t, "hello world", tokens[0][0].GetToken())
 }
 
+func TestFunctionRunnerManagerRunWithRunnerLatestVersionUsesLatestSnapshot(t *testing.T) {
+	manager, _ := newMockFunctionRunnerManager(t)
+	t.Cleanup(manager.Close)
+
+	base := newBM25SignatureTestSchema()
+	require.NoError(t, manager.Alloc(1, "v1", base))
+
+	changedOutput := newSchemaWithChangedOutput(base)
+	require.NoError(t, manager.Update(1, "v1", changedOutput))
+
+	ok, err := manager.RunWithRunner(context.Background(), 1, LatestFunctionRunnerVersion, 104, func(functionType schemapb.FunctionType, runner FunctionRunner) error {
+		require.Equal(t, schemapb.FunctionType_BM25, functionType)
+		require.Equal(t, int64(104), runner.GetOutputFields()[0].GetFieldID())
+		return nil
+	})
+	require.NoError(t, err)
+	require.True(t, ok)
+}
+
+func TestFunctionRunnerManagerSchemaVersionZeroIsExplicit(t *testing.T) {
+	manager, _ := newMockFunctionRunnerManager(t)
+	t.Cleanup(manager.Close)
+
+	base := newBM25SignatureTestSchema()
+	base.Version = 0
+	require.NoError(t, manager.Alloc(1, "v0", base))
+
+	changedOutput := newSchemaWithChangedOutput(base)
+	changedOutput.Version = 1
+	require.NoError(t, manager.Alloc(1, "v1", changedOutput))
+	requireRunnerByOutput(t, manager, 1, base.GetVersion(), 102)
+	requireRunnerByOutput(t, manager, 1, changedOutput.GetVersion(), 104)
+
+	versionZeroBody := newBM25InsertRequest("version zero message")
+	changed, ok, err := manager.TryMaterialize(1, 0, versionZeroBody)
+	require.NoError(t, err)
+	require.True(t, changed)
+	require.True(t, ok)
+	require.True(t, HasFieldData(versionZeroBody.GetFieldsData(), 102))
+	require.False(t, HasFieldData(versionZeroBody.GetFieldsData(), 104))
+
+	latestBody := newBM25InsertRequest("latest message")
+	changed, ok, err = manager.TryMaterialize(1, LatestFunctionRunnerVersion, latestBody)
+	require.NoError(t, err)
+	require.True(t, changed)
+	require.True(t, ok)
+	require.False(t, HasFieldData(latestBody.GetFieldsData(), 102))
+	require.True(t, HasFieldData(latestBody.GetFieldsData(), 104))
+}
+
 func TestFunctionRunnerManagerMaterializeRequiresAllocation(t *testing.T) {
 	manager, _ := newMockFunctionRunnerManager(t)
 	t.Cleanup(manager.Close)
@@ -510,6 +650,24 @@ func TestFunctionRunnerManagerMaterializeRequiresAllocation(t *testing.T) {
 	requireFunctionRunnerEntryRemoved(t, manager, 1)
 }
 
+func TestFunctionRunnerManagerMaterializeNilSchemaUsesLatestSnapshot(t *testing.T) {
+	manager, _ := newMockFunctionRunnerManager(t)
+	t.Cleanup(manager.Close)
+
+	base := newBM25SignatureTestSchema()
+	require.NoError(t, manager.Alloc(1, "v1", base))
+
+	changedOutput := newSchemaWithChangedOutput(base)
+	require.NoError(t, manager.Update(1, "v1", changedOutput))
+
+	body := newBM25InsertRequest("latest message")
+	changed, err := manager.Materialize(context.Background(), 1, nil, body)
+	require.NoError(t, err)
+	require.True(t, changed)
+	require.False(t, HasFieldData(body.GetFieldsData(), 102))
+	require.True(t, HasFieldData(body.GetFieldsData(), 104))
+}
+
 func TestFunctionRunnerManagerMaterializeNoEmbeddingFunctionDoesNotRequireAllocation(t *testing.T) {
 	manager := newFunctionRunnerManager()
 	t.Cleanup(manager.Close)
@@ -660,17 +818,25 @@ func TestFunctionRunnerManagerAllocRetriesFailedInitOnNextRequest(t *testing.T)
 
 	var buildCount atomic.Int32
 	expectedErr := errors.New("mock recover init failed")
+	started := make(chan struct{})
+	release := make(chan struct{})
+	var once sync.Once
 	patchBuildEmbeddingRunner(t, func(schema *schemapb.CollectionSchema, fn *schemapb.FunctionSchema) (FunctionRunner, error) {
-		if buildCount.Add(1) == 1 {
+		count := buildCount.Add(1)
+		if count == 1 {
+			once.Do(func() {
+				close(started)
+			})
+			<-release
 			return nil, expectedErr
 		}
 		return newTestFunctionRunner(schema, fn)
 	})
 
-	errCh := manager.Alloc(1, "v1", schema)
-	require.NotNil(t, errCh)
-	require.NoError(t, <-errCh)
-	requireFunctionRunnerState(t, manager, 1, signature, functionRunnerStateFailed)
+	require.NoError(t, manager.Alloc(1, "v1", schema))
+	<-started
+	close(release)
+	requireFunctionRunnerStateEventually(t, manager, 1, signature, functionRunnerStateFailed)
 
 	body := newBM25InsertRequest("message")
 	changed, err := manager.Materialize(context.Background(), 1, schema, body)
@@ -686,26 +852,22 @@ func TestFunctionRunnerManagerAllocDoesNotStartWhenReady(t *testing.T) {
 	t.Cleanup(manager.Close)
 
 	schema := newBM25SignatureTestSchema()
-	requireNoErrorFromAsync(t, manager.Alloc(1, "v1", schema))
+	require.NoError(t, manager.Alloc(1, "v1", schema))
+	runner := requireRunnerByOutput(t, manager, 1, schema.GetVersion(), 102)
 
-	errCh := manager.Alloc(1, "v1", schema)
-	require.Nil(t, errCh)
+	require.NoError(t, manager.Alloc(1, "v1", schema))
 	require.Equal(t, int32(1), factory.buildCount.Load())
-}
 
-func requireNoErrorFromAsync(t *testing.T, errCh <-chan error) {
-	t.Helper()
-
-	if errCh != nil {
-		require.NoError(t, <-errCh)
-	}
+	manager.Release(1, "v1")
+	requireFunctionRunnerEntryRemoved(t, manager, 1)
+	require.True(t, runner.isClosed())
 }
 
 func allocVChannelForTest(manager *functionRunnerManager, collectionID int64, vchannel string, schemaVersion int32) {
 	entry := manager.getOrCreateEntry(collectionID)
 	entry.mu.Lock()
 	defer entry.mu.Unlock()
-	entry.vchannelVersions[vchannel] = schemaVersion
+	entry.keyVersions[vchannel] = schemaVersion
 }
 
 func functionRunnerEntrySnapshot(
@@ -723,15 +885,36 @@ func functionRunnerEntrySnapshot(
 	entry.mu.RLock()
 	defer entry.mu.RUnlock()
 
-	vchannelVersions := make(map[string]int32, len(entry.vchannelVersions))
-	for vchannel, version := range entry.vchannelVersions {
-		vchannelVersions[vchannel] = version
+	keyVersions := make(map[string]int32, len(entry.keyVersions))
+	for key, version := range entry.keyVersions {
+		keyVersions[key] = version
 	}
 	versionRunners := make(map[int32]struct{}, len(entry.versionRunners))
 	for version := range entry.versionRunners {
 		versionRunners[version] = struct{}{}
 	}
-	return vchannelVersions, versionRunners, len(entry.runners)
+	return keyVersions, versionRunners, len(entry.runners)
+}
+
+func functionRunnerKeyVersionSnapshot(
+	t *testing.T,
+	manager *functionRunnerManager,
+	collectionID int64,
+	key string,
+) int32 {
+	t.Helper()
+
+	manager.mu.RLock()
+	entry := manager.entries[collectionID]
+	manager.mu.RUnlock()
+	require.NotNil(t, entry)
+
+	entry.mu.RLock()
+	defer entry.mu.RUnlock()
+
+	version, ok := entry.keyVersions[key]
+	require.True(t, ok)
+	return version
 }
 
 func requireFunctionRunnerEntryRemoved(t *testing.T, manager *functionRunnerManager, collectionID int64) {
@@ -766,6 +949,58 @@ func requireFunctionRunnerState(
 	require.Equal(t, state, runnerEntry.state)
 }
 
+func requireFunctionRunnerStateEventually(
+	t *testing.T,
+	manager *functionRunnerManager,
+	collectionID int64,
+	signature string,
+	state functionRunnerState,
+) {
+	t.Helper()
+
+	require.Eventually(t, func() bool {
+		manager.mu.RLock()
+		entry := manager.entries[collectionID]
+		manager.mu.RUnlock()
+		if entry == nil {
+			return false
+		}
+
+		entry.mu.RLock()
+		runnerEntry := entry.runners[signature]
+		entry.mu.RUnlock()
+		if runnerEntry == nil {
+			return false
+		}
+
+		runnerEntry.mu.RLock()
+		defer runnerEntry.mu.RUnlock()
+		return runnerEntry.state == state
+	}, time.Second, 10*time.Millisecond)
+}
+
+func requireRunnerByOutput(
+	t *testing.T,
+	manager *functionRunnerManager,
+	collectionID int64,
+	schemaVersion int32,
+	outputFieldID int64,
+) *testFunctionRunner {
+	t.Helper()
+
+	var testRunner *testFunctionRunner
+	ok, err := manager.RunWithRunner(context.Background(), collectionID, schemaVersion, outputFieldID, func(functionType schemapb.FunctionType, runner FunctionRunner) error {
+		castRunner, ok := runner.(*testFunctionRunner)
+		require.True(t, ok)
+		testRunner = castRunner
+		return nil
+	})
+	require.NoError(t, err)
+	require.True(t, ok)
+	require.NotNil(t, testRunner)
+	return testRunner
+}
+
 func firstEmbeddingSignature(t *testing.T, schema *schemapb.CollectionSchema) string {
 	t.Helper()
 
@@ -816,20 +1051,6 @@ func (f *testFunctionRunnerFactory) Build(schema *schemapb.CollectionSchema, fn
 	return runner, nil
 }
 
-func (f *testFunctionRunnerFactory) lastRunnerByOutput(t *testing.T, outputFieldID int64) *testFunctionRunner {
-	t.Helper()
-
-	f.mu.Lock()
-	defer f.mu.Unlock()
-	for i := len(f.runners) - 1; i >= 0; i-- {
-		if f.runners[i].GetOutputFields()[0].GetFieldID() == outputFieldID {
-			return f.runners[i]
-		}
-	}
-	require.Failf(t, "runner not found", "output field: %d", outputFieldID)
-	return nil
-}
-
 type testFunctionRunner struct {
 	schema       *schemapb.FunctionSchema
 	inputFields  []*schemapb.FieldSchema