enhance: Support L0 chain (#51012)

issue: https://github.com/milvus-io/milvus/issues/51011 Signed-off-by: junjie.jiang <junjie.jiang@zilliz.com>
2026-07-21 02:05:41 +00:00 · 2026-07-07 13:38:30 +08:00
parent 4fad6e9dbb
commit 116877a0aa
23 changed files with 1438 additions and 145 deletions
@@ -7,7 +7,7 @@ require (
 	github.com/cockroachdb/errors v1.9.1
 	github.com/google/uuid v1.6.0
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
-	github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260615062223-e378dbfa23a2
+	github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260625075625-7262f8042a55
 	github.com/milvus-io/milvus/pkg/v3 v3.0.0-beta
 	github.com/quasilyte/go-ruleguard/dsl v0.3.23
 	github.com/samber/lo v1.52.0
@@ -197,10 +197,8 @@ github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27k
 github.com/mattn/goveralls v0.0.2/go.mod h1:8d1ZMHsd7fW6IRPKQh46F2WRpyib5/X4FOpevwGNQEw=
 github.com/mediocregopher/radix/v3 v3.4.2/go.mod h1:8FL3F6UQRXHXIBSPUs5h0RybMF8i4n7wVopoX3x7Bv8=
 github.com/microcosm-cc/bluemonday v1.0.2/go.mod h1:iVP4YcDBq+n/5fb23BhYFvIMq/leAFZyRl6bYmGDlGc=
-github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260612100615-43795e8f8f6e h1:axPnOhjsWqqxk0dOQRSIU2mdg52Kyl/8E6iDFZy/IpE=
-github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260612100615-43795e8f8f6e/go.mod h1:rbKpv5JToISTKTTLl0duL5r6wbYnjJ9SsD0QgXMzKy0=
-github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260615062223-e378dbfa23a2 h1:D0aOsXFh2wCaTNV4X6A2/lZIgHVFpR9XhKgQuITD+Zg=
-github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260615062223-e378dbfa23a2/go.mod h1:rbKpv5JToISTKTTLl0duL5r6wbYnjJ9SsD0QgXMzKy0=
+github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260625075625-7262f8042a55 h1:U07yIwkWsk7wpGCkWZlY8lSVEFNWKpmm7M+aF5D+wg8=
+github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260625075625-7262f8042a55/go.mod h1:rbKpv5JToISTKTTLl0duL5r6wbYnjJ9SsD0QgXMzKy0=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
@@ -47,6 +47,36 @@ func validateFunctionChainSearchRequest(request *milvuspb.SearchRequest, isAdvan
 	return nil
 }

+func splitFunctionChainsByStage(chains []*schemapb.FunctionChain) ([]*schemapb.FunctionChain, []*schemapb.FunctionChain, error) {
+	l2Chains := make([]*schemapb.FunctionChain, 0)
+	querynodeChains := make([]*schemapb.FunctionChain, 0)
+	seenStages := make(map[schemapb.FunctionChainStage]struct{}, len(chains))
+
+	for i, chainPB := range chains {
+		if chainPB == nil {
+			return nil, nil, merr.WrapErrParameterInvalidMsg("function chain[%d] is nil", i)
+		}
+		stage := chainPB.GetStage()
+		if _, ok := seenStages[stage]; ok {
+			return nil, nil, merr.WrapErrParameterInvalidMsg("function chain stage %s appears more than once", stage.String())
+		}
+		seenStages[stage] = struct{}{}
+
+		switch stage {
+		case schemapb.FunctionChainStage_FunctionChainStageL2Rerank:
+			l2Chains = append(l2Chains, chainPB)
+		case schemapb.FunctionChainStage_FunctionChainStageL0Rerank:
+			querynodeChains = append(querynodeChains, chainPB)
+		case schemapb.FunctionChainStage_FunctionChainStageL1Rerank:
+			return nil, nil, merr.WrapErrParameterInvalidMsg("function chain[%d] stage %s is not supported yet", i, stage.String())
+		default:
+			return nil, nil, merr.WrapErrParameterInvalidMsg("function chain[%d] stage %s is not supported in search request", i, stage.String())
+		}
+	}
+
+	return l2Chains, querynodeChains, nil
+}
+
 func newFunctionChainRerankMeta(chains []*schemapb.FunctionChain, schema *schemaInfo) (*functionChainRerankMeta, error) {
 	if len(chains) == 0 {
 		return nil, nil
@@ -57,6 +87,9 @@ func newFunctionChainRerankMeta(chains []*schemapb.FunctionChain, schema *schema
 	var repr *chain.ChainRepr

 	for i, pb := range chains {
+		if pb == nil {
+			return nil, merr.WrapErrParameterInvalidMsg("function chain[%d] is nil", i)
+		}
 		stage := pb.GetStage()
 		if _, ok := seenStages[stage]; ok {
 			return nil, merr.WrapErrParameterInvalidMsg("function chain stage %s appears more than once", stage.String())
@@ -58,6 +58,50 @@ func TestValidateFunctionChainSearchRequest(t *testing.T) {
 	})
 }

+func TestSplitFunctionChainsByStage(t *testing.T) {
+	t.Run("split l0 and l2 chains", func(t *testing.T) {
+		l0Chain := l0FunctionChain()
+		l2Chain := l2FunctionChain(mapOp(types.ScoreFieldName, "expr", columnArg(types.ScoreFieldName)))
+
+		l2Chains, querynodeChains, err := splitFunctionChainsByStage([]*schemapb.FunctionChain{l0Chain, l2Chain})
+		require.NoError(t, err)
+		assert.Equal(t, []*schemapb.FunctionChain{l2Chain}, l2Chains)
+		assert.Equal(t, []*schemapb.FunctionChain{l0Chain}, querynodeChains)
+	})
+
+	t.Run("l0 chain is shallow routed without op validation", func(t *testing.T) {
+		l0Chain := l0FunctionChain()
+
+		l2Chains, querynodeChains, err := splitFunctionChainsByStage([]*schemapb.FunctionChain{l0Chain})
+		require.NoError(t, err)
+		assert.Empty(t, l2Chains)
+		assert.Equal(t, []*schemapb.FunctionChain{l0Chain}, querynodeChains)
+	})
+
+	t.Run("nil chain", func(t *testing.T) {
+		_, _, err := splitFunctionChainsByStage([]*schemapb.FunctionChain{nil})
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "function chain[0] is nil")
+	})
+
+	t.Run("duplicate stage", func(t *testing.T) {
+		_, _, err := splitFunctionChainsByStage([]*schemapb.FunctionChain{
+			l0FunctionChain(),
+			l0FunctionChain(),
+		})
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "appears more than once")
+	})
+
+	t.Run("l1 is not supported yet", func(t *testing.T) {
+		_, _, err := splitFunctionChainsByStage([]*schemapb.FunctionChain{{
+			Stage: schemapb.FunctionChainStage_FunctionChainStageL1Rerank,
+		}})
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "is not supported yet")
+	})
+}
+
 func TestNewFunctionChainRerankMeta(t *testing.T) {
 	schema := newFunctionChainTestSchema()

@@ -235,6 +279,13 @@ func l2FunctionChain(ops ...*schemapb.FunctionChainOp) *schemapb.FunctionChain {
 	}
 }

+func l0FunctionChain(ops ...*schemapb.FunctionChainOp) *schemapb.FunctionChain {
+	return &schemapb.FunctionChain{
+		Stage: schemapb.FunctionChainStage_FunctionChainStageL0Rerank,
+		Ops:   ops,
+	}
+}
+
 func l2LimitFunctionChain(limit int64) *schemapb.FunctionChain {
 	return l2FunctionChain(&schemapb.FunctionChainOp{
 		Op: types.OpTypeLimit,
@@ -893,18 +893,26 @@ func (t *searchTask) initSearchRequest(ctx context.Context) error {
 	if err := validateFunctionChainSearchRequest(t.request, false); err != nil {
 		return err
 	}
+	var querynodeFunctionChains []*schemapb.FunctionChain
 	if len(t.request.GetFunctionChains()) > 0 {
-		meta, err := newFunctionChainRerankMeta(t.request.GetFunctionChains(), t.schema)
+		l2Chains, qnChains, err := splitFunctionChainsByStage(t.request.GetFunctionChains())
 		if err != nil {
 			return err
 		}
-		t.rerankMeta = meta
+		if len(l2Chains) > 0 {
+			meta, err := newFunctionChainRerankMeta(l2Chains, t.schema)
+			if err != nil {
+				return err
+			}
+			t.rerankMeta = meta
+		}
+		querynodeFunctionChains = qnChains
 	} else if t.request.FunctionScore != nil {
 		t.rerankMeta = newRerankMeta(t.schema.CollectionSchema, t.request.FunctionScore)
 	}

 	// order_by and function rerank cannot be used together
-	if len(t.orderByFields) > 0 && t.rerankMeta != nil {
+	if len(t.orderByFields) > 0 && (t.rerankMeta != nil || len(querynodeFunctionChains) > 0) {
 		return merr.WrapErrParameterInvalidMsg("order_by and function rerank cannot be used together: they specify conflicting sort criteria")
 	}

@@ -1004,6 +1012,7 @@ func (t *searchTask) initSearchRequest(ctx context.Context) error {
 		}
 	}
 	plan.Namespace = namespaceForPlan(t.schema.CollectionSchema, t.request.Namespace)
+	plan.QuerynodeFunctionChains = querynodeFunctionChains

 	// Propagate agg-path overrides into queryInfo BEFORE plan serialization so
 	// segcore sees the derived topK / groupSize and plural GroupByFieldIds.
@@ -5491,6 +5491,19 @@ func TestSearchTask_FunctionChainRerankMeta(t *testing.T) {
 		}
 		return request
 	}
+	newL0FunctionChainRequest := func() (*milvuspb.SearchRequest, *schemapb.FunctionChain) {
+		request := newRequest()
+		chainPB := l0FunctionChain()
+		request.FunctionChains = []*schemapb.FunctionChain{chainPB}
+		return request, chainPB
+	}
+	newMixedFunctionChainRequest := func() (*milvuspb.SearchRequest, *schemapb.FunctionChain, *schemapb.FunctionChain) {
+		request := newRequest()
+		l0Chain := l0FunctionChain()
+		l2Chain := l2FunctionChain(mapOp("score1", "expr", columnArg("ts")), mapOp("$score", "expr", columnArg("score1"), columnArg("$score")))
+		request.FunctionChains = []*schemapb.FunctionChain{l0Chain, l2Chain}
+		return request, l0Chain, l2Chain
+	}
 	newTask := func(request *milvuspb.SearchRequest) *searchTask {
 		translatedOutputFields, _, _, _, _, err := translateOutputFields(request.GetOutputFields(), schemaInfo, true)
 		require.NoError(t, err)
@@ -5592,6 +5605,44 @@ func TestSearchTask_FunctionChainRerankMeta(t *testing.T) {
 		assert.Equal(t, []string{"ts"}, meta.GetInputFieldNames())
 		assert.Equal(t, []int64{101}, meta.GetInputFieldIDs())
 	})
+
+	t.Run("ordinary search routes l0 chain to querynode plan", func(t *testing.T) {
+		request, chainPB := newL0FunctionChainRequest()
+		task := newTask(request)
+
+		require.NoError(t, task.initSearchRequest(ctx))
+		assert.Nil(t, task.rerankMeta)
+
+		plan := &planpb.PlanNode{}
+		require.NoError(t, proto.Unmarshal(task.SerializedExprPlan, plan))
+		require.Len(t, plan.GetQuerynodeFunctionChains(), 1)
+		assert.True(t, proto.Equal(chainPB, plan.GetQuerynodeFunctionChains()[0]))
+	})
+
+	t.Run("ordinary search routes l0 chain and keeps l2 rerank meta", func(t *testing.T) {
+		request, l0Chain, _ := newMixedFunctionChainRequest()
+		task := newTask(request)
+
+		require.NoError(t, task.initSearchRequest(ctx))
+		meta, ok := task.rerankMeta.(*functionChainRerankMeta)
+		require.True(t, ok)
+		assert.Equal(t, []string{"ts"}, meta.GetInputFieldNames())
+
+		plan := &planpb.PlanNode{}
+		require.NoError(t, proto.Unmarshal(task.SerializedExprPlan, plan))
+		require.Len(t, plan.GetQuerynodeFunctionChains(), 1)
+		assert.True(t, proto.Equal(l0Chain, plan.GetQuerynodeFunctionChains()[0]))
+	})
+
+	t.Run("ordinary search rejects order by with l0 function chain", func(t *testing.T) {
+		request, _ := newL0FunctionChainRequest()
+		request.SearchParams = append(request.SearchParams, &commonpb.KeyValuePair{Key: OrderByFieldsKey, Value: "ts:asc"})
+		task := newTask(request)
+
+		err := task.initSearchRequest(ctx)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "order_by and function rerank cannot be used together")
+	})
 }

 func TestSearchTask_AddHighlightTask(t *testing.T) {
@@ -21,7 +21,6 @@ import (
 	"fmt"

 	"github.com/apache/arrow/go/v17/arrow"
-	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/proto"

 	"github.com/milvus-io/milvus/internal/querynodev2/segments"
@@ -34,29 +33,14 @@ import (
 )

 const (
-	boostScoreColumnPrefix = "$boost_score_"
-	functionScoreColumn    = "$function_score"
+	boostScoreColumnPrefix = "boost_score_"
+	functionScoreColumn    = "function_score"
 )

 func boostScoreColumn(index int) string {
 	return fmt.Sprintf("%s%d", boostScoreColumnPrefix, index)
 }

-func boostReduceColumns(cols []string) []string {
-	selected := make([]string, 0, 4)
-	for _, col := range cols {
-		switch {
-		case col == types.IDFieldName,
-			col == types.ScoreFieldName,
-			col == types.SegOffsetFieldName,
-			col == elementIndicesCol,
-			isGroupByColumnName(col):
-			selected = append(selected, col)
-		}
-	}
-	return selected
-}
-
 func extractPlanScorers(serializedPlan []byte) ([]*planpb.ScoreFunction, error) {
 	plan, err := extractPlanWithScorers(serializedPlan)
 	if err != nil || plan == nil {
@@ -136,9 +120,7 @@ func buildBoostScoreChain(
 		return nil, err
 	}

-	boostChain.Select(boostReduceColumns(df.ColumnNames())...)
-	boostChain.Sort(types.ScoreFieldName, true, types.IDFieldName)
-	return boostChain, nil
+	return appendL0RerankReduceContract(boostChain), nil
 }

 func appendBoostScoreColumns(
@@ -186,14 +168,17 @@ func appendFinalBoostScore(boostChain *chain.FuncChain, functionScoreCol string,
 }

 func (t *SearchTask) applyBoostScores(segDFs []*chain.DataFrame, searchedSegments []segments.Segment, searchReq *segcore.SearchRequest) error {
-	if len(segDFs) != len(searchedSegments) {
-		return merr.WrapErrServiceInternal(fmt.Sprintf("boost_score: DataFrame count %d does not match segment count %d", len(segDFs), len(searchedSegments)))
-	}
-
 	plan, err := extractPlanWithScorers(t.req.GetReq().GetSerializedExprPlan())
 	if err != nil {
 		return merr.WrapErrServiceInternal(fmt.Sprintf("boost_score: failed to parse search plan scorers: %v", err))
 	}
+	return t.applyBoostScoresWithPlan(segDFs, plan, searchedSegments, searchReq)
+}
+
+func (t *SearchTask) applyBoostScoresWithPlan(segDFs []*chain.DataFrame, plan *planpb.PlanNode, searchedSegments []segments.Segment, searchReq *segcore.SearchRequest) error {
+	if len(segDFs) != len(searchedSegments) {
+		return merr.WrapErrServiceInternal(fmt.Sprintf("boost_score: DataFrame count %d does not match segment count %d", len(segDFs), len(searchedSegments)))
+	}
 	if plan == nil || len(plan.GetScorers()) == 0 {
 		return nil
 	}
@@ -208,55 +193,8 @@ func (t *SearchTask) applyBoostScores(segDFs []*chain.DataFrame, searchedSegment
 		return err
 	}

-	boostedDFs := make([]*chain.DataFrame, len(segDFs))
 	scoreFunc := segments.AsyncComputeScorerScoresOnChunkedOffsets
-	boostOneSegment := func(ctx context.Context, i int) error {
-		df := segDFs[i]
-		segment := searchedSegments[i]
-		if df == nil {
-			return merr.WrapErrServiceInternal(fmt.Sprintf("boost_score: DataFrame %d is nil", i))
-		}
-
-		boostChain, err := buildBoostScoreChain(df, segment, searchReq, scorers, scoreFunc, functionMode, boostMode)
-		if err != nil {
-			return err
-		}
-
-		boosted, err := boostChain.ExecuteWithContext(ctx, df)
-		if err != nil {
-			return err
-		}
-
-		boostedDFs[i] = boosted
-		return nil
-	}
-
-	if len(segDFs) == 1 {
-		if err := boostOneSegment(t.ctx, 0); err != nil {
-			return err
-		}
-	} else {
-		errGroup, groupCtx := errgroup.WithContext(t.ctx)
-		for i := range segDFs {
-			idx := i
-			errGroup.Go(func() error {
-				return boostOneSegment(groupCtx, idx)
-			})
-		}
-		if err := errGroup.Wait(); err != nil {
-			for _, boosted := range boostedDFs {
-				if boosted != nil {
-					boosted.Release()
-				}
-			}
-			return err
-		}
-	}
-
-	for i, boosted := range boostedDFs {
-		segDFs[i].Release()
-		segDFs[i] = boosted
-	}
-
-	return nil
+	return executeL0RerankChains(t.ctx, segDFs, func(_ context.Context, i int, df *chain.DataFrame) (*chain.FuncChain, error) {
+		return buildBoostScoreChain(df, searchedSegments[i], searchReq, scorers, scoreFunc, functionMode, boostMode)
+	}, "boost_score")
 }
@@ -43,26 +43,8 @@ import (
 )

 func TestBoostScoreColumn(t *testing.T) {
-	require.Equal(t, "$boost_score_0", boostScoreColumn(0))
-	require.Equal(t, "$boost_score_3", boostScoreColumn(3))
-}
-
-func TestBoostReduceColumnsKeepsReduceColumnsOnly(t *testing.T) {
-	groupCol := groupByColumnName(100)
-	cols := []string{
-		types.IDFieldName,
-		boostScoreColumn(0),
-		types.ScoreFieldName,
-		functionScoreColumn,
-		types.SegOffsetFieldName,
-		elementIndicesCol,
-		groupCol,
-		"user_field",
-	}
-
-	require.Equal(t,
-		[]string{types.IDFieldName, types.ScoreFieldName, types.SegOffsetFieldName, elementIndicesCol, groupCol},
-		boostReduceColumns(cols))
+	require.Equal(t, "boost_score_0", boostScoreColumn(0))
+	require.Equal(t, "boost_score_3", boostScoreColumn(3))
 }

 func withBoostScoreCheckedAllocator(t *testing.T) {
@@ -108,6 +90,36 @@ func makeBoostScoreTestDF(t *testing.T, ids []int64, scores []float32, offsets [
 	return builder.Build()
 }

+func addBoostScorePruningColumns(t *testing.T, df *chain.DataFrame) (*chain.DataFrame, string) {
+	builder := chain.NewDataFrameBuilder()
+	builder.SetChunkSizes(df.ChunkSizes())
+	require.NoError(t, builder.AddColumnFrom(df, types.IDFieldName))
+	require.NoError(t, builder.AddColumnFrom(df, types.ScoreFieldName))
+	require.NoError(t, builder.AddColumnFrom(df, types.SegOffsetFieldName))
+
+	elementIndicesBuilder := array.NewInt32Builder(defaultAllocator)
+	elementIndicesBuilder.AppendValues([]int32{0, 1, 2}, nil)
+	elementIndicesArr := elementIndicesBuilder.NewArray()
+	elementIndicesBuilder.Release()
+	require.NoError(t, builder.AddColumnFromChunks(elementIndicesCol, []arrow.Array{elementIndicesArr}))
+
+	groupByCol := groupByColumnName(100)
+	groupByBuilder := array.NewInt64Builder(defaultAllocator)
+	groupByBuilder.AppendValues([]int64{1000, 2000, 3000}, nil)
+	groupByArr := groupByBuilder.NewArray()
+	groupByBuilder.Release()
+	require.NoError(t, builder.AddColumnFromChunks(groupByCol, []arrow.Array{groupByArr}))
+
+	userFieldBuilder := array.NewInt64Builder(defaultAllocator)
+	userFieldBuilder.AppendValues([]int64{7, 8, 9}, nil)
+	userFieldArr := userFieldBuilder.NewArray()
+	userFieldBuilder.Release()
+	require.NoError(t, builder.AddColumnFromChunks("user_field", []arrow.Array{userFieldArr}))
+
+	df.Release()
+	return builder.Build(), groupByCol
+}
+
 func makeBoostScoreTestTask(t *testing.T, plan *planpb.PlanNode) *SearchTask {
 	if plan.GetVectorAnns().GetQueryInfo().GetMetricType() == "" {
 		plan.Node = &planpb.PlanNode_VectorAnns{
@@ -207,7 +219,7 @@ func TestBuildBoostScoreChainSkipsFunctionCombineForSingleScorer(t *testing.T) {
 	)
 	require.NoError(t, err)

-	result, err := boostChain.Execute(df)
+	result, err := boostChain.ExecuteWithOptions(context.Background(), chain.ExecuteOptions{EnableColumnPruning: true}, df)
 	require.NoError(t, err)
 	defer result.Release()

@@ -241,7 +253,7 @@ func TestBuildBoostScoreChainCombinesMultipleScorers(t *testing.T) {
 	)
 	require.NoError(t, err)

-	result, err := boostChain.Execute(df)
+	result, err := boostChain.ExecuteWithOptions(context.Background(), chain.ExecuteOptions{EnableColumnPruning: true}, df)
 	require.NoError(t, err)
 	defer result.Release()

@@ -252,7 +264,7 @@ func TestBuildBoostScoreChainCombinesMultipleScorers(t *testing.T) {
 	require.InDelta(t, 2.5, scores.Value(0), 1e-6)
 }

-func TestApplyBoostScoresSingleScorerCombinesAndSorts(t *testing.T) {
+func TestApplyBoostScoresPrunesTempsAndPreservesReduceSystemColumns(t *testing.T) {
 	withBoostScoreCheckedAllocator(t)

 	oldFactory := boostScoreRunnerFactory
@@ -268,6 +280,7 @@ func TestApplyBoostScoresSingleScorerCombinesAndSorts(t *testing.T) {
 		[]int64{10, 20, 30},
 		[]int64{3},
 	)
+	df, groupByCol := addBoostScorePruningColumns(t, df)
 	segDFs := []*chain.DataFrame{df}

 	task := makeBoostScoreTestTask(t, &planpb.PlanNode{
@@ -283,6 +296,14 @@ func TestApplyBoostScoresSingleScorerCombinesAndSorts(t *testing.T) {
 	result := segDFs[0]
 	ids := result.Column(types.IDFieldName).Chunk(0).(*array.Int64)
 	scores := result.Column(types.ScoreFieldName).Chunk(0).(*array.Float32)
+	require.False(t, result.HasColumn(boostScoreColumn(0)))
+	require.False(t, result.HasColumn(functionScoreColumn))
+	require.False(t, result.HasColumn("user_field"))
+	require.True(t, result.HasColumn(types.IDFieldName))
+	require.True(t, result.HasColumn(types.ScoreFieldName))
+	require.True(t, result.HasColumn(types.SegOffsetFieldName))
+	require.True(t, result.HasColumn(elementIndicesCol))
+	require.True(t, result.HasColumn(groupByCol))
 	require.Equal(t, int64(2), ids.Value(0))
 	require.InDelta(t, 2.0, scores.Value(0), 1e-6)
 	require.Equal(t, int64(3), ids.Value(1))
@@ -0,0 +1,264 @@
+// 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 tasks
+
+import (
+	"context"
+	"fmt"
+
+	"golang.org/x/sync/errgroup"
+
+	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
+	"github.com/milvus-io/milvus/internal/querynodev2/segments"
+	"github.com/milvus-io/milvus/internal/util/function/chain"
+	chaintypes "github.com/milvus-io/milvus/internal/util/function/chain/types"
+	"github.com/milvus-io/milvus/internal/util/segcore"
+	"github.com/milvus-io/milvus/pkg/v3/proto/planpb"
+	"github.com/milvus-io/milvus/pkg/v3/util/merr"
+	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
+)
+
+type preparedQueryNodeFunctionChains struct {
+	plan          *planpb.PlanNode
+	l0Chains      []*chain.ChainRepr
+	extraFieldIDs []int64
+}
+
+func prepareQueryNodeFunctionChains(serializedPlan []byte, schema *schemapb.CollectionSchema) (*preparedQueryNodeFunctionChains, error) {
+	plan, err := extractPlanWithScorers(serializedPlan)
+	if err != nil {
+		return nil, merr.WrapErrServiceInternalErr(err, "querynode function chain: failed to parse search plan")
+	}
+	return prepareQueryNodeFunctionChainsFromPlan(plan, schema)
+}
+
+func prepareQueryNodeFunctionChainsFromPlan(plan *planpb.PlanNode, schema *schemapb.CollectionSchema) (*preparedQueryNodeFunctionChains, error) {
+	prepared := &preparedQueryNodeFunctionChains{plan: plan}
+	if plan == nil || len(plan.GetQuerynodeFunctionChains()) == 0 {
+		return prepared, nil
+	}
+	if len(plan.GetScorers()) > 0 {
+		return nil, merr.WrapErrParameterInvalidMsg("boost score and L0 rerank function chain cannot be used together")
+	}
+
+	schemaHelper, err := typeutil.CreateSchemaHelper(schema)
+	if err != nil {
+		return nil, merr.WrapErrServiceInternalErr(err, "querynode function chain: failed to create schema helper")
+	}
+
+	seenStages := make(map[schemapb.FunctionChainStage]struct{}, len(plan.GetQuerynodeFunctionChains()))
+	seenInputFields := make(map[string]struct{})
+	for i, chainPB := range plan.GetQuerynodeFunctionChains() {
+		if chainPB == nil {
+			return nil, merr.WrapErrParameterInvalidMsg("querynode function chain[%d] is nil", i)
+		}
+		stage := chainPB.GetStage()
+		if _, ok := seenStages[stage]; ok {
+			return nil, merr.WrapErrParameterInvalidMsg("querynode function chain stage %s appears more than once", stage.String())
+		}
+		seenStages[stage] = struct{}{}
+
+		if stage != schemapb.FunctionChainStage_FunctionChainStageL0Rerank {
+			return nil, merr.WrapErrParameterInvalidMsg("querynode function chain[%d] stage %s is not supported", i, stage.String())
+		}
+		if len(chainPB.GetOps()) == 0 {
+			return nil, merr.WrapErrParameterInvalidMsg("querynode function chain[%d] must contain at least one op", i)
+		}
+
+		repr, err := chain.ProtoChainToRepr(chainPB)
+		if err != nil {
+			return nil, merr.Wrapf(err, "querynode function chain[%d]", i)
+		}
+		if err := validateL0FunctionChainOps(repr); err != nil {
+			return nil, merr.Wrapf(err, "querynode function chain[%d]", i)
+		}
+		if err := validateL0FunctionChainSystemOutputs(repr); err != nil {
+			return nil, merr.Wrapf(err, "querynode function chain[%d]", i)
+		}
+
+		extraFieldIDs, err := planL0FunctionChainInputs(repr, schemaHelper, seenInputFields)
+		if err != nil {
+			return nil, merr.Wrapf(err, "querynode function chain[%d]", i)
+		}
+		prepared.extraFieldIDs = append(prepared.extraFieldIDs, extraFieldIDs...)
+		prepared.l0Chains = append(prepared.l0Chains, repr)
+	}
+	return prepared, nil
+}
+
+type l0RerankChainBuilder func(context.Context, int, *chain.DataFrame) (*chain.FuncChain, error)
+
+func appendL0RerankReduceContract(fc *chain.FuncChain) *chain.FuncChain {
+	fc.Sort(chaintypes.ScoreFieldName, true, chaintypes.IDFieldName)
+	return fc
+}
+
+func executeL0RerankChains(ctx context.Context, segDFs []*chain.DataFrame, buildChain l0RerankChainBuilder, errPrefix string) error {
+	rerankedDFs := make([]*chain.DataFrame, len(segDFs))
+	executeOneSegment := func(ctx context.Context, i int) error {
+		df := segDFs[i]
+		if df == nil {
+			return merr.WrapErrServiceInternal(fmt.Sprintf("%s: DataFrame %d is nil", errPrefix, i))
+		}
+
+		fc, err := buildChain(ctx, i, df)
+		if err != nil {
+			return err
+		}
+		if fc == nil {
+			return merr.WrapErrServiceInternal(fmt.Sprintf("%s: function chain %d is nil", errPrefix, i))
+		}
+
+		reranked, err := fc.ExecuteWithOptions(ctx, chain.ExecuteOptions{EnableColumnPruning: true}, df)
+		if err != nil {
+			return err
+		}
+		rerankedDFs[i] = reranked
+		return nil
+	}
+
+	if len(segDFs) == 1 {
+		if err := executeOneSegment(ctx, 0); err != nil {
+			return err
+		}
+	} else {
+		errGroup, groupCtx := errgroup.WithContext(ctx)
+		for i := range segDFs {
+			idx := i
+			errGroup.Go(func() error {
+				return executeOneSegment(groupCtx, idx)
+			})
+		}
+		if err := errGroup.Wait(); err != nil {
+			for _, reranked := range rerankedDFs {
+				if reranked != nil {
+					reranked.Release()
+				}
+			}
+			return err
+		}
+	}
+
+	for i, reranked := range rerankedDFs {
+		segDFs[i].Release()
+		segDFs[i] = reranked
+	}
+	return nil
+}
+
+func (t *SearchTask) applyL0Rerank(segDFs []*chain.DataFrame, prepared *preparedQueryNodeFunctionChains, searchedSegments []segments.Segment, searchReq *segcore.SearchRequest) error {
+	if prepared == nil {
+		return merr.WrapErrServiceInternalMsg("l0_rerank: prepared querynode function chains is nil")
+	}
+	if len(prepared.l0Chains) > 0 {
+		return t.applyPublicL0Rerank(segDFs, prepared)
+	}
+	return t.applyBoostScoresWithPlan(segDFs, prepared.plan, searchedSegments, searchReq)
+}
+
+func (t *SearchTask) applyPublicL0Rerank(segDFs []*chain.DataFrame, prepared *preparedQueryNodeFunctionChains) error {
+	if len(segDFs) == 0 {
+		return nil
+	}
+	if segDFs[0] == nil {
+		return merr.WrapErrServiceInternal("l0_rerank: DataFrame 0 is nil")
+	}
+	if len(prepared.l0Chains) != 1 {
+		return merr.WrapErrServiceInternal(fmt.Sprintf("l0_rerank: expected one L0 function chain, got %d", len(prepared.l0Chains)))
+	}
+
+	repr := prepared.l0Chains[0]
+	// Public L0 avoids reparsing proto by reusing the prepared ChainRepr, but builds
+	// a fresh FuncChain for each segment so operator/function execution state is not
+	// shared across concurrent per-segment execution.
+	return executeL0RerankChains(t.ctx, segDFs, func(context.Context, int, *chain.DataFrame) (*chain.FuncChain, error) {
+		fc, err := chain.FuncChainFromReprWithContext(repr, defaultAllocator, chaintypes.FunctionBuildContext{})
+		if err != nil {
+			return nil, err
+		}
+		return appendL0RerankReduceContract(fc), nil
+	}, "l0_rerank")
+}
+
+func validateL0FunctionChainOps(repr *chain.ChainRepr) error {
+	if repr == nil {
+		return merr.WrapErrParameterInvalidMsg("function chain repr is nil")
+	}
+	for opIdx, op := range repr.Operators {
+		if op.Type != chaintypes.OpTypeMap {
+			return merr.WrapErrParameterInvalidMsg("op[%d] type %q is not supported by L0 rerank function chain", opIdx, op.Type)
+		}
+	}
+	return nil
+}
+
+func validateL0FunctionChainSystemOutputs(repr *chain.ChainRepr) error {
+	if repr == nil {
+		return merr.WrapErrParameterInvalidMsg("function chain repr is nil")
+	}
+	for opIdx, op := range repr.Info.Ops {
+		for _, output := range op.WriteNames {
+			if !chain.IsFunctionChainSystemName(output) {
+				continue
+			}
+			if output != chaintypes.ScoreFieldName {
+				return merr.WrapErrParameterInvalidMsg("op[%d] system output %q is not writable by L0 rerank function chain", opIdx, output)
+			}
+		}
+	}
+	return nil
+}
+
+func planL0FunctionChainInputs(repr *chain.ChainRepr, schemaHelper *typeutil.SchemaHelper, seenInputFields map[string]struct{}) ([]int64, error) {
+	if repr == nil {
+		return nil, merr.WrapErrParameterInvalidMsg("function chain repr is nil")
+	}
+
+	inputFieldIDs := make([]int64, 0)
+	for _, input := range repr.Info.RequiredInputs {
+		if chain.IsFunctionChainSystemName(input) {
+			if !isReadableL0SystemInput(input) {
+				return nil, merr.WrapErrParameterInvalidMsg("system input %q is not readable by L0 rerank function chain", input)
+			}
+			continue
+		}
+		if _, ok := seenInputFields[input]; ok {
+			continue
+		}
+
+		field, err := schemaHelper.GetFieldFromName(input)
+		if err != nil {
+			return nil, merr.WrapErrParameterInvalidMsg("function chain input %q is neither a previous output nor a collection field", input)
+		}
+		if _, err := chain.ToArrowType(field.GetDataType()); err != nil {
+			return nil, merr.WrapErrParameterInvalidMsg("function chain input %q has unsupported field type %s", input, field.GetDataType().String())
+		}
+
+		seenInputFields[input] = struct{}{}
+		inputFieldIDs = append(inputFieldIDs, field.GetFieldID())
+	}
+	return inputFieldIDs, nil
+}
+
+func isReadableL0SystemInput(input string) bool {
+	switch input {
+	case chaintypes.IDFieldName, chaintypes.ScoreFieldName:
+		return true
+	default:
+		return false
+	}
+}
@@ -0,0 +1,296 @@
+// 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 tasks
+
+import (
+	"testing"
+
+	"github.com/apache/arrow/go/v17/arrow"
+	"github.com/apache/arrow/go/v17/arrow/array"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
+	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
+	"github.com/milvus-io/milvus/internal/util/function/chain"
+	chainexpr "github.com/milvus-io/milvus/internal/util/function/chain/expr"
+	"github.com/milvus-io/milvus/internal/util/function/chain/types"
+	"github.com/milvus-io/milvus/pkg/v3/proto/planpb"
+)
+
+func TestPrepareQueryNodeFunctionChainsFromPlan(t *testing.T) {
+	schema := &schemapb.CollectionSchema{
+		Fields: []*schemapb.FieldSchema{
+			{FieldID: 100, Name: "pk", DataType: schemapb.DataType_Int64, IsPrimaryKey: true},
+			{FieldID: 101, Name: "ts", DataType: schemapb.DataType_Int64},
+			{FieldID: 102, Name: "tag", DataType: schemapb.DataType_VarChar},
+			{FieldID: 103, Name: "vec", DataType: schemapb.DataType_FloatVector, TypeParams: []*commonpb.KeyValuePair{{Key: "dim", Value: "4"}}},
+		},
+	}
+
+	t.Run("empty plan", func(t *testing.T) {
+		prepared, err := prepareQueryNodeFunctionChainsFromPlan(nil, schema)
+		require.NoError(t, err)
+		require.NotNil(t, prepared)
+		assert.Empty(t, prepared.l0Chains)
+		assert.Empty(t, prepared.extraFieldIDs)
+	})
+
+	t.Run("l0 chain derives schema input field ids", func(t *testing.T) {
+		plan := &planpb.PlanNode{
+			QuerynodeFunctionChains: []*schemapb.FunctionChain{
+				l0FunctionChainForTest(
+					mapOpForTest("score1", "expr", columnArgForTest("ts"), columnArgForTest(types.ScoreFieldName)),
+					mapOpForTest(types.ScoreFieldName, "expr", columnArgForTest("score1"), columnArgForTest("tag")),
+				),
+			},
+		}
+
+		prepared, err := prepareQueryNodeFunctionChainsFromPlan(plan, schema)
+		require.NoError(t, err)
+		require.Len(t, prepared.l0Chains, 1)
+		assert.Equal(t, []int64{101, 102}, prepared.extraFieldIDs)
+	})
+
+	t.Run("readable system inputs do not become extra fields", func(t *testing.T) {
+		plan := &planpb.PlanNode{
+			QuerynodeFunctionChains: []*schemapb.FunctionChain{
+				l0FunctionChainForTest(mapOpForTest(types.ScoreFieldName, "expr", columnArgForTest(types.ScoreFieldName), columnArgForTest(types.IDFieldName))),
+			},
+		}
+
+		prepared, err := prepareQueryNodeFunctionChainsFromPlan(plan, schema)
+		require.NoError(t, err)
+		assert.Empty(t, prepared.extraFieldIDs)
+	})
+
+	t.Run("internal system input is not readable", func(t *testing.T) {
+		plan := &planpb.PlanNode{
+			QuerynodeFunctionChains: []*schemapb.FunctionChain{
+				l0FunctionChainForTest(mapOpForTest(types.ScoreFieldName, "expr", columnArgForTest(types.SegOffsetFieldName))),
+			},
+		}
+
+		_, err := prepareQueryNodeFunctionChainsFromPlan(plan, schema)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "system input \"$seg_offset\" is not readable")
+	})
+
+	t.Run("unknown system input is not readable", func(t *testing.T) {
+		plan := &planpb.PlanNode{
+			QuerynodeFunctionChains: []*schemapb.FunctionChain{
+				l0FunctionChainForTest(mapOpForTest(types.ScoreFieldName, "expr", columnArgForTest("$unknown"))),
+			},
+		}
+
+		_, err := prepareQueryNodeFunctionChainsFromPlan(plan, schema)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "system input \"$unknown\" is not readable")
+	})
+
+	t.Run("duplicate inputs are planned once", func(t *testing.T) {
+		plan := &planpb.PlanNode{
+			QuerynodeFunctionChains: []*schemapb.FunctionChain{
+				l0FunctionChainForTest(
+					mapOpForTest("score1", "expr", columnArgForTest("ts")),
+					mapOpForTest(types.ScoreFieldName, "expr", columnArgForTest("ts"), columnArgForTest("score1")),
+				),
+			},
+		}
+
+		prepared, err := prepareQueryNodeFunctionChainsFromPlan(plan, schema)
+		require.NoError(t, err)
+		assert.Equal(t, []int64{101}, prepared.extraFieldIDs)
+	})
+
+	t.Run("boost score and l0 are mutually exclusive", func(t *testing.T) {
+		plan := &planpb.PlanNode{
+			Scorers: []*planpb.ScoreFunction{{}},
+			QuerynodeFunctionChains: []*schemapb.FunctionChain{
+				l0FunctionChainForTest(mapOpForTest(types.ScoreFieldName, "expr", columnArgForTest(types.ScoreFieldName))),
+			},
+		}
+
+		_, err := prepareQueryNodeFunctionChainsFromPlan(plan, schema)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "boost score and L0 rerank function chain cannot be used together")
+	})
+
+	t.Run("l1 is unsupported", func(t *testing.T) {
+		plan := &planpb.PlanNode{QuerynodeFunctionChains: []*schemapb.FunctionChain{{
+			Stage: schemapb.FunctionChainStage_FunctionChainStageL1Rerank,
+		}}}
+
+		_, err := prepareQueryNodeFunctionChainsFromPlan(plan, schema)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "is not supported")
+	})
+
+	t.Run("empty l0 chain", func(t *testing.T) {
+		plan := &planpb.PlanNode{QuerynodeFunctionChains: []*schemapb.FunctionChain{l0FunctionChainForTest()}}
+
+		_, err := prepareQueryNodeFunctionChainsFromPlan(plan, schema)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "must contain at least one op")
+	})
+
+	t.Run("only map op is supported", func(t *testing.T) {
+		plan := &planpb.PlanNode{QuerynodeFunctionChains: []*schemapb.FunctionChain{
+			l0FunctionChainForTest(&schemapb.FunctionChainOp{Op: types.OpTypeLimit}),
+		}}
+
+		_, err := prepareQueryNodeFunctionChainsFromPlan(plan, schema)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "type \"limit\" is not supported by L0 rerank function chain")
+	})
+
+	t.Run("only score is writable system output", func(t *testing.T) {
+		plan := &planpb.PlanNode{QuerynodeFunctionChains: []*schemapb.FunctionChain{
+			l0FunctionChainForTest(mapOpForTest(types.IDFieldName, "expr", columnArgForTest(types.ScoreFieldName))),
+		}}
+
+		_, err := prepareQueryNodeFunctionChainsFromPlan(plan, schema)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "system output \"$id\" is not writable")
+	})
+
+	t.Run("unknown input field", func(t *testing.T) {
+		plan := &planpb.PlanNode{QuerynodeFunctionChains: []*schemapb.FunctionChain{
+			l0FunctionChainForTest(mapOpForTest(types.ScoreFieldName, "expr", columnArgForTest("unknown"))),
+		}}
+
+		_, err := prepareQueryNodeFunctionChainsFromPlan(plan, schema)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "unknown")
+		assert.Contains(t, err.Error(), "neither a previous output nor a collection field")
+	})
+
+	t.Run("unsupported input field type", func(t *testing.T) {
+		plan := &planpb.PlanNode{QuerynodeFunctionChains: []*schemapb.FunctionChain{
+			l0FunctionChainForTest(mapOpForTest(types.ScoreFieldName, "expr", columnArgForTest("vec"))),
+		}}
+
+		_, err := prepareQueryNodeFunctionChainsFromPlan(plan, schema)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "unsupported field type")
+	})
+}
+
+func TestApplyL0RerankRejectsNilPreparedChains(t *testing.T) {
+	task := &SearchTask{ctx: t.Context()}
+
+	err := task.applyL0Rerank(nil, nil, nil, nil)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "prepared querynode function chains is nil")
+}
+
+func TestApplyPublicL0RerankPrunesInputsAndPreservesReduceSystemColumns(t *testing.T) {
+	withBoostScoreCheckedAllocator(t)
+
+	df := makeBoostScoreTestDF(t,
+		[]int64{1, 2, 3},
+		[]float32{0.5, 0.2, 0.9},
+		[]int64{10, 20, 30},
+		[]int64{3},
+	)
+	builder := chain.NewDataFrameBuilder()
+	builder.SetChunkSizes(df.ChunkSizes())
+	require.NoError(t, builder.AddColumnFrom(df, types.IDFieldName))
+	require.NoError(t, builder.AddColumnFrom(df, types.ScoreFieldName))
+	require.NoError(t, builder.AddColumnFrom(df, types.SegOffsetFieldName))
+	elementIndicesBuilder := array.NewInt32Builder(defaultAllocator)
+	elementIndicesBuilder.AppendValues([]int32{0, 1, 2}, nil)
+	elementIndicesArr := elementIndicesBuilder.NewArray()
+	elementIndicesBuilder.Release()
+	require.NoError(t, builder.AddColumnFromChunks(elementIndicesCol, []arrow.Array{elementIndicesArr}))
+	groupByCol := groupByColumnName(100)
+	groupByBuilder := array.NewInt64Builder(defaultAllocator)
+	groupByBuilder.AppendValues([]int64{1000, 2000, 3000}, nil)
+	groupByArr := groupByBuilder.NewArray()
+	groupByBuilder.Release()
+	require.NoError(t, builder.AddColumnFromChunks(groupByCol, []arrow.Array{groupByArr}))
+	tsBuilder := array.NewFloat32Builder(defaultAllocator)
+	tsBuilder.AppendValues([]float32{0.1, 3.0, 0.1}, nil)
+	tsArr := tsBuilder.NewArray()
+	tsBuilder.Release()
+	require.NoError(t, builder.AddColumnFromChunks("ts", []arrow.Array{tsArr}))
+	df.Release()
+	df = builder.Build()
+	segDFs := []*chain.DataFrame{df}
+
+	repr, err := chain.ProtoChainToRepr(l0FunctionChainForTest(mapOpWithParamsForTest(
+		types.ScoreFieldName,
+		chainexpr.NumCombineFuncName,
+		map[string]*schemapb.FunctionParamValue{
+			types.NumCombineParamMode: stringParamForTest(types.NumCombineModeSum),
+		},
+		columnArgForTest(types.ScoreFieldName),
+		columnArgForTest("ts"),
+	)))
+	require.NoError(t, err)
+	task := &SearchTask{ctx: t.Context()}
+
+	require.NoError(t, task.applyPublicL0Rerank(segDFs, &preparedQueryNodeFunctionChains{l0Chains: []*chain.ChainRepr{repr}}))
+	defer segDFs[0].Release()
+
+	result := segDFs[0]
+	ids := result.Column(types.IDFieldName).Chunk(0).(*array.Int64)
+	scores := result.Column(types.ScoreFieldName).Chunk(0).(*array.Float32)
+	require.False(t, result.HasColumn("ts"))
+	require.True(t, result.HasColumn(types.IDFieldName))
+	require.True(t, result.HasColumn(types.ScoreFieldName))
+	require.True(t, result.HasColumn(types.SegOffsetFieldName))
+	require.True(t, result.HasColumn(elementIndicesCol))
+	require.True(t, result.HasColumn(groupByCol))
+	require.Equal(t, int64(2), ids.Value(0))
+	require.InDelta(t, 3.2, scores.Value(0), 1e-6)
+	require.Equal(t, int64(3), ids.Value(1))
+	require.InDelta(t, 1.0, scores.Value(1), 1e-6)
+	require.Equal(t, int64(1), ids.Value(2))
+	require.InDelta(t, 0.6, scores.Value(2), 1e-6)
+}
+
+func l0FunctionChainForTest(ops ...*schemapb.FunctionChainOp) *schemapb.FunctionChain {
+	return &schemapb.FunctionChain{
+		Stage: schemapb.FunctionChainStage_FunctionChainStageL0Rerank,
+		Ops:   ops,
+	}
+}
+
+func mapOpForTest(output string, exprName string, args ...*schemapb.FunctionChainExprArg) *schemapb.FunctionChainOp {
+	return mapOpWithParamsForTest(output, exprName, map[string]*schemapb.FunctionParamValue{}, args...)
+}
+
+func mapOpWithParamsForTest(output string, exprName string, params map[string]*schemapb.FunctionParamValue, args ...*schemapb.FunctionChainExprArg) *schemapb.FunctionChainOp {
+	return &schemapb.FunctionChainOp{
+		Op:      types.OpTypeMap,
+		Outputs: []string{output},
+		Expr: &schemapb.FunctionChainExpr{
+			Name:   exprName,
+			Args:   args,
+			Params: params,
+		},
+	}
+}
+
+func columnArgForTest(name string) *schemapb.FunctionChainExprArg {
+	return &schemapb.FunctionChainExprArg{Arg: &schemapb.FunctionChainExprArg_Column{Column: &schemapb.FunctionChainColumnArg{Name: name}}}
+}
+
+func stringParamForTest(value string) *schemapb.FunctionParamValue {
+	return &schemapb.FunctionParamValue{Value: &schemapb.FunctionParamValue_StringValue{StringValue: value}}
+}
@@ -276,9 +276,13 @@ func (t *SearchTask) Execute() error {
 		return err
 	}

+	preparedChains, err := prepareQueryNodeFunctionChains(req.GetReq().GetSerializedExprPlan(), t.collection.Schema())
+	if err != nil {
+		return err
+	}
+
 	// Export per-segment results as Arrow DataFrames
-	// TODO: extract extra field IDs from L0 rerank scorer filters when rerank is configured
-	segDFs, err := t.exportSearchResultsAsArrow(results, searchReq.Plan(), nil)
+	segDFs, err := t.exportSearchResultsAsArrow(results, searchReq.Plan(), preparedChains.extraFieldIDs)
 	if err != nil {
 		return err
 	}
@@ -290,7 +294,7 @@ func (t *SearchTask) Execute() error {
 		}
 	}()

-	if err := t.applyBoostScores(segDFs, searchedSegments, searchReq); err != nil {
+	if err := t.applyL0Rerank(segDFs, preparedChains, searchedSegments, searchReq); err != nil {
 		return err
 	}

@@ -151,11 +151,12 @@ func (s *ChainTestSuite) TestSelectOp() {
 	s.Require().NoError(err)
 	defer result.Release()

-	// Verify only selected columns exist
+	// Verify selected non-system columns exist and unselected non-system columns are removed.
+	// System columns are preserved by SelectOp because downstream reduce paths may need them.
 	s.True(result.HasColumn(types.IDFieldName))
+	s.True(result.HasColumn(types.ScoreFieldName))
 	s.True(result.HasColumn("age"))
 	s.False(result.HasColumn("name"))
-	s.False(result.HasColumn(types.ScoreFieldName))

 	// Verify data integrity
 	s.Equal(df.NumRows(), result.NumRows())
@@ -56,11 +56,19 @@ func (o *SelectOp) Execute(ctx *types.FuncContext, input *DataFrame) (*DataFrame

 	builder.SetChunkSizes(input.chunkSizes)

+	columns := append([]string(nil), o.inputs...)
+	selected := NewColumnSet(columns...)
+	for _, colName := range input.ColumnNames() {
+		if IsFunctionChainSystemName(colName) && !selected.Contains(colName) {
+			columns = append(columns, colName)
+		}
+	}
+
 	// Use ChunkCollector for selected chunks
-	collector := NewChunkCollector(o.inputs, input.NumChunks())
+	collector := NewChunkCollector(columns, input.NumChunks())
 	defer collector.Release()

-	for _, colName := range o.inputs {
+	for _, colName := range columns {
 		col := input.Column(colName)
 		if col == nil {
 			return nil, merr.WrapErrServiceInternalMsg("select_op: column %q not found", colName)
@@ -48,6 +48,10 @@ func TestSelectOpTestSuite(t *testing.T) {
 }

 func (s *SelectOpTestSuite) createThreeColumnDF() *DataFrame {
+	return s.createTestDF(false)
+}
+
+func (s *SelectOpTestSuite) createTestDF(withSystemColumns bool) *DataFrame {
 	builder := NewDataFrameBuilder()
 	builder.SetChunkSizes([]int64{3})

@@ -73,6 +77,15 @@ func (s *SelectOpTestSuite) createThreeColumnDF() *DataFrame {
 	err = builder.AddColumnFromChunks("name", []arrow.Array{nameChunk})
 	s.Require().NoError(err)

+	if withSystemColumns {
+		segOffsetBuilder := array.NewInt64Builder(s.pool)
+		segOffsetBuilder.AppendValues([]int64{10, 20, 30}, nil)
+		segOffsetChunk := segOffsetBuilder.NewArray()
+		segOffsetBuilder.Release()
+		err = builder.AddColumnFromChunks(types.SegOffsetFieldName, []arrow.Array{segOffsetChunk})
+		s.Require().NoError(err)
+	}
+
 	return builder.Build()
 }

@@ -92,6 +105,22 @@ func (s *SelectOpTestSuite) TestSelectSubset() {
 	s.Nil(result.Column("name"))
 }

+func (s *SelectOpTestSuite) TestSelectKeepsSystemColumns() {
+	df := s.createTestDF(true)
+	defer df.Release()
+
+	op := NewSelectOp([]string{"name"})
+	ctx := types.NewFuncContextFull(context.TODO(), s.pool, "rerank")
+	result, err := op.Execute(ctx, df)
+	s.Require().NoError(err)
+	defer result.Release()
+
+	s.NotNil(result.Column("name"))
+	s.NotNil(result.Column(types.IDFieldName))
+	s.NotNil(result.Column(types.ScoreFieldName))
+	s.NotNil(result.Column(types.SegOffsetFieldName))
+}
+
 func (s *SelectOpTestSuite) TestSelectColumnNotFound() {
 	df := s.createThreeColumnDF()
 	defer df.Release()
@@ -21,7 +21,7 @@ require (
 	github.com/jolestar/go-commons-pool/v2 v2.1.2
 	github.com/json-iterator/go v1.1.13-0.20220915233716-71ac16282d12
 	github.com/klauspost/compress v1.18.0
-	github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260615062223-e378dbfa23a2
+	github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260625075625-7262f8042a55
 	github.com/minio/minio-go/v7 v7.0.73
 	github.com/panjf2000/ants/v2 v2.11.3
 	github.com/prometheus/client_golang v1.20.5
@@ -503,10 +503,8 @@ github.com/milvus-io/cgosymbolizer v0.0.0-20250318084424-114f4050c3a6 h1:YHMFI6L
 github.com/milvus-io/cgosymbolizer v0.0.0-20250318084424-114f4050c3a6/go.mod h1:DvXTE/K/RtHehxU8/GtDs4vFtfw64jJ3PaCnFri8CRg=
 github.com/milvus-io/gorocksdb v0.0.0-20220624081344-8c5f4212846b h1:TfeY0NxYxZzUfIfYe5qYDBzt4ZYRqzUjTR6CvUzjat8=
 github.com/milvus-io/gorocksdb v0.0.0-20220624081344-8c5f4212846b/go.mod h1:iwW+9cWfIzzDseEBCCeDSN5SD16Tidvy8cwQ7ZY8Qj4=
-github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260612100615-43795e8f8f6e h1:axPnOhjsWqqxk0dOQRSIU2mdg52Kyl/8E6iDFZy/IpE=
-github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260612100615-43795e8f8f6e/go.mod h1:rbKpv5JToISTKTTLl0duL5r6wbYnjJ9SsD0QgXMzKy0=
-github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260615062223-e378dbfa23a2 h1:D0aOsXFh2wCaTNV4X6A2/lZIgHVFpR9XhKgQuITD+Zg=
-github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260615062223-e378dbfa23a2/go.mod h1:rbKpv5JToISTKTTLl0duL5r6wbYnjJ9SsD0QgXMzKy0=
+github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260625075625-7262f8042a55 h1:U07yIwkWsk7wpGCkWZlY8lSVEFNWKpmm7M+aF5D+wg8=
+github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260625075625-7262f8042a55/go.mod h1:rbKpv5JToISTKTTLl0duL5r6wbYnjJ9SsD0QgXMzKy0=
 github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=
 github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM=
 github.com/minio/minio-go/v7 v7.0.73 h1:qr2vi96Qm7kZ4v7LLebjte+MQh621fFWnv93p12htEo=
@@ -414,4 +414,5 @@ message PlanNode {
  PlanOption plan_options = 7;
  ScoreOption score_option = 8;
  optional string namespace = 9;
+  repeated schema.FunctionChain querynode_function_chains = 10;
 }
@@ -3666,13 +3666,14 @@ type PlanNode struct {
 	//	*PlanNode_VectorAnns
 	//	*PlanNode_Predicates
 	//	*PlanNode_Query
-	Node           isPlanNode_Node  `protobuf_oneof:"node"`
-	OutputFieldIds []int64          `protobuf:"varint,3,rep,packed,name=output_field_ids,json=outputFieldIds,proto3" json:"output_field_ids,omitempty"`
-	DynamicFields  []string         `protobuf:"bytes,5,rep,name=dynamic_fields,json=dynamicFields,proto3" json:"dynamic_fields,omitempty"`
-	Scorers        []*ScoreFunction `protobuf:"bytes,6,rep,name=scorers,proto3" json:"scorers,omitempty"`
-	PlanOptions    *PlanOption      `protobuf:"bytes,7,opt,name=plan_options,json=planOptions,proto3" json:"plan_options,omitempty"`
-	ScoreOption    *ScoreOption     `protobuf:"bytes,8,opt,name=score_option,json=scoreOption,proto3" json:"score_option,omitempty"`
-	Namespace      *string          `protobuf:"bytes,9,opt,name=namespace,proto3,oneof" json:"namespace,omitempty"`
+	Node                    isPlanNode_Node           `protobuf_oneof:"node"`
+	OutputFieldIds          []int64                   `protobuf:"varint,3,rep,packed,name=output_field_ids,json=outputFieldIds,proto3" json:"output_field_ids,omitempty"`
+	DynamicFields           []string                  `protobuf:"bytes,5,rep,name=dynamic_fields,json=dynamicFields,proto3" json:"dynamic_fields,omitempty"`
+	Scorers                 []*ScoreFunction          `protobuf:"bytes,6,rep,name=scorers,proto3" json:"scorers,omitempty"`
+	PlanOptions             *PlanOption               `protobuf:"bytes,7,opt,name=plan_options,json=planOptions,proto3" json:"plan_options,omitempty"`
+	ScoreOption             *ScoreOption              `protobuf:"bytes,8,opt,name=score_option,json=scoreOption,proto3" json:"score_option,omitempty"`
+	Namespace               *string                   `protobuf:"bytes,9,opt,name=namespace,proto3,oneof" json:"namespace,omitempty"`
+	QuerynodeFunctionChains []*schemapb.FunctionChain `protobuf:"bytes,10,rep,name=querynode_function_chains,json=querynodeFunctionChains,proto3" json:"querynode_function_chains,omitempty"`
 }

 func (x *PlanNode) Reset() {
@@ -3777,6 +3778,13 @@ func (x *PlanNode) GetNamespace() string {
 	return ""
 }

+func (x *PlanNode) GetQuerynodeFunctionChains() []*schemapb.FunctionChain {
+	if x != nil {
+		return x.QuerynodeFunctionChains
+	}
+	return nil
+}
+
 type isPlanNode_Node interface {
 	isPlanNode_Node()
 }
@@ -4394,7 +4402,7 @@ var file_plan_proto_rawDesc = []byte{
 	0x0a, 0x0a, 0x50, 0x6c, 0x61, 0x6e, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x2d, 0x0a, 0x13,
 	0x65, 0x78, 0x70, 0x72, 0x5f, 0x75, 0x73, 0x65, 0x5f, 0x6a, 0x73, 0x6f, 0x6e, 0x5f, 0x73, 0x74,
 	0x61, 0x74, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x08, 0x52, 0x10, 0x65, 0x78, 0x70, 0x72, 0x55,
-	0x73, 0x65, 0x4a, 0x73, 0x6f, 0x6e, 0x53, 0x74, 0x61, 0x74, 0x73, 0x22, 0x8c, 0x04, 0x0a, 0x08,
+	0x73, 0x65, 0x4a, 0x73, 0x6f, 0x6e, 0x53, 0x74, 0x61, 0x74, 0x73, 0x22, 0xec, 0x04, 0x0a, 0x08,
 	0x50, 0x6c, 0x61, 0x6e, 0x4e, 0x6f, 0x64, 0x65, 0x12, 0x40, 0x0a, 0x0b, 0x76, 0x65, 0x63, 0x74,
 	0x6f, 0x72, 0x5f, 0x61, 0x6e, 0x6e, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1d, 0x2e,
 	0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x70, 0x6c, 0x61,
@@ -4426,7 +4434,13 @@ var file_plan_proto_rawDesc = []byte{
 	0x74, 0x69, 0x6f, 0x6e, 0x52, 0x0b, 0x73, 0x63, 0x6f, 0x72, 0x65, 0x4f, 0x70, 0x74, 0x69, 0x6f,
 	0x6e, 0x12, 0x21, 0x0a, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x18, 0x09,
 	0x20, 0x01, 0x28, 0x09, 0x48, 0x01, 0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63,
-	0x65, 0x88, 0x01, 0x01, 0x42, 0x06, 0x0a, 0x04, 0x6e, 0x6f, 0x64, 0x65, 0x42, 0x0c, 0x0a, 0x0a,
+	0x65, 0x88, 0x01, 0x01, 0x12, 0x5e, 0x0a, 0x19, 0x71, 0x75, 0x65, 0x72, 0x79, 0x6e, 0x6f, 0x64,
+	0x65, 0x5f, 0x66, 0x75, 0x6e, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x63, 0x68, 0x61, 0x69, 0x6e,
+	0x73, 0x18, 0x0a, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x22, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73,
+	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x2e, 0x46, 0x75,
+	0x6e, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x68, 0x61, 0x69, 0x6e, 0x52, 0x17, 0x71, 0x75, 0x65,
+	0x72, 0x79, 0x6e, 0x6f, 0x64, 0x65, 0x46, 0x75, 0x6e, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x68,
+	0x61, 0x69, 0x6e, 0x73, 0x42, 0x06, 0x0a, 0x04, 0x6e, 0x6f, 0x64, 0x65, 0x42, 0x0c, 0x0a, 0x0a,
 	0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x2a, 0x8e, 0x02, 0x0a, 0x06, 0x4f,
 	0x70, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0b, 0x0a, 0x07, 0x49, 0x6e, 0x76, 0x61, 0x6c, 0x69, 0x64,
 	0x10, 0x00, 0x12, 0x0f, 0x0a, 0x0b, 0x47, 0x72, 0x65, 0x61, 0x74, 0x65, 0x72, 0x54, 0x68, 0x61,
@@ -4562,6 +4576,7 @@ var file_plan_proto_goTypes = []interface{}{
 	(*PlanNode)(nil),                    // 49: milvus.proto.plan.PlanNode
 	(schemapb.DataType)(0),              // 50: milvus.proto.schema.DataType
 	(*commonpb.KeyValuePair)(nil),       // 51: milvus.proto.common.KeyValuePair
+	(*schemapb.FunctionChain)(nil),      // 52: milvus.proto.schema.FunctionChain
 }
 var file_plan_proto_depIdxs = []int32{
 	14, // 0: milvus.proto.plan.GenericValue.array_val:type_name -> milvus.proto.plan.Array
@@ -4659,11 +4674,12 @@ var file_plan_proto_depIdxs = []int32{
 	46, // 92: milvus.proto.plan.PlanNode.scorers:type_name -> milvus.proto.plan.ScoreFunction
 	48, // 93: milvus.proto.plan.PlanNode.plan_options:type_name -> milvus.proto.plan.PlanOption
 	47, // 94: milvus.proto.plan.PlanNode.score_option:type_name -> milvus.proto.plan.ScoreOption
-	95, // [95:95] is the sub-list for method output_type
-	95, // [95:95] is the sub-list for method input_type
-	95, // [95:95] is the sub-list for extension type_name
-	95, // [95:95] is the sub-list for extension extendee
-	0,  // [0:95] is the sub-list for field type_name
+	52, // 95: milvus.proto.plan.PlanNode.querynode_function_chains:type_name -> milvus.proto.schema.FunctionChain
+	96, // [96:96] is the sub-list for method output_type
+	96, // [96:96] is the sub-list for method input_type
+	96, // [96:96] is the sub-list for extension type_name
+	96, // [96:96] is the sub-list for extension extendee
+	0,  // [0:96] is the sub-list for field type_name
 }

 func init() { file_plan_proto_init() }
@@ -5,7 +5,7 @@ go 1.26.4
 require (
 	github.com/apache/arrow/go/v17 v17.0.0
 	github.com/cockroachdb/errors v1.9.1
-	github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260615062223-e378dbfa23a2
+	github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260625075625-7262f8042a55
 	github.com/milvus-io/milvus/client/v2 v2.0.0-20241125024034-0b9edb62a92d
 	github.com/milvus-io/milvus/pkg/v3 v3.0.0-beta
 	github.com/minio/minio-go/v7 v7.0.73
@@ -221,10 +221,8 @@ github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27k
 github.com/mattn/goveralls v0.0.2/go.mod h1:8d1ZMHsd7fW6IRPKQh46F2WRpyib5/X4FOpevwGNQEw=
 github.com/mediocregopher/radix/v3 v3.4.2/go.mod h1:8FL3F6UQRXHXIBSPUs5h0RybMF8i4n7wVopoX3x7Bv8=
 github.com/microcosm-cc/bluemonday v1.0.2/go.mod h1:iVP4YcDBq+n/5fb23BhYFvIMq/leAFZyRl6bYmGDlGc=
-github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260612100615-43795e8f8f6e h1:axPnOhjsWqqxk0dOQRSIU2mdg52Kyl/8E6iDFZy/IpE=
-github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260612100615-43795e8f8f6e/go.mod h1:rbKpv5JToISTKTTLl0duL5r6wbYnjJ9SsD0QgXMzKy0=
-github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260615062223-e378dbfa23a2 h1:D0aOsXFh2wCaTNV4X6A2/lZIgHVFpR9XhKgQuITD+Zg=
-github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260615062223-e378dbfa23a2/go.mod h1:rbKpv5JToISTKTTLl0duL5r6wbYnjJ9SsD0QgXMzKy0=
+github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260625075625-7262f8042a55 h1:U07yIwkWsk7wpGCkWZlY8lSVEFNWKpmm7M+aF5D+wg8=
+github.com/milvus-io/milvus-proto/go-api/v3 v3.0.0-20260625075625-7262f8042a55/go.mod h1:rbKpv5JToISTKTTLl0duL5r6wbYnjJ9SsD0QgXMzKy0=
 github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 h1:AMFGa4R4MiIpspGNG7Z948v4n35fFGB3RR3G/ry4FWs=
 github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8/go.mod h1:mC1jAcsrzbxHt8iiaC+zU4b1ylILSosueou12R++wfY=
 github.com/minio/c2goasm v0.0.0-20190812172519-36a3d3bbc4f3 h1:+n/aFZefKZp7spd8DFdX7uMikMLXX4oubIzJF4kv/wI=
@@ -23,8 +23,8 @@ pytest-sugar==0.9.5
 pytest-random-order

 # pymilvus
-pymilvus==3.1.0rc52
-pymilvus[bulk_writer]==3.1.0rc52
+pymilvus==3.1.0rc54
+pymilvus[bulk_writer]==3.1.0rc54
 # for protobuf
 protobuf>=5.29.5

@@ -0,0 +1,503 @@
+import pytest
+from base.client_v2_base import TestMilvusClientV2Base
+from common import common_func as cf
+from common import common_type as ct
+from common.common_type import CaseLabel, CheckTasks
+from pymilvus import DataType, Function, FunctionChain, FunctionChainStage, FunctionScore, FunctionType
+from pymilvus.function_chain import col, fn
+
+prefix = "function_chain"
+
+
+class TestFunctionChain(TestMilvusClientV2Base):
+    """Test pymilvus FunctionChain SDK integration."""
+
+    dim = 2
+    vector_field = "vector"
+    scalar_field = "ts"
+
+    def _create_function_chain_collection(self, client):
+        collection_name = cf.gen_unique_str(prefix)
+        schema = self.create_schema(client, auto_id=False, enable_dynamic_field=False)[0]
+        schema.add_field("id", DataType.INT64, is_primary=True)
+        schema.add_field(self.scalar_field, DataType.INT64)
+        schema.add_field(self.vector_field, DataType.FLOAT_VECTOR, dim=self.dim)
+
+        index_params = self.prepare_index_params(client)[0]
+        index_params.add_index(field_name=self.vector_field, index_type="FLAT", metric_type="L2")
+        self.create_collection(
+            client,
+            collection_name,
+            schema=schema,
+            index_params=index_params,
+            consistency_level="Strong",
+        )
+
+        rows = [
+            {"id": 1, self.scalar_field: 10, self.vector_field: [0.0, 0.0]},
+            {"id": 2, self.scalar_field: 20, self.vector_field: [0.01, 0.0]},
+            {"id": 3, self.scalar_field: 30, self.vector_field: [0.02, 0.0]},
+        ]
+        self.insert(client, collection_name, rows)
+        self.flush(client, collection_name)
+        self.load_collection(client, collection_name)
+        return collection_name
+
+    def _score_plus_ts_chain(self, stage):
+        chain = FunctionChain(stage, name="score_plus_ts").map(
+            "$score",
+            fn.num_combine(col("$score"), col(self.scalar_field), mode="sum"),
+        )
+        if stage == FunctionChainStage.L2_RERANK:
+            chain.sort(col("$score"), desc=True, tie_break_col=col("$id"))
+        return chain
+
+    def _assert_search_error(self, client, collection_name, function_chains, err_msg, **kwargs):
+        self.search(
+            client,
+            collection_name,
+            data=[[0.0, 0.0]],
+            anns_field=self.vector_field,
+            search_params={"metric_type": "L2"},
+            limit=3,
+            function_chains=function_chains,
+            check_task=CheckTasks.err_res,
+            check_items={ct.err_code: 1100, ct.err_msg: err_msg},
+            **kwargs,
+        )
+
+    @staticmethod
+    def _hit_field(hit, field):
+        if field in hit:
+            return hit[field]
+        return hit.get("entity", {}).get(field)
+
+    @staticmethod
+    def _expected_l0_score(hit):
+        vector = hit.get("entity", {}).get("vector", hit.get("vector"))
+        l2_distance = sum(value * value for value in vector)
+        return TestFunctionChain._hit_field(hit, "ts") - l2_distance
+
+    @pytest.mark.skip(reason="official pymilvus SDK does not support L0 function chain yet")
+    @pytest.mark.tags(CaseLabel.L0)
+    def test_search_with_l0_function_chain_sdk_reranks_by_scalar_field(self):
+        """
+        target: test pymilvus FunctionChain SDK with L0 rerank
+        method: map $score = num_combine($score, ts) at L0 stage
+        expected: search succeeds and result order follows rewritten score
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+
+        res, _ = self.search(
+            client,
+            collection_name,
+            data=[[0.0, 0.0]],
+            anns_field=self.vector_field,
+            search_params={"metric_type": "L2"},
+            limit=3,
+            output_fields=[self.scalar_field, self.vector_field],
+            function_chains=self._score_plus_ts_chain(FunctionChainStage.L0_RERANK),
+        )
+
+        assert [hit["id"] for hit in res[0]] == [3, 2, 1]
+        assert [self._hit_field(hit, self.scalar_field) for hit in res[0]] == [30, 20, 10]
+        expected_scores = [self._expected_l0_score(hit) for hit in res[0]]
+        assert expected_scores == sorted(expected_scores, reverse=True)
+        assert [pytest.approx(abs(hit["distance"]), rel=1e-5) for hit in res[0]] == expected_scores
+
+    @pytest.mark.skip(reason="official pymilvus SDK does not support L0 function chain yet")
+    @pytest.mark.tags(CaseLabel.L0)
+    def test_search_with_l0_function_chain_sdk_uses_hidden_input_field(self):
+        """
+        target: test L0 FunctionChain SDK can use fields that are not returned
+        method: rerank by ts while only requesting primary key output
+        expected: search succeeds, result order follows ts, and ts is not returned
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+
+        res, _ = self.search(
+            client,
+            collection_name,
+            data=[[0.0, 0.0]],
+            anns_field=self.vector_field,
+            search_params={"metric_type": "L2"},
+            limit=3,
+            output_fields=["id"],
+            function_chains=self._score_plus_ts_chain(FunctionChainStage.L0_RERANK),
+        )
+
+        assert [hit["id"] for hit in res[0]] == [3, 2, 1]
+        assert all(self._hit_field(hit, self.scalar_field) is None for hit in res[0])
+
+    @pytest.mark.skip(reason="official pymilvus SDK does not support L0 function chain yet")
+    @pytest.mark.tags(CaseLabel.L0)
+    def test_search_with_l0_function_chain_sdk_can_read_id_system_input(self):
+        """
+        target: test L0 FunctionChain SDK can read public system input $id
+        method: map $score = num_combine($score, $id) at L0 stage
+        expected: search succeeds and result order follows rewritten score
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        chain = FunctionChain(FunctionChainStage.L0_RERANK, name="score_plus_id").map(
+            "$score",
+            fn.num_combine(col("$score"), col("$id"), mode="sum"),
+        )
+
+        res, _ = self.search(
+            client,
+            collection_name,
+            data=[[0.0, 0.0]],
+            anns_field=self.vector_field,
+            search_params={"metric_type": "L2"},
+            limit=3,
+            function_chains=chain,
+        )
+
+        assert [hit["id"] for hit in res[0]] == [3, 2, 1]
+
+    @pytest.mark.skip(reason="official pymilvus SDK does not support L0 function chain yet")
+    @pytest.mark.tags(CaseLabel.L0)
+    def test_search_rejects_l0_function_chain_sort_op(self):
+        """
+        target: test L0 FunctionChain SDK rejects non-map operators
+        method: use sort op at L0 stage
+        expected: request fails because public L0 currently only supports map op
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        chain = FunctionChain(FunctionChainStage.L0_RERANK, name="bad_l0_sort").sort(
+            col("$score"),
+            desc=True,
+            tie_break_col=col("$id"),
+        )
+
+        self._assert_search_error(
+            client, collection_name, chain, 'type "sort" is not supported by L0 rerank function chain'
+        )
+
+    @pytest.mark.skip(reason="official pymilvus SDK does not support L0 function chain yet")
+    @pytest.mark.tags(CaseLabel.L0)
+    def test_search_rejects_l0_function_chain_write_readonly_system_column(self):
+        """
+        target: test L0 FunctionChain SDK rejects writes to read-only system columns
+        method: write map output to $id
+        expected: request fails because only $score is writable in public L0 chains
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        chain = FunctionChain(FunctionChainStage.L0_RERANK, name="bad_l0_write_id").map(
+            "$id",
+            fn.num_combine(col("$score"), col(self.scalar_field), mode="sum"),
+        )
+
+        self._assert_search_error(client, collection_name, chain, 'system output "$id" is not writable')
+
+    @pytest.mark.skip(reason="official pymilvus SDK does not support L0 function chain yet")
+    @pytest.mark.tags(CaseLabel.L0)
+    def test_search_rejects_l0_function_chain_read_internal_system_input(self):
+        """
+        target: test L0 FunctionChain SDK rejects internal system input columns
+        method: read $seg_offset from a map expression
+        expected: request fails because public L0 only exposes $id and $score as readable system inputs
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        chain = FunctionChain(FunctionChainStage.L0_RERANK, name="bad_l0_seg_offset_input").map(
+            "$score",
+            fn.num_combine(col("$seg_offset"), col("$score"), mode="sum"),
+        )
+
+        self._assert_search_error(client, collection_name, chain, 'system input "$seg_offset" is not readable')
+
+    @pytest.mark.skip(reason="official pymilvus SDK does not support L0 function chain yet")
+    @pytest.mark.tags(CaseLabel.L0)
+    def test_search_rejects_l0_function_chain_read_unknown_system_input(self):
+        """
+        target: test L0 FunctionChain SDK rejects unknown system input columns
+        method: read $tmp_score from a map expression before it is produced
+        expected: request fails because users cannot invent new $-prefixed system columns
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        chain = FunctionChain(FunctionChainStage.L0_RERANK, name="bad_l0_unknown_system_input").map(
+            "$score",
+            fn.num_combine(col("$tmp_score"), col("$score"), mode="sum"),
+        )
+
+        self._assert_search_error(client, collection_name, chain, 'system input "$tmp_score" is not readable')
+
+    @pytest.mark.skip(reason="official pymilvus SDK does not support L0 function chain yet")
+    @pytest.mark.tags(CaseLabel.L0)
+    def test_search_rejects_l0_function_chain_reserved_temp_output(self):
+        """
+        target: test L0 FunctionChain SDK rejects user temporary columns in system namespace
+        method: write a map output named $tmp_score
+        expected: request fails because $ prefix is reserved for system columns
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        chain = FunctionChain(FunctionChainStage.L0_RERANK, name="bad_l0_reserved_temp_output").map(
+            "$tmp_score",
+            fn.num_combine(col("$score"), col(self.scalar_field), mode="sum"),
+        )
+
+        self._assert_search_error(client, collection_name, chain, 'system output "$tmp_score" is not writable')
+
+    @pytest.mark.skip(reason="official pymilvus SDK does not support L0 function chain yet")
+    @pytest.mark.tags(CaseLabel.L0)
+    def test_search_rejects_l0_function_chain_with_function_score(self):
+        """
+        target: test search rejects ambiguous L0 rerank APIs
+        method: send boost FunctionScore and L0 function chain together
+        expected: request fails because function_score and function_chains are mutually exclusive
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        function = Function(
+            name="boost_ts",
+            function_type=FunctionType.RERANK,
+            input_field_names=[],
+            output_field_names=[],
+            params={"reranker": "boost", "weight": "1.5"},
+        )
+        function_score = FunctionScore(functions=[function])
+
+        self._assert_search_error(
+            client,
+            collection_name,
+            self._score_plus_ts_chain(FunctionChainStage.L0_RERANK),
+            "function_chains and ranker cannot be used together",
+            ranker=function_score,
+        )
+
+    @pytest.mark.skip(reason="official pymilvus SDK does not support L0 function chain yet")
+    @pytest.mark.tags(CaseLabel.L0)
+    def test_search_rejects_l0_function_chain_with_order_by(self):
+        """
+        target: test search rejects order_by with L0 function rerank
+        method: send order_by_fields and L0 function chain together
+        expected: request fails because they define conflicting sort criteria
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+
+        self._assert_search_error(
+            client,
+            collection_name,
+            self._score_plus_ts_chain(FunctionChainStage.L0_RERANK),
+            "order_by and function rerank cannot be used together",
+            order_by_fields=[{"field": self.scalar_field, "order": "asc"}],
+        )
+
+    @pytest.mark.tags(CaseLabel.L2)
+    def test_search_with_l2_function_chain_sdk_reranks_by_scalar_field(self):
+        """
+        target: test pymilvus FunctionChain SDK with L2 rerank
+        method: map $score = num_combine($score, ts), then sort by $score desc
+        expected: search succeeds and result order follows rewritten score
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+
+        res, _ = self.search(
+            client,
+            collection_name,
+            data=[[0.0, 0.0]],
+            anns_field=self.vector_field,
+            search_params={"metric_type": "L2"},
+            limit=3,
+            output_fields=[self.scalar_field],
+            function_chains=self._score_plus_ts_chain(FunctionChainStage.L2_RERANK),
+        )
+
+        assert [hit["id"] for hit in res[0]] == [3, 2, 1]
+        assert [self._hit_field(hit, self.scalar_field) for hit in res[0]] == [30, 20, 10]
+
+    @pytest.mark.tags(CaseLabel.L2)
+    def test_search_with_l2_function_chain_sdk_uses_hidden_input_field(self):
+        """
+        target: test L2 FunctionChain SDK can use fields that are not returned
+        method: rerank by ts while only requesting primary key output
+        expected: search succeeds, result order follows ts, and ts is not returned
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+
+        res, _ = self.search(
+            client,
+            collection_name,
+            data=[[0.0, 0.0]],
+            anns_field=self.vector_field,
+            search_params={"metric_type": "L2"},
+            limit=3,
+            output_fields=["id"],
+            function_chains=self._score_plus_ts_chain(FunctionChainStage.L2_RERANK),
+        )
+
+        assert [hit["id"] for hit in res[0]] == [3, 2, 1]
+        assert all(self._hit_field(hit, self.scalar_field) is None for hit in res[0])
+
+    @pytest.mark.tags(CaseLabel.L2)
+    def test_search_with_l2_function_chain_sdk_temp_column_not_returned(self):
+        """
+        target: test L2 FunctionChain SDK can use ordinary temporary columns
+        method: write tmp_score, write it back to $score, then sort by $score desc
+        expected: search succeeds, rerank order is correct, and tmp_score is not returned
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        chain = (
+            FunctionChain(FunctionChainStage.L2_RERANK, name="l2_temp_score")
+            .map("tmp_score", fn.num_combine(col("$score"), col(self.scalar_field), mode="sum"))
+            .map("$score", fn.num_combine(col("tmp_score"), col("$score"), mode="sum"))
+            .sort(col("$score"), desc=True, tie_break_col=col("$id"))
+        )
+
+        res, _ = self.search(
+            client,
+            collection_name,
+            data=[[0.0, 0.0]],
+            anns_field=self.vector_field,
+            search_params={"metric_type": "L2"},
+            limit=3,
+            output_fields=[self.scalar_field],
+            function_chains=chain,
+        )
+
+        assert [hit["id"] for hit in res[0]] == [3, 2, 1]
+        assert [self._hit_field(hit, self.scalar_field) for hit in res[0]] == [30, 20, 10]
+        assert all(self._hit_field(hit, "tmp_score") is None for hit in res[0])
+
+    @pytest.mark.tags(CaseLabel.L2)
+    def test_search_with_l2_function_chain_sdk_limit_op(self):
+        """
+        target: test L2 FunctionChain SDK supports limit operator
+        method: request limit=3 and apply function chain limit op with limit=2
+        expected: search succeeds and returns only function-chain-limited results
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        chain = FunctionChain(FunctionChainStage.L2_RERANK, name="l2_limit").limit(2)
+
+        res, _ = self.search(
+            client,
+            collection_name,
+            data=[[0.0, 0.0]],
+            anns_field=self.vector_field,
+            search_params={"metric_type": "L2"},
+            limit=3,
+            function_chains=chain,
+        )
+
+        assert len(res[0]) == 2
+
+    @pytest.mark.tags(CaseLabel.L2)
+    def test_search_rejects_l2_function_chain_write_readonly_system_column(self):
+        """
+        target: test L2 FunctionChain SDK rejects writes to read-only system columns
+        method: write map output to $id
+        expected: request fails because only $score is writable in L2 rerank chains
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        chain = FunctionChain(FunctionChainStage.L2_RERANK, name="bad_l2_write_id").map(
+            "$id",
+            fn.num_combine(col("$score"), col(self.scalar_field), mode="sum"),
+        )
+
+        self._assert_search_error(client, collection_name, chain, 'system output "$id" is not writable')
+
+    @pytest.mark.tags(CaseLabel.L2)
+    def test_search_rejects_l2_function_chain_reserved_temp_output(self):
+        """
+        target: test L2 FunctionChain SDK rejects user temporary columns in system namespace
+        method: write a map output named $tmp_score
+        expected: request fails because $ prefix is reserved for system columns
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        chain = FunctionChain(FunctionChainStage.L2_RERANK, name="bad_l2_reserved_temp_output").map(
+            "$tmp_score",
+            fn.num_combine(col("$score"), col(self.scalar_field), mode="sum"),
+        )
+
+        self._assert_search_error(client, collection_name, chain, 'system output "$tmp_score" is not writable')
+
+    @pytest.mark.tags(CaseLabel.L2)
+    def test_search_rejects_l2_function_chain_read_internal_system_input(self):
+        """
+        target: test L2 FunctionChain SDK rejects internal system input columns
+        method: read $seg_offset from a map expression
+        expected: request fails because L2 only exposes selected system inputs
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        chain = FunctionChain(FunctionChainStage.L2_RERANK, name="bad_l2_seg_offset_input").map(
+            "$score",
+            fn.num_combine(col("$seg_offset"), col("$score"), mode="sum"),
+        )
+
+        self._assert_search_error(client, collection_name, chain, 'system input "$seg_offset" is not supported')
+
+    @pytest.mark.tags(CaseLabel.L2)
+    def test_search_rejects_l2_function_chain_read_unknown_system_input(self):
+        """
+        target: test L2 FunctionChain SDK rejects unknown system input columns
+        method: read $tmp_score from a map expression before it is produced
+        expected: request fails because users cannot invent new $-prefixed system columns
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        chain = FunctionChain(FunctionChainStage.L2_RERANK, name="bad_l2_unknown_system_input").map(
+            "$score",
+            fn.num_combine(col("$tmp_score"), col("$score"), mode="sum"),
+        )
+
+        self._assert_search_error(client, collection_name, chain, 'system input "$tmp_score" is not supported')
+
+    @pytest.mark.tags(CaseLabel.L2)
+    def test_search_rejects_l2_function_chain_with_function_score(self):
+        """
+        target: test search rejects ambiguous L2 rerank APIs
+        method: send boost FunctionScore and L2 function chain together
+        expected: request fails because function chains and ranker are mutually exclusive
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+        function = Function(
+            name="boost_ts",
+            function_type=FunctionType.RERANK,
+            input_field_names=[],
+            output_field_names=[],
+            params={"reranker": "boost", "weight": "1.5"},
+        )
+        function_score = FunctionScore(functions=[function])
+
+        self._assert_search_error(
+            client,
+            collection_name,
+            self._score_plus_ts_chain(FunctionChainStage.L2_RERANK),
+            "function_chains and ranker cannot be used together",
+            ranker=function_score,
+        )
+
+    @pytest.mark.tags(CaseLabel.L2)
+    def test_search_rejects_l2_function_chain_with_order_by(self):
+        """
+        target: test search rejects order_by with L2 function rerank
+        method: send order_by_fields and L2 function chain together
+        expected: request fails because they define conflicting sort criteria
+        """
+        client = self._client()
+        collection_name = self._create_function_chain_collection(client)
+
+        self._assert_search_error(
+            client,
+            collection_name,
+            self._score_plus_ts_chain(FunctionChainStage.L2_RERANK),
+            "order_by and function rerank cannot be used together",
+            order_by_fields=[{"field": self.scalar_field, "order": "asc"}],
+        )
@@ -15,7 +15,7 @@ class TestFunctionChainAPI(TestBase):
    ******************************************************************
    """

-    def _create_function_chain_collection(self):
+    def _create_function_chain_collection(self, data=None):
        name = gen_collection_name(prefix)
        self.name = name
        payload = {
@@ -36,11 +36,12 @@ class TestFunctionChainAPI(TestBase):
        rsp = self.collection_client.collection_create(payload)
        assert rsp["code"] == 0, f"create collection failed: {rsp}"

-        data = [
-            {"id": 1, "ts": 10, "vector": [0.0, 0.0]},
-            {"id": 2, "ts": 20, "vector": [0.01, 0.0]},
-            {"id": 3, "ts": 30, "vector": [0.02, 0.0]},
-        ]
+        if data is None:
+            data = [
+                {"id": 1, "ts": 10, "vector": [0.0, 0.0]},
+                {"id": 2, "ts": 20, "vector": [0.01, 0.0]},
+                {"id": 3, "ts": 30, "vector": [0.02, 0.0]},
+            ]
        rsp = self.vector_client.vector_insert({"collectionName": name, "data": data})
        assert rsp["code"] == 0, f"insert failed: {rsp}"
        assert rsp["data"]["insertCount"] == len(data)
@@ -76,6 +77,79 @@ class TestFunctionChainAPI(TestBase):
            }
        ]

+    def _l0_boost_equivalent_function_chain(self):
+        return [
+            {
+                "name": "l0_boost_ts_flag",
+                "stage": "FunctionChainStageL0Rerank",
+                "ops": [
+                    {
+                        "op": "map",
+                        "outputs": ["$score"],
+                        "expr": {
+                            "name": "num_combine",
+                            "args": [
+                                {"column": "$score"},
+                                {"column": "ts"},
+                            ],
+                            "params": {"mode": "weighted", "weights": [1, 10]},
+                        },
+                    }
+                ],
+            }
+        ]
+
+    def _boost_ts_flag_function_score(self):
+        return {
+            "functions": [
+                {
+                    "name": "boost_ts_flag",
+                    "type": "Rerank",
+                    "inputFieldNames": [],
+                    "params": {
+                        "reranker": "boost",
+                        "filter": "ts > 0",
+                        "weight": 10,
+                    },
+                }
+            ],
+            "params": {"boost_mode": "sum"},
+        }
+
+    def test_search_l0_function_chain_matches_boost_rank(self):
+        """
+        target: test REST v2 L0 functionChains can cover boost rank semantics
+        method: compare L0 chain score + 10 * ts_flag with boost ranker filter ts > 0 and weight 10
+        expected: both requests return the same reranked ids
+        """
+        data = [
+            {"id": 1, "ts": 0, "vector": [0.0, 0.0]},
+            {"id": 2, "ts": 1, "vector": [0.2, 0.0]},
+            {"id": 3, "ts": 0, "vector": [0.1, 0.0]},
+        ]
+        name = self._create_function_chain_collection(data=data)
+
+        base_payload = {
+            "collectionName": name,
+            "data": [[0.0, 0.0]],
+            "annsField": "vector",
+            "limit": 3,
+            "outputFields": ["ts"],
+        }
+        chain_rsp = self.vector_client.vector_search(
+            {**base_payload, "functionChains": self._l0_boost_equivalent_function_chain()}
+        )
+        assert chain_rsp["code"] == 0, f"search with L0 functionChains failed: {chain_rsp}"
+
+        boost_rsp = self.vector_client.vector_search(
+            {**base_payload, "functionScore": self._boost_ts_flag_function_score()}
+        )
+        assert boost_rsp["code"] == 0, f"search with boost rank failed: {boost_rsp}"
+
+        chain_ids = [item["id"] for item in chain_rsp["data"]]
+        boost_ids = [item["id"] for item in boost_rsp["data"]]
+        assert chain_ids == boost_ids == [2, 1, 3]
+
    def test_search_with_function_chains_reranks_by_scalar_field(self):
        """
        target: test REST v2 search with functionChains