fix: harden membership filter planning (#53100)

## What this fixes

- Keep unsupported membership-filter leaves conservative during
clustering-key scalar pruning by cloning the all-true bitmap before
mutable logical operations. This prevents false-negative segment pruning
when an unsupported leaf is reused under nested AND/OR expressions.
- Move each Bloom blob out of its uniquely owned parsed `PlanNode` /
`ScoreFunction` into an immutable shared string. Serialized search,
retrieve, and standalone scorer paths no longer create a third owning
copy of a large MBF1 blob, and they do not retain unrelated protobuf
fields; legacy const-reference entry points retain safe-copy behavior.
- Reject bytes template values before arithmetic operand type inference,
returning a parameter error instead of dereferencing a null internal
value expression.
- Parse each JSON membership value and resolve its path once before
dispatching to the string or int64 Bloom domain. Double and out-of-range
uint64 values remain valid non-members; missing/null/bool/object/array
values remain unknown.
- Avoid widening narrow signed integer slices into a full `[]int64`
before Roaring construction. Ordered inputs allocate no full-length
conversion buffer; unordered inputs allocate only the required
`[]uint64` sorting buffer, while `Build([]int64)` remains
source-compatible.

No syntax or wire-format changes are introduced. The JSON Bloom
scheduling-cost follow-up remains intentionally deferred.

Related feature PR: #52778
Original 3.0 feature backport: #53017
3.0 unified follow-up: #53019
Design:
https://github.com/milvus-io/milvus/blob/master/docs/design-docs/design_docs/20260822-membership-match-expression.md

## Tests

- `LOCALSTORAGEPATH=/tmp/milvus-membership-master-parser go test -tags
dynamic ./internal/parser/planparserv2 -count=1`
- `go test -tags test ./internal/querynodev2/delegator -run
TestUnsupportedBloomLeavesStayConservative -count=1`
- Remote C++ build/link against `3d14599542`
- `all_tests
--gtest_filter=PlanProto.BloomBlobMovesFromOwnedSearchPlanNode:PlanProto.BloomBlobMovesFromOwnedRetrievePlanNode:PlanProto.BloomBlobMovesFromOwnedScoreFunction:BloomFilterExprTest.ProtoParserDispatchAndTypeCheck`
(4/4 passed)
- Remote C++ build/link against `3445f06a37`
- `test_json_uint64
--gtest_filter=JsonNumericTest.AtStringOrInt64DispatchesRuntimeTypes:JsonNumericTest.AtStringOrInt64SupportsRootPointer`
(2/2 passed)
- `all_tests
--gtest_filter=BloomFilterExprEvalTest.JsonPathGrowingAndSealed:BloomFilterExprEvalTest.JsonPathStringBlobKeepsTypesDistinct:BloomFilterExprEvalTest.JsonPathOffsetInput`
(3/3 passed, including escaped-string membership and offset-path
NULL/uint64 coverage)
- `go test -tags dynamic,test -gcflags='all=-N -l' -count=1
./membership/roaringfilter ./milvusclient`
- Added signed-width, named-type, byte-equivalence, non-mutation, and
ordered/shuffled allocation-parity regression coverage for Roaring
construction.

- Final style-only head `c1e842b42a`: clang-format-15 dry-run passed for
all 11 changed C/C++ files

Signed-off-by: xiaofanluan <xf@hjjaq.com>
Co-authored-by: xiaofanluan <xf@hjjaq.com>

client/membership/roaringfilter/roaring_filter.go

@@ -93,15 +93,25 @@ const (
 	EstimatedLowContainerOverheadBytes = 64
 )
 
-// Build deduplicates members into a portable Roaring64 bitmap and wraps it in
-// an MRB1 envelope.
+// Build deduplicates int64 members into a portable Roaring64 bitmap and wraps
+// it in an MRB1 envelope. It retains the original []int64 API; callers with a
+// narrower signed slice can use BuildSigned to avoid widening the whole input.
 func Build(members []int64) ([]byte, error) {
+	return BuildSigned(members)
+}
+
+// BuildSigned deduplicates signed integer members into a portable Roaring64
+// bitmap and wraps it in an MRB1 envelope. Narrow signed values are extended to
+// int64 before their two's-complement bits become Roaring keys. The conversion
+// happens as each value is consumed, so callers do not need an intermediate
+// []int64 allocation.
+func BuildSigned[T ~int | ~int8 | ~int16 | ~int32 | ~int64](members []T) ([]byte, error) {
 	var sortedKeys []uint64
 	ascending, highContainerCount, lowContainerCount := memberContainerCounts(members)
 	if !ascending {
 		sortedKeys = make([]uint64, len(members))
 		for i, member := range members {
-			sortedKeys[i] = uint64(member)
+			sortedKeys[i] = uint64(int64(member))
 		}
 		slices.Sort(sortedKeys)
 		highContainerCount, lowContainerCount = sortedKeyContainerCounts(sortedKeys)
@@ -128,7 +138,7 @@ func Build(members []int64) ([]byte, error) {
 			// Add extends the last container or appends a new one, so this is
 			// linear and, unlike the sort path, allocates nothing per member.
 			for _, member := range members {
-				bitmap.Add(uint64(member))
+				bitmap.Add(uint64(int64(member)))
 			}
 		} else {
 			bitmap.AddMany(sortedKeys)
@@ -178,15 +188,15 @@ func Build(members []int64) ([]byte, error) {
 // memberContainerCounts reports whether members is already non-decreasing in
 // the unsigned key space and, when it is, counts distinct high-32 children and
 // Roaring32 low containers without allocating. Equal keys are duplicates.
-func memberContainerCounts(members []int64) (bool, uint64, uint64) {
+func memberContainerCounts[T ~int | ~int8 | ~int16 | ~int32 | ~int64](members []T) (bool, uint64, uint64) {
 	if len(members) == 0 {
 		return true, 0, 0
 	}
-	previous := uint64(members[0])
+	previous := uint64(int64(members[0]))
 	highContainerCount := uint64(1)
 	lowContainerCount := uint64(1)
 	for i := 1; i < len(members); i++ {
-		key := uint64(members[i])
+		key := uint64(int64(members[i]))
 		if key < previous {
 			return false, 0, 0
 		}

client/membership/roaringfilter/roaring_filter_test.go

@@ -26,6 +26,8 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+var _ func([]int64) ([]byte, error) = Build
+
 // The structural rejection suite for this format lives with the validator, in
 // pkg/util/roaringfilter, and the cross-check that a blob built here is one the
 // proxy accepts lives in internal/parser/planparserv2 (it needs both modules).
@@ -99,6 +101,40 @@ func TestBuildPreservesSignedBitPattern(t *testing.T) {
 	}
 }
 
+type namedSignedInt16 int16
+
+func requireBuildSignedMatchesBuild[T ~int | ~int8 | ~int16 | ~int32 | ~int64](
+	t *testing.T,
+	members []T,
+) {
+	t.Helper()
+	widened := make([]int64, len(members))
+	for i, member := range members {
+		widened[i] = int64(member)
+	}
+
+	want, err := Build(widened)
+	require.NoError(t, err)
+	got, err := BuildSigned(members)
+	require.NoError(t, err)
+	require.Equal(t, want, got)
+}
+
+func TestBuildSignedMatchesInt64API(t *testing.T) {
+	requireBuildSignedMatchesBuild(t, []int{-1, 0, 1, 42})
+	requireBuildSignedMatchesBuild(t, []int8{math.MinInt8, -1, 0, math.MaxInt8})
+	requireBuildSignedMatchesBuild(t, []int16{math.MinInt16, -1, 0, math.MaxInt16})
+	requireBuildSignedMatchesBuild(t, []int32{math.MinInt32, -1, 0, math.MaxInt32})
+	requireBuildSignedMatchesBuild(t, []int64{math.MinInt64, -1, 0, math.MaxInt64})
+	requireBuildSignedMatchesBuild(t, []namedSignedInt16{-1, 0, 42})
+
+	members := []int16{9, 3, -5, 1}
+	original := append([]int16(nil), members...)
+	_, err := BuildSigned(members)
+	require.NoError(t, err)
+	require.Equal(t, original, members, "BuildSigned must not reorder the caller's slice")
+}
+
 func TestBuildEmptySet(t *testing.T) {
 	blob, err := Build(nil)
 	require.NoError(t, err)

client/milvusclient/bloom_filter.go

@@ -52,9 +52,10 @@ type BloomFilterBlob []byte
 // (docs/design-docs/design_docs/20260707-bloom-filter-expression.md).
 //
 // The blob body must fit proxy.maxMembershipFilterSize (64 MiB by default),
-// all membership-filter-bearing plans in the request must fit
-// proxy.maxMembershipFilterPlanSize (128 MiB by default), and the complete
-// request must fit proxy.grpc.serverMaxRecvSize. Use
+// the request-wide serialized membership plans (and, for Roaring filters,
+// aggregate estimated decoded bytes) must fit proxy.maxMembershipFilterPlanSize
+// (128 MiB by default), and the complete request must fit
+// proxy.grpc.serverMaxRecvSize. Use
 // sbbf.EstimateMarshalSize(n, fpr) to check the exact blob size for a planned
 // member count before building the filter. Bodies are powers of two, so a
 // count just past a tier boundary doubles the blob and a slightly higher fpr

client/milvusclient/roaring_filter.go

@@ -27,39 +27,29 @@ import (
 // template value and contains an MRB1 envelope around portable Roaring64 data.
 type RoaringBitmapBlob []byte
 
-type signedInteger interface {
-	~int | ~int8 | ~int16 | ~int32 | ~int64
-}
-
-func widenSignedIntegers[T signedInteger](members []T) []int64 {
-	values := make([]int64, len(members))
-	for i, member := range members {
-		values[i] = int64(member)
-	}
-	return values
-}
-
 // NewRoaringBitmapBlob builds an exact membership bitmap from a slice of
 // signed integers. Supported input types are []int, []int8, []int16, []int32,
 // and []int64; values are sign-extended to int64 before serialization.
 func NewRoaringBitmapBlob(members any) (RoaringBitmapBlob, error) {
-	var values []int64
+	var (
+		blob []byte
+		err  error
+	)
 	switch typed := members.(type) {
 	case []int:
-		values = widenSignedIntegers(typed)
+		blob, err = roaringfilter.BuildSigned(typed)
 	case []int8:
-		values = widenSignedIntegers(typed)
+		blob, err = roaringfilter.BuildSigned(typed)
 	case []int16:
-		values = widenSignedIntegers(typed)
+		blob, err = roaringfilter.BuildSigned(typed)
 	case []int32:
-		values = widenSignedIntegers(typed)
+		blob, err = roaringfilter.BuildSigned(typed)
 	case []int64:
-		values = typed
+		blob, err = roaringfilter.Build(typed)
 	default:
 		return nil, errors.Errorf(
 			"roaring bitmap members must be []int, []int8, []int16, []int32, or []int64, got %T", members)
 	}
-	blob, err := roaringfilter.Build(values)
 	if err != nil {
 		return nil, err
 	}

client/milvusclient/roaring_filter_test.go

@@ -106,3 +106,72 @@ func TestRoaringBitmapBlobTemplateMarshaling(t *testing.T) {
 	require.True(t, ok, "RoaringBitmapBlob must marshal as native protobuf bytes")
 	require.Equal(t, []byte(blob), bytesValue.BytesVal)
 }
+
+var benchmarkRoaringBitmapBlobSink RoaringBitmapBlob
+
+func makeSignedMembers[T ~int32 | ~int64](n int, shuffled bool) []T {
+	members := make([]T, n)
+	for i := range members {
+		value := i
+		if shuffled {
+			value = (i*2053)%n - n/2
+		}
+		members[i] = T(value)
+	}
+	return members
+}
+
+func roaringBitmapBlobAllocsPerRun(members any) (float64, error) {
+	var err error
+	allocs := testing.AllocsPerRun(20, func() {
+		benchmarkRoaringBitmapBlobSink, err = NewRoaringBitmapBlob(members)
+	})
+	return allocs, err
+}
+
+func TestNewRoaringBitmapBlobNarrowAllocationParity(t *testing.T) {
+	for _, shuffled := range []bool{false, true} {
+		name := "ordered"
+		if shuffled {
+			name = "shuffled"
+		}
+		t.Run(name, func(t *testing.T) {
+			int32Members := makeSignedMembers[int32](4096, shuffled)
+			int64Members := makeSignedMembers[int64](4096, shuffled)
+			int32Allocs, err := roaringBitmapBlobAllocsPerRun(int32Members)
+			require.NoError(t, err)
+			int64Allocs, err := roaringBitmapBlobAllocsPerRun(int64Members)
+			require.NoError(t, err)
+			require.Equal(t, int64Allocs, int32Allocs,
+				"a narrow input must not allocate an intermediate []int64")
+		})
+	}
+}
+
+func benchmarkNewRoaringBitmapBlob(b *testing.B, members any) {
+	b.Helper()
+	blob, err := NewRoaringBitmapBlob(members)
+	if err != nil {
+		b.Fatal(err)
+	}
+	benchmarkRoaringBitmapBlobSink = blob
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		blob, err := NewRoaringBitmapBlob(members)
+		if err != nil {
+			b.Fatal(err)
+		}
+		benchmarkRoaringBitmapBlobSink = blob
+	}
+}
+
+func BenchmarkNewRoaringBitmapBlobInt32Shuffled(b *testing.B) {
+	members := makeSignedMembers[int32](64*1024, true)
+	benchmarkNewRoaringBitmapBlob(b, members)
+}
+
+func BenchmarkNewRoaringBitmapBlobInt64Shuffled(b *testing.B) {
+	members := makeSignedMembers[int64](64*1024, true)
+	benchmarkNewRoaringBitmapBlob(b, members)
+}

configs/milvus.yaml

@@ -402,7 +402,7 @@ proxy:
   maxVectorFieldNum: 10 # The maximum number of vector fields that can be specified in a collection
   maxShardNum: 16 # The maximum number of shards can be created when creating in a collection.
   maxMembershipFilterSize: 67108864 # The maximum byte size of one client pre-built membership_match filter body accepted by the proxy (the fixed 32-byte MBF1/MRB1 header is allowed on top). The blob is embedded into the query plan and fanned out to every QueryNode, so this bounds per-membership-filter memory/network amplification. Must not exceed the format cap (128 MiB). Bloom SBBF bodies are powers of two, so the default admits bodies up to 64 MiB: ~48.6M int64 members at the default fpr=0.005, ~55.4M at fpr=0.01. Raising the member count past a tier boundary requires raising the fpr too, not only this limit. Roaring filters remain independently limited to 262144 high containers and a 64 MiB decoded-memory estimate per filter. Keep proxy.grpc.serverMaxRecvSize above this plus the rest of the request.
-  maxMembershipFilterPlanSize: 134217728 # The maximum aggregate serialized byte size of membership-filter-bearing expression plans in one Search, HybridSearch, Query, or complex Delete request. The proxy checks the assembled plans with proto.Size before proto.Marshal, and hybrid sub-searches share the configured budget. Must be positive; invalid values fall back to 128 MiB.
+  maxMembershipFilterPlanSize: 134217728 # The request-wide membership-filter budget in bytes. It independently limits both the aggregate serialized size of membership-filter-bearing expression plans and the aggregate estimated decoded size of Roaring filters in one Search, HybridSearch, Query, or complex Delete request. The proxy checks assembled plans with proto.Size before proto.Marshal, and hybrid sub-searches and scorer filters share both totals. Must be positive; invalid values fall back to 128 MiB.
   maxDimension: 32768 # The maximum number of dimensions of a vector can have when creating in a collection.
   # Whether to produce gin logs.\n
   # please adjust in embedded Milvus: false

docs/design-docs/design_docs/20260707-bloom-filter-expression.md

@@ -474,13 +474,15 @@ framework, and the delete-safety guard are all encoding-agnostic and reused unch
   in [64 MiB, 128 MiB) admits the same Bloom filters, so the default is set to the smallest
   of them and states the real ceiling. The 128 MiB MBF1 num_blocks format cap remains the
   hard Bloom per-blob ceiling. The legacy `proxy.maxBloomFilterSize` key remains a fallback.
-- **`proxy.maxMembershipFilterPlanSize`** (default 128 MiB, refreshable): request-wide cap
-  on the serialized size of every membership-filter-bearing plan, enforced with
-  `proto.Size(plan)` before `proto.Marshal`. Reusing one template in multiple expressions
-  therefore consumes the budget for every embedded copy. Hybrid search accumulates all
-  membership-filter-bearing sub-plans against the same request budget, and scorer filters are
-  included: a membership expression is allowed in a scorer filter and its blob is charged to
-  the same budget.
+- **`proxy.maxMembershipFilterPlanSize`** (default 128 MiB, refreshable): request-wide
+  membership budget. It independently caps the serialized size of every
+  membership-filter-bearing plan and, for Roaring filters, the aggregate estimated decoded
+  size. The serialized total is enforced with `proto.Size(plan)` before `proto.Marshal`;
+  Roaring decoded estimates are charged before materialization. Every embedded copy consumes
+  the serialized total; only Roaring occurrences additionally consume the decoded total.
+  Hybrid search accumulates all membership-filter-bearing sub-plans against the same request
+  budget, and scorer filters are included: a membership expression is allowed in a scorer
+  filter and its applicable costs are charged to the same budget.
   Exceeding the cap returns
   `ErrParameterTooLarge` (1102, InputError, non-retriable) at the proxy, before QueryNode routing,
   so an over-budget Bloom request cannot turn into a gRPC `ResourceExhausted` or blacklist a

docs/design-docs/design_docs/20260714-roaring-exact-membership-expression.md

@@ -277,7 +277,7 @@ be too expensive to fan out. Two limits bound that independently.
 | Limit | Default | Bounds |
 | --- | --- | --- |
 | `proxy.maxMembershipFilterSize` | 64 MiB | One MBF1 or MRB1 body. The fixed 32-byte header is allowed on top. For Roaring, decoded-memory admission usually becomes the tighter bound before the full wire budget is usable. |
-| `proxy.maxMembershipFilterPlanSize` | 128 MiB | The aggregate serialized size of every membership-filter-bearing plan in one Search, HybridSearch, Query, or complex Delete. |
+| `proxy.maxMembershipFilterPlanSize` | 128 MiB | The request-wide membership budget. It independently bounds both the aggregate serialized size of membership-filter-bearing plans and the aggregate estimated decoded size of all MRB1 occurrences in one Search, HybridSearch, Query, or complex Delete. |
 | MRB1 high-container admission | 262,144 | Separately allocated Roaring32 children per expression, enforced by SDK, Proxy, and QueryNode. |
 | MRB1 estimated decoded admission | 64 MiB | `body + 128 bytes/high container + 64 bytes/low container`, enforced by SDK, Proxy, and QueryNode. |
 
@@ -290,13 +290,15 @@ in the serialized plan and fanned out to every QueryNode — so a request carryi
 one of each must not be allowed to spend twice what either alone may. The legacy
 Bloom-specific keys remain fallbacks for deployments that already set them.
 
-After template materialization, Proxy applies the exact `proto.Size` gate before
+Before materialization, Proxy applies the same request-wide configured value to
+the aggregate estimated decoded size of all MRB1 occurrences. After template
+materialization, it independently applies the exact `proto.Size` gate before
 `proto.Marshal`; this catches repeated embedded copies, protobuf overhead, and
 HybridSearch accumulation. Each Roaring template is structurally validated and
-must independently satisfy the 64 MiB decoded estimate. Materialized scorer
-filters may carry membership expressions and their blobs count against the same
-per-request budget, so a blob embedded in a scorer filter is charged exactly as
-one embedded in the main predicate.
+must independently satisfy the fixed 64 MiB decoded estimate. Materialized
+scorer filters may carry membership expressions and their serialized and decoded
+costs count against the same request-wide budget, so a blob embedded in a scorer
+filter is charged exactly as one embedded in the main predicate.
 
 Sizing by bytes rather than member count is the point. Unlike an SBBF body,
 whose size follows `(n, fpr)` alone, a Roaring body's size is driven by the
@@ -553,7 +555,9 @@ the 64 MiB decoded estimate; the tightest bound is distribution-sensitive.
 - Proxy validation is linear in supplied bytes, allocation-free on success, and
   never materializes a bitmap or rebuilds a raw list.
 - Each materialized Roaring expression is validated independently at Proxy and
-  QueryNode and must satisfy the per-filter decoded estimate.
+  QueryNode and must satisfy the fixed per-filter decoded estimate. Proxy also
+  charges every occurrence to the request-wide decoded total controlled by
+  `proxy.maxMembershipFilterPlanSize`.
 - QueryNode parses each logical expression once and shares its immutable decoded
   bitmap across the corresponding per-segment physical expressions.
 - The raw path performs one probe per active valid row.
@@ -745,8 +749,8 @@ Existing collections, segments, and indexes require no rewrite.
 - Materialize `RoaringFilterExpr` at oneof field 23 and verify proto round-trip.
 - Reject malformed MRB1 at Proxy.
 - Repeat one template across multiple AST leaves and verify each materialized
-  expression remains valid while the serialized plan-size gate counts every
-  embedded copy.
+  expression remains valid while the request-wide gate counts every embedded
+  copy independently in both the serialized and estimated-decoded totals.
 - Prove a Roaring plan is not classified as `BloomFilterExpr`.
 
 ### QueryNode logical and physical execution tests

docs/design-docs/design_docs/20260822-membership-match-expression.md

@@ -111,15 +111,16 @@ All in `internal/parser/planparserv2/`:
    64 MiB) limits one blob body and falls back to the released
    `proxy.maxBloomFilterSize` key (plus the development-only
    `proxy.maxRoaringFilterSize` predecessor). The separate
-   `proxy.maxMembershipFilterPlanSize` (default 128 MiB) limits aggregate
-   serialized membership-bearing plans for one request and falls back to
-   `proxy.maxBloomFilterPlanSize`. The parser uses that same aggregate ceiling
-   for a conservative body-occurrence preflight shared by all HybridSearch
-   sub-requests and scorer filters; exact `proto.Size` accounting remains the
-   final gate. Keeping the two dimensions separate prevents a legacy plan
-   setting from silently widening the per-blob admission limit. The fixed MRB1
-   admissions (262,144 high containers, 64 MiB decoded estimate per occurrence,
-   aggregate) are unchanged.
+   `proxy.maxMembershipFilterPlanSize` (default 128 MiB) is the request-wide
+   membership budget and falls back to `proxy.maxBloomFilterPlanSize`. Its value
+   independently limits aggregate serialized membership-bearing plans and
+   aggregate estimated decoded bytes for all MRB1 occurrences. The parser shares
+   both totals across HybridSearch sub-requests and scorer filters; exact
+   `proto.Size` accounting remains the final serialized gate. Keeping the two
+   configurable dimensions separate prevents a legacy plan setting from
+   silently widening the per-blob admission limit. The fixed MRB1 admissions
+   (262,144 high containers and a 64 MiB decoded estimate per occurrence) remain
+   unchanged.
 
 ### Guards unified
 

internal/core/src/common/Json.h

@@ -20,6 +20,7 @@
 #include <boost/algorithm/string/join.hpp>
 #include <boost/algorithm/string/replace.hpp>
 #include <cstddef>
+#include <cstdint>
 #include <cstdlib>
 #include <optional>
 #include <string>
@@ -112,6 +113,20 @@ ExtractSubJson(std::string_view json, const std::vector<std::string>& keys) {
 using document = simdjson::ondemand::document;
 template <typename T>
 using value_result = simdjson::simdjson_result<T>;
+
+struct JsonStringOrInt64 {
+    enum class Kind : uint8_t {
+        NoProbeValue,
+        String,
+        Int64,
+        OtherNumber,
+    };
+
+    Kind kind{Kind::NoProbeValue};
+    std::string_view string_value{};
+    int64_t int64_value{};
+};
+
 class Json {
  public:
     Json() = default;
@@ -282,6 +297,60 @@ class Json {
         return doc().at_pointer(pointer).get_number();
     }
 
+    // Parse the document and resolve the JSON pointer exactly once, then
+    // classify the value for callers that accept only strings or int64s.
+    // string_value must be consumed before the next ondemand parse on this
+    // thread because simdjson may store decoded strings in its parser buffer.
+    JsonStringOrInt64
+    at_string_or_int64(std::string_view pointer) const {
+        const auto extract = [](auto&& value) {
+            auto type = value.type();
+            if (type.error()) {
+                return JsonStringOrInt64{};
+            }
+
+            switch (type.value()) {
+                case simdjson::ondemand::json_type::string: {
+                    auto str = value.get_string(false);
+                    if (str.error()) {
+                        return JsonStringOrInt64{};
+                    }
+                    return JsonStringOrInt64{
+                        JsonStringOrInt64::Kind::String, str.value(), 0};
+                }
+                case simdjson::ondemand::json_type::number: {
+                    auto number = value.get_number();
+                    if (number.error()) {
+                        return JsonStringOrInt64{};
+                    }
+                    auto n = number.value();
+                    if (n.is_int64()) {
+                        return JsonStringOrInt64{
+                            JsonStringOrInt64::Kind::Int64, {}, n.get_int64()};
+                    }
+                    return JsonStringOrInt64{
+                        JsonStringOrInt64::Kind::OtherNumber, {}, 0};
+                }
+                default:
+                    return JsonStringOrInt64{};
+            }
+        };
+
+        auto document = doc();
+        if (document.error()) {
+            return {};
+        }
+        if (pointer.empty()) {
+            return extract(document.value());
+        }
+
+        auto value = document.value().at_pointer(pointer);
+        if (value.error()) {
+            return {};
+        }
+        return extract(value.value());
+    }
+
     value_result<std::string>
     at_string_any(std::string_view pointer) const {
         if (data_.empty()) {

internal/core/src/exec/expression/Expr.cpp

@@ -776,7 +776,17 @@ ReorderConjunctExpr(std::shared_ptr<milvus::exec::PhyConjunctFilterExpr>& expr,
         // indexed predicates that can prune what it has to probe.
         if (input->name() == "PhyBloomFilterExpr" ||
             input->name() == "PhyRoaringFilterExpr") {
-            membership_expr.push_back(i);
+            // A JSON-path membership probe parses JSON and resolves a pointer
+            // per evaluated row, which is a heavy operation; bucket it with
+            // the other JSON predicates instead of the string-compare tier so
+            // it runs after cheaper predicates that can prune its rows.
+            if (input->IsSource() && input->GetColumnInfo().has_value() &&
+                IsJsonDataType(input->GetColumnInfo()->data_type_)) {
+                json_expr.push_back(i);
+                has_heavy_operation = true;
+            } else {
+                membership_expr.push_back(i);
+            }
             continue;
         }
 

internal/core/src/exec/expression/Expr.h

@@ -750,6 +750,75 @@ class SegmentExpr : public Expr {
             evaluate_batch, input, res, valid_res, &candidate_mask, values...);
     }
 
+    // Sequential reverse-lookup over the contiguous global row range
+    // [start_offset, start_offset + batch_size). This is the shape of the
+    // no-offset-input index-only scan, and it avoids materializing a
+    // per-batch OffsetVector solely to encode a start plus a count.
+    template <typename T, typename BatchEvaluator, typename... ValTypes>
+    int64_t
+    ProcessIndexLookupSequentialWithMask(BatchEvaluator evaluate_batch,
+                                         int64_t start_offset,
+                                         int64_t batch_size,
+                                         TargetBitmapView res,
+                                         TargetBitmapView valid_res,
+                                         const TargetBitmap& candidate_mask,
+                                         const ValTypes&... values) {
+        AssertInfo(num_index_chunk_ == 1, "scalar index chunk num must be 1");
+        using IndexInnerType = std::
+            conditional_t<std::is_same_v<T, std::string_view>, std::string, T>;
+        using Index = index::ScalarIndex<IndexInnerType>;
+        auto scalar_index = dynamic_cast<const Index*>(pinned_index_[0].get());
+        auto* index_ptr = const_cast<Index*>(scalar_index);
+        const auto& valid_result = GetCachedIndexValidBitmap(index_ptr);
+        const bool all_valid = cached_index_all_valid_;
+        const bool has_candidate_mask = !candidate_mask.empty();
+        AssertInfo(!has_candidate_mask ||
+                       candidate_mask.size() == static_cast<size_t>(batch_size),
+                   "candidate mask size {} does not match offset batch size {}",
+                   candidate_mask.size(),
+                   batch_size);
+
+        for (auto i = 0; i < batch_size; ++i) {
+            if (has_candidate_mask && !candidate_mask[i]) {
+                evaluate_batch.template operator()<FilterType::random>(
+                    nullptr,
+                    nullptr,
+                    nullptr,
+                    1,
+                    res + i,
+                    valid_res + i,
+                    values...);
+                continue;
+            }
+            auto offset = start_offset + i;
+            auto raw = index_ptr->Reverse_Lookup(offset);
+            if (!raw.has_value()) {
+                res[i] = valid_res[i] = false;
+                evaluate_batch.template operator()<FilterType::random>(
+                    nullptr,
+                    ValidityView{},
+                    nullptr,
+                    1,
+                    res + i,
+                    valid_res + i,
+                    values...);
+                continue;
+            }
+            T raw_data = raw.value();
+            bool valid_data = all_valid || valid_result[offset];
+            evaluate_batch.template operator()<FilterType::random>(
+                &raw_data,
+                ValidityView::FromExpanded(&valid_data),
+                nullptr,
+                1,
+                res + i,
+                valid_res + i,
+                values...);
+        }
+
+        return batch_size;
+    }
+
     template <typename T, typename BatchEvaluator, typename... ValTypes>
     int64_t
     ProcessIndexLookupByOffsetsImpl(BatchEvaluator evaluate_batch,

internal/core/src/exec/expression/ExprJsonUint64Test.cpp

@@ -85,6 +85,42 @@ TEST(JsonNumericTest, AtNumericDoubleDetection) {
     EXPECT_DOUBLE_EQ(n.get_double(), 3.14);
 }
 
+TEST(JsonNumericTest, AtStringOrInt64DispatchesRuntimeTypes) {
+    auto j = makeJson(
+        R"({"str":"alice","escaped":"a\u006cice","int":-7,"uint":18446744073709551615,"double":5.0,"bool":true})");
+
+    const auto str = j.at_string_or_int64("/str");
+    EXPECT_EQ(str.kind, JsonStringOrInt64::Kind::String);
+    EXPECT_EQ(str.string_value, "alice");
+
+    const auto escaped = j.at_string_or_int64("/escaped");
+    EXPECT_EQ(escaped.kind, JsonStringOrInt64::Kind::String);
+    EXPECT_EQ(escaped.string_value, "alice");
+
+    const auto integer = j.at_string_or_int64("/int");
+    EXPECT_EQ(integer.kind, JsonStringOrInt64::Kind::Int64);
+    EXPECT_EQ(integer.int64_value, -7);
+
+    const auto uint = j.at_string_or_int64("/uint");
+    EXPECT_EQ(uint.kind, JsonStringOrInt64::Kind::OtherNumber);
+
+    const auto floating = j.at_string_or_int64("/double");
+    EXPECT_EQ(floating.kind, JsonStringOrInt64::Kind::OtherNumber);
+
+    const auto other = j.at_string_or_int64("/bool");
+    EXPECT_EQ(other.kind, JsonStringOrInt64::Kind::NoProbeValue);
+
+    const auto missing = j.at_string_or_int64("/missing");
+    EXPECT_EQ(missing.kind, JsonStringOrInt64::Kind::NoProbeValue);
+}
+
+TEST(JsonNumericTest, AtStringOrInt64SupportsRootPointer) {
+    auto j = makeJson(R"(-9223372036854775808)");
+    const auto root = j.at_string_or_int64("");
+    EXPECT_EQ(root.kind, JsonStringOrInt64::Kind::Int64);
+    EXPECT_EQ(root.int64_value, std::numeric_limits<int64_t>::min());
+}
+
 // ═══════════════════════════════════════════════════════════════════════════════
 // Document why static_cast<int64_t> is wrong (the bug we're fixing)
 // ═══════════════════════════════════════════════════════════════════════════════

internal/core/src/exec/expression/MembershipFilterExpr.cpp

@@ -275,18 +275,19 @@ PhyMembershipFilterExpr<LogicalExpr, ProbePolicy>::ExecVisitorImplForIndex(
                                                          bitmap_input);
     } else {
         // No offset input: reverse-look-up the contiguous global row range
-        // [current_index_chunk_pos_, +real_batch_size) for this batch.
-        OffsetVector batch_offsets(real_batch_size);
-        auto start = current_index_chunk_pos_;
-        for (int64_t i = 0; i < real_batch_size; ++i) {
-            batch_offsets[i] = static_cast<int32_t>(start + i);
-        }
-        processed_size = ProcessIndexLookupByOffsetsWithMask<T>(
-            execute_sub_batch, &batch_offsets, res, valid_res, bitmap_input);
-        // ProcessIndexLookupByOffsets is stateless; advance the index cursor
-        // for the next batch. MoveCursor() honors the has_offset_input_ guard
-        // and, on the ScalarIndex path with no raw data, advances only the
-        // index cursor.
+        // [current_index_chunk_pos_, +real_batch_size) for this batch, without
+        // materializing a per-batch OffsetVector.
+        processed_size =
+            ProcessIndexLookupSequentialWithMask<T>(execute_sub_batch,
+                                                    current_index_chunk_pos_,
+                                                    real_batch_size,
+                                                    res,
+                                                    valid_res,
+                                                    bitmap_input);
+        // ProcessIndexLookupSequentialWithMask is stateless; advance the index
+        // cursor for the next batch. MoveCursor() honors the has_offset_input_
+        // guard and, on the ScalarIndex path with no raw data, advances only
+        // the index cursor.
         MoveCursor();
     }
     AssertInfo(processed_size == real_batch_size,
@@ -382,19 +383,22 @@ PhyMembershipFilterExpr<LogicalExpr, ProbePolicy>::ExecVisitorImplJson(
             // The blob's declared value domains gate the probes themselves,
             // so a JSON string cannot alias an int64-only filter.
             const auto& json = data[offset];
-            auto str = json.template at<std::string_view>(pointer);
-            if (!str.error()) {
-                res[i] = probe_.TestBytesValue(str.value().data(),
-                                               str.value().size());
-                continue;
-            }
-            auto num = json.at_numeric(pointer);
-            if (num.error()) {
-                res[i] = valid_res[i] = false;
-                continue;
-            }
-            if (auto n = num.value(); n.is_int64()) {
-                res[i] = probe_.TestInt64Value(n.get_int64());
+            const auto value = json.at_string_or_int64(pointer);
+            switch (value.kind) {
+                case JsonStringOrInt64::Kind::String:
+                    res[i] = probe_.TestBytesValue(value.string_value.data(),
+                                                   value.string_value.size());
+                    break;
+                case JsonStringOrInt64::Kind::Int64:
+                    res[i] = probe_.TestInt64Value(value.int64_value);
+                    break;
+                case JsonStringOrInt64::Kind::OtherNumber:
+                    // A numeric value outside the int64 domain is a definite
+                    // non-member, but it is still a valid JSON scalar.
+                    break;
+                case JsonStringOrInt64::Kind::NoProbeValue:
+                    res[i] = valid_res[i] = false;
+                    break;
             }
         }
         processed_cursor += size;

internal/core/src/exec/expression/MembershipFilterExpr.h

@@ -74,9 +74,9 @@ struct BloomMembershipProbe {
 
     // Envelope parsed and validated exactly once per physical expr; throws
     // SegcoreError{ExprInvalid} on malformed blobs. The view aliases
-    // expr.filter_blob_, owned by the logical node held via shared_ptr.
+    // *expr.filter_blob_, owned by the logical node held via shared_ptr.
     explicit BloomMembershipProbe(const milvus::expr::BloomFilterExpr& expr)
-        : filter(SplitBlockBloomFilterView::Parse(expr.filter_blob_)) {
+        : filter(SplitBlockBloomFilterView::Parse(*expr.filter_blob_)) {
     }
 
     template <typename T>

internal/core/src/expr/ITypeExpr.h

@@ -990,15 +990,24 @@ class RoaringFilterExpr : public ITypeFilterExpr {
 // BloomFilterExpr: approximate membership filter (`membership_match`, MBF1).
 // See docs/design-docs/design_docs/20260707-bloom-filter-expression.md.
 //
-// filter_blob_ owns the MBF1-enveloped parquet SBBF blob (copied once out of
-// the transient plan proto at parse time). Physical expressions compiled from
-// this node keep zero-copy views into filter_blob_; they hold this logical
-// expr via shared_ptr, so the blob outlives every per-segment physical expr
-// of the query and is never duplicated per segment.
+// filter_blob_ owns the MBF1-enveloped parquet SBBF blob. Serialized
+// plan/scorer parsing moves the blob into this immutable shared string; direct
+// callers keep safe ownership through the string overload below.
+// Physical expressions hold this logical expr via shared_ptr, so their
+// zero-copy Bloom views remain valid for the query lifetime.
 class BloomFilterExpr : public ITypeFilterExpr {
  public:
     BloomFilterExpr(const ColumnInfo& column, std::string filter_blob)
+        : BloomFilterExpr(
+              column,
+              std::make_shared<const std::string>(std::move(filter_blob))) {
+    }
+
+    BloomFilterExpr(const ColumnInfo& column,
+                    std::shared_ptr<const std::string> filter_blob)
         : column_(column), filter_blob_(std::move(filter_blob)) {
+        AssertInfo(filter_blob_ != nullptr,
+                   "BloomFilterExpr filter blob owner must not be null");
     }
 
     std::string
@@ -1010,17 +1019,17 @@ class BloomFilterExpr : public ITypeFilterExpr {
         // the (up to 128 MiB) blob: report the blob length and the declared
         // element count from the MBF1 header instead.
         const uint64_t n_declared =
-            milvus::bloom_envelope::DecodeNDeclared(filter_blob_);
+            milvus::bloom_envelope::DecodeNDeclared(*filter_blob_);
         return fmt::format(
             "BloomFilterExpr:[Column: {}, BlobBytes: {}, DeclaredElements: {}]",
             column_.ToString(),
-            filter_blob_.size(),
+            filter_blob_->size(),
             n_declared);
     }
 
  public:
     const ColumnInfo column_;
-    const std::string filter_blob_;
+    const std::shared_ptr<const std::string> filter_blob_;
 };
 
 }  // namespace expr

internal/core/src/query/Plan.cpp

@@ -250,9 +250,9 @@ CreateSearchPlanByExpr(SchemaPtr schema,
                        const void* serialized_expr_plan,
                        const int64_t size) {
     // Note: serialized_expr_plan is of binary format
-    proto::plan::PlanNode plan_node;
-    ParsePlanNodeProto(plan_node, serialized_expr_plan, size);
-    return ProtoParser(std::move(schema)).CreatePlan(plan_node);
+    auto plan_node = std::make_unique<proto::plan::PlanNode>();
+    ParsePlanNodeProto(*plan_node, serialized_expr_plan, size);
+    return ProtoParser(std::move(schema)).CreatePlan(std::move(plan_node));
 }
 
 std::unique_ptr<Plan>
@@ -265,9 +265,10 @@ std::unique_ptr<RetrievePlan>
 CreateRetrievePlanByExpr(SchemaPtr schema,
                          const void* serialized_expr_plan,
                          const int64_t size) {
-    proto::plan::PlanNode plan_node;
-    ParsePlanNodeProto(plan_node, serialized_expr_plan, size);
-    return ProtoParser(std::move(schema)).CreateRetrievePlan(plan_node);
+    auto plan_node = std::make_unique<proto::plan::PlanNode>();
+    ParsePlanNodeProto(*plan_node, serialized_expr_plan, size);
+    return ProtoParser(std::move(schema))
+        .CreateRetrievePlan(std::move(plan_node));
 }
 
 int64_t

internal/core/src/query/PlanProto.cpp

@@ -1031,6 +1031,138 @@ RedactMembershipFilterBlobs(google::protobuf::Message* message,
     }
 }
 
+using BloomBlobOwners =
+    std::unordered_map<const std::string*, std::shared_ptr<const std::string>>;
+
+// Move one Bloom body out of a uniquely owned protobuf tree before parsing.
+// The string object's address remains stable and is used only as a temporary
+// key; logical expressions take shared ownership of the moved string.
+static void
+ExtractBloomFilterBlob(planpb::BloomFilterExpr* bloom,
+                       BloomBlobOwners* owners) {
+    auto* source_blob = bloom->mutable_filter_blob();
+    const auto inserted =
+        owners
+            ->emplace(
+                source_blob,
+                std::make_shared<const std::string>(std::move(*source_blob)))
+            .second;
+    AssertInfo(inserted, "duplicate Bloom blob protobuf address");
+}
+
+// Follow only schema edges that can contain another Expr. In particular, Term
+// and JSONContains GenericValue lists are leaves, so large IN predicates do not
+// pay a reflection walk over every value while looking for Bloom expressions.
+static void
+ExtractBloomFilterBlobs(planpb::Expr* expr, BloomBlobOwners* owners) {
+    switch (expr->expr_case()) {
+        case planpb::Expr::kBloomFilterExpr:
+            ExtractBloomFilterBlob(expr->mutable_bloom_filter_expr(), owners);
+            return;
+        case planpb::Expr::kUnaryExpr: {
+            auto* unary = expr->mutable_unary_expr();
+            if (unary->has_child()) {
+                ExtractBloomFilterBlobs(unary->mutable_child(), owners);
+            }
+            return;
+        }
+        case planpb::Expr::kBinaryExpr: {
+            auto* binary = expr->mutable_binary_expr();
+            if (binary->has_left()) {
+                ExtractBloomFilterBlobs(binary->mutable_left(), owners);
+            }
+            if (binary->has_right()) {
+                ExtractBloomFilterBlobs(binary->mutable_right(), owners);
+            }
+            return;
+        }
+        case planpb::Expr::kBinaryArithExpr: {
+            auto* binary = expr->mutable_binary_arith_expr();
+            if (binary->has_left()) {
+                ExtractBloomFilterBlobs(binary->mutable_left(), owners);
+            }
+            if (binary->has_right()) {
+                ExtractBloomFilterBlobs(binary->mutable_right(), owners);
+            }
+            return;
+        }
+        case planpb::Expr::kCallExpr: {
+            auto* call = expr->mutable_call_expr();
+            for (int i = 0; i < call->function_parameters_size(); ++i) {
+                ExtractBloomFilterBlobs(call->mutable_function_parameters(i),
+                                        owners);
+            }
+            return;
+        }
+        case planpb::Expr::kRandomSampleExpr: {
+            auto* random_sample = expr->mutable_random_sample_expr();
+            if (random_sample->has_predicate()) {
+                ExtractBloomFilterBlobs(random_sample->mutable_predicate(),
+                                        owners);
+            }
+            return;
+        }
+        case planpb::Expr::kElementFilterExpr: {
+            auto* element_filter = expr->mutable_element_filter_expr();
+            if (element_filter->has_element_expr()) {
+                ExtractBloomFilterBlobs(element_filter->mutable_element_expr(),
+                                        owners);
+            }
+            if (element_filter->has_predicate()) {
+                ExtractBloomFilterBlobs(element_filter->mutable_predicate(),
+                                        owners);
+            }
+            return;
+        }
+        case planpb::Expr::kMatchExpr: {
+            auto* match = expr->mutable_match_expr();
+            if (match->has_predicate()) {
+                ExtractBloomFilterBlobs(match->mutable_predicate(), owners);
+            }
+            return;
+        }
+        default:
+            return;
+    }
+}
+
+static BloomBlobOwners
+ExtractBloomFilterBlobs(planpb::PlanNode* root) {
+    BloomBlobOwners owners;
+    switch (root->node_case()) {
+        case planpb::PlanNode::kVectorAnns: {
+            auto* vector_anns = root->mutable_vector_anns();
+            if (vector_anns->has_predicates()) {
+                ExtractBloomFilterBlobs(vector_anns->mutable_predicates(),
+                                        &owners);
+            }
+            break;
+        }
+        case planpb::PlanNode::kPredicates:
+            ExtractBloomFilterBlobs(root->mutable_predicates(), &owners);
+            break;
+        case planpb::PlanNode::kQuery: {
+            auto* query = root->mutable_query();
+            if (query->has_predicates()) {
+                ExtractBloomFilterBlobs(query->mutable_predicates(), &owners);
+            }
+            break;
+        }
+        default:
+            break;
+    }
+    return owners;
+}
+
+static BloomBlobOwners
+ExtractBloomFilterBlobs(planpb::ScoreFunction* root) {
+    BloomBlobOwners owners;
+    if (root->has_filter()) {
+        ExtractBloomFilterBlobs(root->mutable_filter(), &owners);
+    }
+    return owners;
+}
+
 std::string
 PlanProtoDebugString(const proto::plan::PlanNode& plan_node_proto) {
     const auto size = plan_node_proto.ByteSizeLong();
@@ -1067,6 +1199,11 @@ LogPlanProtoDebug(const char* what,
 std::unique_ptr<Plan>
 ProtoParser::CreatePlan(const proto::plan::PlanNode& plan_node_proto) {
     LogPlanProtoDebug("search", plan_node_proto);
+    return CreatePlanImpl(plan_node_proto);
+}
+
+std::unique_ptr<Plan>
+ProtoParser::CreatePlanImpl(const proto::plan::PlanNode& plan_node_proto) {
     auto plan = std::make_unique<Plan>(schema);
     plan->plan_node_ = PlanNodeFromProto(plan_node_proto);
     plan->tag2field_["$0"] = plan->plan_node_->search_info_.field_id_;
@@ -1086,9 +1223,25 @@ ProtoParser::CreatePlan(const proto::plan::PlanNode& plan_node_proto) {
     return plan;
 }
 
+std::unique_ptr<Plan>
+ProtoParser::CreatePlan(
+    std::unique_ptr<proto::plan::PlanNode> plan_node_proto) {
+    AssertInfo(plan_node_proto != nullptr, "plan node owner must not be null");
+    LogPlanProtoDebug("search", *plan_node_proto);
+    auto bloom_blob_owners = ExtractBloomFilterBlobs(plan_node_proto.get());
+    return ProtoParser(schema, std::move(bloom_blob_owners))
+        .CreatePlanImpl(*plan_node_proto);
+}
+
 std::unique_ptr<RetrievePlan>
 ProtoParser::CreateRetrievePlan(const proto::plan::PlanNode& plan_node_proto) {
     LogPlanProtoDebug("retrieve", plan_node_proto);
+    return CreateRetrievePlanImpl(plan_node_proto);
+}
+
+std::unique_ptr<RetrievePlan>
+ProtoParser::CreateRetrievePlanImpl(
+    const proto::plan::PlanNode& plan_node_proto) {
     auto retrieve_plan = std::make_unique<RetrievePlan>(schema);
     retrieve_plan->plan_node_ = RetrievePlanNodeFromProto(plan_node_proto);
     retrieve_plan->access_entries_ = CollectAccessFieldIDs(plan_node_proto);
@@ -1102,6 +1255,16 @@ ProtoParser::CreateRetrievePlan(const proto::plan::PlanNode& plan_node_proto) {
     return retrieve_plan;
 }
 
+std::unique_ptr<RetrievePlan>
+ProtoParser::CreateRetrievePlan(
+    std::unique_ptr<proto::plan::PlanNode> plan_node_proto) {
+    AssertInfo(plan_node_proto != nullptr, "plan node owner must not be null");
+    LogPlanProtoDebug("retrieve", *plan_node_proto);
+    auto bloom_blob_owners = ExtractBloomFilterBlobs(plan_node_proto.get());
+    return ProtoParser(schema, std::move(bloom_blob_owners))
+        .CreateRetrievePlanImpl(*plan_node_proto);
+}
+
 expr::TypedExprPtr
 ProtoParser::ParseUnaryRangeExprs(const proto::plan::UnaryRangeExpr& expr_pb) {
     auto& column_info = expr_pb.column_info();
@@ -1253,8 +1416,17 @@ ProtoParser::ParseBloomFilterExprs(
                       "data type: {}",
                       data_type);
     }
+    std::shared_ptr<const std::string> filter_blob;
+    if (const auto it =
+            bloom_blob_owners_.find(std::addressof(expr_pb.filter_blob()));
+        it != bloom_blob_owners_.end()) {
+        filter_blob = it->second;
+    } else {
+        filter_blob =
+            std::make_shared<const std::string>(expr_pb.filter_blob());
+    }
     return std::make_shared<expr::BloomFilterExpr>(
-        expr::ColumnInfo(column_info), expr_pb.filter_blob());
+        expr::ColumnInfo(column_info), std::move(filter_blob));
 }
 
 expr::TypedExprPtr
@@ -1605,6 +1777,14 @@ ProtoParser::ParseScorer(const proto::plan::ScoreFunction& function) {
     }
 }
 
+std::shared_ptr<rescores::Scorer>
+ProtoParser::ParseScorer(std::unique_ptr<proto::plan::ScoreFunction> function) {
+    AssertInfo(function != nullptr, "score function owner must not be null");
+    auto bloom_blob_owners = ExtractBloomFilterBlobs(function.get());
+    return ProtoParser(schema, std::move(bloom_blob_owners))
+        .ParseScorer(*function);
+}
+
 std::shared_ptr<plan::PlanNode>
 ProtoParser::ExtractFilterOnlyPlan(
     const std::shared_ptr<plan::PlanNode>& root_node) {

internal/core/src/query/PlanProto.h

@@ -15,6 +15,7 @@
 #include <functional>
 #include <memory>
 #include <string>
+#include <unordered_map>
 
 #include "Plan.h"
 #include "PlanNode.h"
@@ -55,9 +56,15 @@ class ProtoParser {
     std::unique_ptr<Plan>
     CreatePlan(const proto::plan::PlanNode& plan_node_proto);
 
+    std::unique_ptr<Plan>
+    CreatePlan(std::unique_ptr<proto::plan::PlanNode> plan_node_proto);
+
     std::unique_ptr<RetrievePlan>
     CreateRetrievePlan(const proto::plan::PlanNode& plan_node_proto);
 
+    std::unique_ptr<RetrievePlan>
+    CreateRetrievePlan(std::unique_ptr<proto::plan::PlanNode> plan_node_proto);
+
     expr::TypedExprPtr
     ParseExprs(const proto::plan::Expr& expr_pb,
                TypeCheckFunction type_check = TypeIsBool);
@@ -65,6 +72,9 @@ class ProtoParser {
     std::shared_ptr<rescores::Scorer>
     ParseScorer(const proto::plan::ScoreFunction& function);
 
+    std::shared_ptr<rescores::Scorer>
+    ParseScorer(std::unique_ptr<proto::plan::ScoreFunction> function);
+
     // Extract only filter-related nodes (FilterBitsNode, ElementFilterBitsNode)
     // from a plan tree, avoiding VectorSearchNode, SearchGroupByNode, etc.
     // Returns MvccNode only if no pre-filter nodes are found.
@@ -72,6 +82,21 @@ class ProtoParser {
     ExtractFilterOnlyPlan(const std::shared_ptr<plan::PlanNode>& root_node);
 
  private:
+    using BloomBlobOwners =
+        std::unordered_map<const std::string*,
+                           std::shared_ptr<const std::string>>;
+
+    ProtoParser(SchemaPtr schema, BloomBlobOwners bloom_blob_owners)
+        : schema(std::move(schema)),
+          bloom_blob_owners_(std::move(bloom_blob_owners)) {
+    }
+
+    std::unique_ptr<Plan>
+    CreatePlanImpl(const proto::plan::PlanNode& plan_node_proto);
+
+    std::unique_ptr<RetrievePlan>
+    CreateRetrievePlanImpl(const proto::plan::PlanNode& plan_node_proto);
+
     expr::TypedExprPtr
     CreateAlwaysTrueExprs();
 
@@ -141,6 +166,12 @@ class ProtoParser {
 
  private:
     const SchemaPtr schema;
+    // Owner-bound entry points move each Bloom string out of their uniquely
+    // owned protobuf root. The map ties the original string object's address
+    // to its new shared owner only while the root is being parsed; logical
+    // expressions retain the shared strings they consume. Legacy const-ref
+    // entry points leave this empty and keep their safe-copy behavior.
+    const BloomBlobOwners bloom_blob_owners_;
 };
 
 }  // namespace milvus::query

internal/core/src/query/PlanProtoTest.cpp

@@ -31,6 +31,7 @@
 #include "knowhere/comp/index_param.h"
 #include "pb/plan.pb.h"
 #include "pb/schema.pb.h"
+#include "plan/PlanNode.h"
 #include "query/Plan.h"
 #include "query/PlanProto.h"
 
@@ -126,6 +127,23 @@ BuildSearchPlanNode(float search_topk_ratio,
     return plan_node;
 }
 
+std::shared_ptr<milvus::plan::FilterBitsNode>
+FindFilterBitsNode(const std::shared_ptr<milvus::plan::PlanNode>& node) {
+    if (node == nullptr) {
+        return nullptr;
+    }
+    if (auto filter =
+            std::dynamic_pointer_cast<milvus::plan::FilterBitsNode>(node)) {
+        return filter;
+    }
+    for (const auto& source : node->sources()) {
+        if (auto filter = FindFilterBitsNode(source)) {
+            return filter;
+        }
+    }
+    return nullptr;
+}
+
 }  // namespace
 
 TEST(PlanProto, NotSetUnsupported) {
@@ -307,6 +325,201 @@ TEST(PlanProto, SupportsMembershipFiltersInScorers) {
     }
 }
 
+TEST(PlanProto, BloomBlobMovesFromOwnedSearchPlanNode) {
+    using namespace milvus;
+    namespace planpb = milvus::proto::plan;
+
+    auto schema = BuildSchema();
+    const auto vector_field_id = schema->get_field_id(FieldName("fakevec"));
+    const auto scalar_field_id = schema->get_field_id(FieldName("age"));
+    auto plan_node = std::make_unique<planpb::PlanNode>(
+        BuildSearchPlanNode(1.0F, 1.0F, vector_field_id));
+    auto* bloom = plan_node->mutable_vector_anns()
+                      ->mutable_predicates()
+                      ->mutable_bloom_filter_expr();
+    bloom->mutable_column_info()->set_field_id(scalar_field_id.get());
+    bloom->mutable_column_info()->set_data_type(proto::schema::DataType::Int64);
+    bloom->set_filter_blob(BuildEmptyMbf1());
+
+    const auto protobuf_blob_object =
+        reinterpret_cast<uintptr_t>(std::addressof(bloom->filter_blob()));
+    const auto* protobuf_blob_data = bloom->filter_blob().data();
+    auto parsed_plan =
+        query::ProtoParser(schema).CreatePlan(std::move(plan_node));
+    auto filter_node = FindFilterBitsNode(parsed_plan->plan_node_->plannodes_);
+    ASSERT_NE(filter_node, nullptr);
+    auto bloom_expr = std::dynamic_pointer_cast<const expr::BloomFilterExpr>(
+        filter_node->filter());
+    ASSERT_NE(bloom_expr, nullptr);
+    EXPECT_NE(reinterpret_cast<uintptr_t>(bloom_expr->filter_blob_.get()),
+              protobuf_blob_object);
+    EXPECT_EQ(bloom_expr->filter_blob_->data(), protobuf_blob_data);
+    std::weak_ptr<const std::string> weak_blob = bloom_expr->filter_blob_;
+
+    EXPECT_EQ(plan_node, nullptr);
+    EXPECT_FALSE(weak_blob.expired());
+    EXPECT_EQ(*bloom_expr->filter_blob_, BuildEmptyMbf1());
+
+    parsed_plan.reset();
+    filter_node.reset();
+    bloom_expr.reset();
+    EXPECT_TRUE(weak_blob.expired());
+}
+
+TEST(PlanProto, BloomBlobMovesThroughNestedLogicBesideLargeTerm) {
+    using namespace milvus;
+    namespace planpb = milvus::proto::plan;
+
+    auto schema = BuildSchema();
+    const auto vector_field_id = schema->get_field_id(FieldName("fakevec"));
+    const auto scalar_field_id = schema->get_field_id(FieldName("age"));
+    auto plan_node = std::make_unique<planpb::PlanNode>(
+        BuildSearchPlanNode(1.0F, 1.0F, vector_field_id));
+
+    // OR(NOT(Bloom), AND(large IN, Bloom)) exercises nested recursive Expr
+    // carriers while keeping the GenericValue list as a leaf for extraction.
+    auto* outer = plan_node->mutable_vector_anns()
+                      ->mutable_predicates()
+                      ->mutable_binary_expr();
+    outer->set_op(planpb::BinaryExpr::LogicalOr);
+    auto* unary = outer->mutable_left()->mutable_unary_expr();
+    unary->set_op(planpb::UnaryExpr::Not);
+    auto* left_bloom = unary->mutable_child()->mutable_bloom_filter_expr();
+    left_bloom->mutable_column_info()->set_field_id(scalar_field_id.get());
+    left_bloom->mutable_column_info()->set_data_type(
+        proto::schema::DataType::Int64);
+    left_bloom->set_filter_blob(BuildEmptyMbf1());
+
+    auto* inner = outer->mutable_right()->mutable_binary_expr();
+    inner->set_op(planpb::BinaryExpr::LogicalAnd);
+    auto* term = inner->mutable_left()->mutable_term_expr();
+    term->mutable_column_info()->set_field_id(scalar_field_id.get());
+    term->mutable_column_info()->set_data_type(proto::schema::DataType::Int64);
+    constexpr int kTermValueCount = 4096;
+    for (int i = 0; i < kTermValueCount; ++i) {
+        term->add_values()->set_int64_val(i);
+    }
+    auto* right_bloom = inner->mutable_right()->mutable_bloom_filter_expr();
+    right_bloom->mutable_column_info()->set_field_id(scalar_field_id.get());
+    right_bloom->mutable_column_info()->set_data_type(
+        proto::schema::DataType::Int64);
+    right_bloom->set_filter_blob(BuildEmptyMbf1());
+
+    const auto* left_blob_data = left_bloom->filter_blob().data();
+    const auto* right_blob_data = right_bloom->filter_blob().data();
+    auto parsed_plan =
+        query::ProtoParser(schema).CreatePlan(std::move(plan_node));
+    auto filter_node = FindFilterBitsNode(parsed_plan->plan_node_->plannodes_);
+    ASSERT_NE(filter_node, nullptr);
+
+    auto parsed_outer =
+        std::dynamic_pointer_cast<const expr::LogicalBinaryExpr>(
+            filter_node->filter());
+    ASSERT_NE(parsed_outer, nullptr);
+    ASSERT_EQ(parsed_outer->inputs().size(), 2);
+    auto parsed_unary = std::dynamic_pointer_cast<const expr::LogicalUnaryExpr>(
+        parsed_outer->inputs()[0]);
+    ASSERT_NE(parsed_unary, nullptr);
+    ASSERT_EQ(parsed_unary->inputs().size(), 1);
+    auto parsed_left_bloom =
+        std::dynamic_pointer_cast<const expr::BloomFilterExpr>(
+            parsed_unary->inputs()[0]);
+    ASSERT_NE(parsed_left_bloom, nullptr);
+
+    auto parsed_inner =
+        std::dynamic_pointer_cast<const expr::LogicalBinaryExpr>(
+            parsed_outer->inputs()[1]);
+    ASSERT_NE(parsed_inner, nullptr);
+    ASSERT_EQ(parsed_inner->inputs().size(), 2);
+    auto parsed_term = std::dynamic_pointer_cast<const expr::TermFilterExpr>(
+        parsed_inner->inputs()[0]);
+    ASSERT_NE(parsed_term, nullptr);
+    ASSERT_EQ(parsed_term->vals_.size(), kTermValueCount);
+    EXPECT_EQ(parsed_term->vals_.front().int64_val(), 0);
+    EXPECT_EQ(parsed_term->vals_.back().int64_val(), kTermValueCount - 1);
+    auto parsed_right_bloom =
+        std::dynamic_pointer_cast<const expr::BloomFilterExpr>(
+            parsed_inner->inputs()[1]);
+    ASSERT_NE(parsed_right_bloom, nullptr);
+
+    EXPECT_EQ(parsed_left_bloom->filter_blob_->data(), left_blob_data);
+    EXPECT_EQ(parsed_right_bloom->filter_blob_->data(), right_blob_data);
+}
+
+TEST(PlanProto, BloomBlobMovesFromOwnedRetrievePlanNode) {
+    using namespace milvus;
+    namespace planpb = milvus::proto::plan;
+
+    auto schema = BuildSchema();
+    const auto scalar_field_id = schema->get_field_id(FieldName("age"));
+    auto plan_node = std::make_unique<planpb::PlanNode>();
+    auto* bloom = plan_node->mutable_query()
+                      ->mutable_predicates()
+                      ->mutable_bloom_filter_expr();
+    bloom->mutable_column_info()->set_field_id(scalar_field_id.get());
+    bloom->mutable_column_info()->set_data_type(proto::schema::DataType::Int64);
+    bloom->set_filter_blob(BuildEmptyMbf1());
+
+    const auto protobuf_blob_object =
+        reinterpret_cast<uintptr_t>(std::addressof(bloom->filter_blob()));
+    const auto* protobuf_blob_data = bloom->filter_blob().data();
+    auto parsed_plan =
+        query::ProtoParser(schema).CreateRetrievePlan(std::move(plan_node));
+    auto filter_node = FindFilterBitsNode(parsed_plan->plan_node_->plannodes_);
+    ASSERT_NE(filter_node, nullptr);
+    auto bloom_expr = std::dynamic_pointer_cast<const expr::BloomFilterExpr>(
+        filter_node->filter());
+    ASSERT_NE(bloom_expr, nullptr);
+    EXPECT_NE(reinterpret_cast<uintptr_t>(bloom_expr->filter_blob_.get()),
+              protobuf_blob_object);
+    EXPECT_EQ(bloom_expr->filter_blob_->data(), protobuf_blob_data);
+    std::weak_ptr<const std::string> weak_blob = bloom_expr->filter_blob_;
+
+    EXPECT_EQ(plan_node, nullptr);
+    EXPECT_FALSE(weak_blob.expired());
+    EXPECT_EQ(*bloom_expr->filter_blob_, BuildEmptyMbf1());
+
+    parsed_plan.reset();
+    filter_node.reset();
+    bloom_expr.reset();
+    EXPECT_TRUE(weak_blob.expired());
+}
+
+TEST(PlanProto, BloomBlobMovesFromOwnedScoreFunction) {
+    using namespace milvus;
+    namespace planpb = milvus::proto::plan;
+
+    auto schema = BuildSchema();
+    const auto scalar_field_id = schema->get_field_id(FieldName("age"));
+    auto function = std::make_unique<planpb::ScoreFunction>();
+    function->set_type(planpb::FunctionTypeWeight);
+    function->set_weight(2.0F);
+    auto* bloom = function->mutable_filter()->mutable_bloom_filter_expr();
+    bloom->mutable_column_info()->set_field_id(scalar_field_id.get());
+    bloom->mutable_column_info()->set_data_type(proto::schema::DataType::Int64);
+    bloom->set_filter_blob(BuildEmptyMbf1());
+
+    const auto protobuf_blob_object =
+        reinterpret_cast<uintptr_t>(std::addressof(bloom->filter_blob()));
+    const auto* protobuf_blob_data = bloom->filter_blob().data();
+    auto scorer = query::ProtoParser(schema).ParseScorer(std::move(function));
+    auto bloom_expr = std::dynamic_pointer_cast<const expr::BloomFilterExpr>(
+        scorer->filter());
+    ASSERT_NE(bloom_expr, nullptr);
+    EXPECT_NE(reinterpret_cast<uintptr_t>(bloom_expr->filter_blob_.get()),
+              protobuf_blob_object);
+    EXPECT_EQ(bloom_expr->filter_blob_->data(), protobuf_blob_data);
+    std::weak_ptr<const std::string> weak_blob = bloom_expr->filter_blob_;
+
+    EXPECT_EQ(function, nullptr);
+    EXPECT_FALSE(weak_blob.expired());
+    EXPECT_EQ(*bloom_expr->filter_blob_, BuildEmptyMbf1());
+
+    scorer.reset();
+    bloom_expr.reset();
+    EXPECT_TRUE(weak_blob.expired());
+}
+
 TEST(PlanProto, RejectsGlobalRefineRatiosBelowOne) {
     using namespace milvus::query;
 

internal/core/src/segcore/boost_score.cpp

@@ -98,13 +98,14 @@ std::shared_ptr<milvus::rescores::Scorer>
 ParseScorer(milvus::query::Plan* plan,
             const void* serialized_score_function,
             int64_t serialized_score_function_size) {
-    milvus::proto::plan::ScoreFunction score_function;
-    auto ok = score_function.ParseFromArray(serialized_score_function,
-                                            serialized_score_function_size);
+    auto score_function =
+        std::make_unique<milvus::proto::plan::ScoreFunction>();
+    auto ok = score_function->ParseFromArray(serialized_score_function,
+                                             serialized_score_function_size);
     AssertInfo(ok, "failed to parse score function");
 
     milvus::query::ProtoParser parser(plan->schema_);
-    return parser.ParseScorer(score_function);
+    return parser.ParseScorer(std::move(score_function));
 }
 
 OffsetChunks

internal/core/unittest/test_bloom_filter_expr.cpp

@@ -616,7 +616,7 @@ TEST(BloomFilterExprTest, ProtoParserDispatchAndTypeCheck) {
             std::dynamic_pointer_cast<const expr::BloomFilterExpr>(parsed);
         ASSERT_NE(bf_expr, nullptr);
         EXPECT_EQ(bf_expr->column_.field_id_, fid);
-        EXPECT_EQ(bf_expr->filter_blob_, blob);
+        EXPECT_EQ(*bf_expr->filter_blob_, blob);
     }
 
     // JSON parses too, and the nested path survives into the logical expr.
@@ -945,20 +945,21 @@ TEST_F(BloomFilterExprEvalTest, VarcharGrowingAndSealed) {
 namespace {
 
 enum class JsonVariant : int {
-    kIntMember = 0,   // {"uid": 5}          member of the int blob
-    kIntOther,        // {"uid": 7}          valid non-member
-    kDoubleIntegral,  // {"uid": 5.0}        strictly typed: NOT a member
-    kDoubleFraction,  // {"uid": 5.5}        definite non-member, valid
-    kStrAlice,        // {"uid": "alice"}    string probe
-    kStrFive,         // {"uid": "5"}        string, must not alias int64 5
-    kBool,            // {"uid": true}       no probe value
-    kNull,            // {"uid": null}       no probe value
-    kMissing,         // {"other": 1}        no probe value
-    kObject,          // {"uid": {"a": 1}}   no probe value
-    kArray,           // {"uid": [5]}        no probe value
-    kHugeUint,        // {"uid": 2^64-1}     definite non-member, valid
-    kIntMin,          // {"uid": INT64_MIN}  int64 member probe
-    kRowNull,         // whole-row NULL
+    kIntMember = 0,    // {"uid": 5}          member of the int blob
+    kIntOther,         // {"uid": 7}          valid non-member
+    kDoubleIntegral,   // {"uid": 5.0}        strictly typed: NOT a member
+    kDoubleFraction,   // {"uid": 5.5}        definite non-member, valid
+    kStrAlice,         // {"uid": "alice"}    string probe
+    kStrEscapedAlice,  // {"uid": "a\u006cice"} decoded string probe
+    kStrFive,          // {"uid": "5"}        string, must not alias int64 5
+    kBool,             // {"uid": true}       no probe value
+    kNull,             // {"uid": null}       no probe value
+    kMissing,          // {"other": 1}        no probe value
+    kObject,           // {"uid": {"a": 1}}   no probe value
+    kArray,            // {"uid": [5]}        no probe value
+    kHugeUint,         // {"uid": 2^64-1}     definite non-member, valid
+    kIntMin,           // {"uid": INT64_MIN}  int64 member probe
+    kRowNull,          // whole-row NULL
     kVariantCount,
 };
 
@@ -982,6 +983,8 @@ JsonRowFor(JsonVariant v) {
             return R"({"uid": 5.5})";
         case JsonVariant::kStrAlice:
             return R"({"uid": "alice"})";
+        case JsonVariant::kStrEscapedAlice:
+            return R"({"uid": "a\u006cice"})";
         case JsonVariant::kStrFive:
             return R"({"uid": "5"})";
         case JsonVariant::kBool:
@@ -1025,6 +1028,7 @@ JsonReference(const SplitBlockBloomFilterView& view, JsonVariant v) {
         case JsonVariant::kDoubleFraction:
             return {true, false};
         case JsonVariant::kStrAlice:
+        case JsonVariant::kStrEscapedAlice:
             return {true, view.TestBytes("alice", 5)};
         case JsonVariant::kStrFive:
             return {true, view.TestBytes("5", 1)};
@@ -1252,11 +1256,11 @@ TEST_F(BloomFilterExprEvalTest, JsonPathOffsetInput) {
     ASSERT_TRUE(view.TestInt64(5));
 
     // Non-contiguous, out-of-order candidates spanning the whole segment,
-    // covering member/non-member/gap/whole-row-NULL variants (1777 % 14 == 13
-    // is a whole-row NULL; 1999 % 14 == 11 is the huge uint64).
+    // covering member/non-member/gap/whole-row-NULL variants (1784 % 15 == 14
+    // is a whole-row NULL; 1992 % 15 == 12 is the huge uint64).
     milvus::exec::OffsetVector offsets;
     for (auto o : std::vector<int32_t>{
-             1999, 3, 402, 57, 1500, 800, 17, 1234, 999, 6, 1777, 450}) {
+             1992, 3, 402, 57, 1500, 800, 17, 1234, 999, 6, 1784, 450}) {
         offsets.emplace_back(o);
     }
 
@@ -1281,7 +1285,7 @@ TEST_F(BloomFilterExprEvalTest, JsonPathOffsetInput) {
                 << "candidate k=" << k << " row=" << row;
         }
 
-        // Candidate 0 (row 1999, huge uint64 -> false) vs sequential row 0
+        // Candidate 0 (row 1992, huge uint64 -> false) vs sequential row 0
         // (int member 5 -> true): a first-N sequential scan provably differs,
         // so this test depends on the offset-input path.
         const auto [v0, p0] = JsonReference(view, JsonVariantAt(0));

internal/parser/planparserv2/bloom_match.go

@@ -115,10 +115,11 @@ func validateBloomFilterBlob(blob []byte) ([]byte, error) {
 	// body, ~half the member capacity). The 128 MiB MBF1 num_blocks format cap
 	// (checked in validateMBF1Envelope) remains the hard ceiling above this.
 	//
-	// A separate proxy.maxMembershipFilterPlanSize budget bounds the aggregate
-	// serialized plans before proto.Marshal. This per-blob gate remains necessary:
-	// it rejects one oversized filter at the input boundary, while the aggregate
-	// gate limits repeated otherwise-valid occurrences across the request.
+	// A separate proxy.maxMembershipFilterPlanSize budget bounds aggregate
+	// membership resources across the request: serialized plan bytes for every
+	// filter and estimated decoded bytes for Roaring filters. This per-blob gate
+	// remains necessary: it rejects one oversized filter at the input boundary,
+	// while the request-wide gate limits repeated otherwise-valid occurrences.
 	if maxSize := paramtable.Get().ProxyCfg.MaxMembershipFilterSize.GetAsInt(); len(blob) > maxSize+mbf1HeaderSize {
 		return nil, merr.WrapErrParameterInvalidMsg(
 			"membership_match bloom filter blob body is %d bytes, exceeding proxy.maxMembershipFilterSize (%d)%s",

internal/parser/planparserv2/fill_expression_value.go

@@ -16,11 +16,13 @@ import (
 // must be shared by every expression in one request: the main predicate, each
 // hybrid sub-request predicate, and each function-scorer filter.
 //
-// Scope matters. proxy.maxMembershipFilterPlanSize is the per-request ceiling,
-// but plan construction parses one expression per sub-request plus one per
-// scorer. Threading one budget through the whole request rejects repeated blob
-// occurrences before materialization and avoids re-validating the same Roaring
-// template once per scorer or hybrid sub-request.
+// Scope matters. proxy.maxMembershipFilterPlanSize is the per-request ceiling
+// for both aggregate serialized membership-plan bytes and aggregate estimated
+// Roaring decoded bytes (checked independently), but plan construction parses
+// one expression per sub-request plus one per scorer. Threading one budget
+// through the whole request rejects repeated blob occurrences before
+// materialization and avoids re-validating the same Roaring template once per
+// scorer or hybrid sub-request.
 //
 // A nil *MembershipPreflightBudget is not shared state; callers that parse a
 // standalone expression can pass nil and get single-expression scope.
@@ -252,14 +254,15 @@ func preflightMembershipFilterValues(
 		}
 		occurrences := roaringOccurrenceCounts[template.name]
 		cost := validated.summary.EstimatedDecodedBytes
+		maxDecodedBytes := uint64(budget.maxPlanSize)
 		remaining := uint64(0)
-		if budget.aggregateDecodedBytes <= roaringfilter.MaxEstimatedDecodedBytes {
-			remaining = roaringfilter.MaxEstimatedDecodedBytes - budget.aggregateDecodedBytes
+		if budget.aggregateDecodedBytes <= maxDecodedBytes {
+			remaining = maxDecodedBytes - budget.aggregateDecodedBytes
 		}
 		if cost != 0 && occurrences > remaining/cost {
 			return nil, merr.WrapErrParameterTooLarge(fmt.Sprintf(
-				"aggregate membership_match roaring estimated decoded size exceeds maximum before plan materialization: %d + %d*%d > %d bytes",
-				budget.aggregateDecodedBytes, occurrences, cost, roaringfilter.MaxEstimatedDecodedBytes))
+				"aggregate membership_match roaring estimated decoded size exceeds proxy.maxMembershipFilterPlanSize before plan materialization: %d + %d*%d > %d bytes",
+				budget.aggregateDecodedBytes, occurrences, cost, maxDecodedBytes))
 		}
 		budget.aggregateDecodedBytes += occurrences * cost
 		ctx.validatedRoaringBlobs[template.name] = validated
@@ -520,7 +523,13 @@ func FillBinaryArithOpEvalRangeExpressionValue(expr *planpb.BinaryArithOpEvalRan
 			}
 		}
 
+		if IsBytes(operand) {
+			return bytesTemplateValueError()
+		}
 		operandExpr := toValueExpr(operand)
+		if operandExpr == nil {
+			return merr.WrapErrQueryPlanMsg("unsupported arithmetic operand")
+		}
 		lDataType, rDataType := expr.GetColumnInfo().GetDataType(), operandExpr.dataType
 		if typeutil.IsArrayType(expr.GetColumnInfo().GetDataType()) {
 			lDataType = expr.GetColumnInfo().GetElementType()

internal/parser/planparserv2/fill_expression_value_test.go

@@ -591,6 +591,9 @@ func (s *FillExpressionValueSuite) TestBinaryArithOpEvalRange() {
 	})
 
 	s.Run("failed case", func() {
+		bytesTemplate := &schemapb.TemplateValue{
+			Val: &schemapb.TemplateValue_BytesVal{BytesVal: []byte("MBF1")},
+		}
 		testcases := []testcase{
 			{`Int64Field + 6 == 12.5`, nil},
 			{`Int64Field - {offset} == {target}`, map[string]*schemapb.TemplateValue{
@@ -626,6 +629,45 @@ func (s *FillExpressionValueSuite) TestBinaryArithOpEvalRange() {
 					generateTemplateArrayValue(schemapb.DataType_Int64, []int64{int64(1), int64(2), int64(3)})),
 				"target": generateTemplateValue(schemapb.DataType_Int64, int64(5)),
 			}},
+			{`Int64Field + {filter} > 0`, map[string]*schemapb.TemplateValue{
+				"filter": bytesTemplate,
+			}},
+			{`0 < Int64Field + {filter}`, map[string]*schemapb.TemplateValue{
+				"filter": bytesTemplate,
+			}},
+			{`{filter} + Int64Field > 0`, map[string]*schemapb.TemplateValue{
+				"filter": bytesTemplate,
+			}},
+			{`Int64Field - {filter} > 0`, map[string]*schemapb.TemplateValue{
+				"filter": bytesTemplate,
+			}},
+			{`Int64Field * {filter} > 0`, map[string]*schemapb.TemplateValue{
+				"filter": bytesTemplate,
+			}},
+			{`{filter} * Int64Field > 0`, map[string]*schemapb.TemplateValue{
+				"filter": bytesTemplate,
+			}},
+			{`Int64Field / {filter} > 0`, map[string]*schemapb.TemplateValue{
+				"filter": bytesTemplate,
+			}},
+			{`Int64Field % {filter} > 0`, map[string]*schemapb.TemplateValue{
+				"filter": bytesTemplate,
+			}},
+			{`(Int64Field & {filter}) > 0`, map[string]*schemapb.TemplateValue{
+				"filter": bytesTemplate,
+			}},
+			{`(Int64Field | {filter}) > 0`, map[string]*schemapb.TemplateValue{
+				"filter": bytesTemplate,
+			}},
+			{`(Int64Field ^ {filter}) > 0`, map[string]*schemapb.TemplateValue{
+				"filter": bytesTemplate,
+			}},
+			{`(Int64Field << {filter}) > 0`, map[string]*schemapb.TemplateValue{
+				"filter": bytesTemplate,
+			}},
+			{`(Int64Field >> {filter}) > 0`, map[string]*schemapb.TemplateValue{
+				"filter": bytesTemplate,
+			}},
 			{`array_length(ArrayField) == {length}`, map[string]*schemapb.TemplateValue{
 				"length": generateTemplateValue(schemapb.DataType_String, "abc"),
 			}},

internal/parser/planparserv2/membership_filter.go

@@ -49,33 +49,10 @@ const (
 	membershipRoaring
 )
 
-// membershipFilterSpec records the per-kind properties that control flow needs.
-// Everything else about a kind stays in its format validator; only behavior
-// switches belong here.
-type membershipFilterSpec struct {
-	kind membershipKind
-	// allowInDelete reports whether the kind is safe for delete expressions:
-	// false positives would remove rows outside the caller's set, so only the
-	// exact kinds are allowed (design doc 20260707-bloom-filter-expression).
-	allowInDelete bool
-}
-
-// membershipSpecByName covers the accepted surface name. The unified
-// entry has kind membershipUnknown: its kind comes from the blob magic at fill
-// time, and its allowInDelete default is the conservative answer for any code
-// path that must decide before the blob is inspected.
-var membershipSpecByName = map[string]membershipFilterSpec{
-	MembershipMatchFunctionName: {
-		kind:          membershipUnknown,
-		allowInDelete: false,
-	},
-}
-
 // isMembershipFunctionName reports whether name is the membership-filter
 // surface function handled by the unified skeleton.
 func isMembershipFunctionName(name string) bool {
-	_, ok := membershipSpecByName[name]
-	return ok
+	return name == MembershipMatchFunctionName
 }
 
 // sniffMembershipKind identifies the blob format from its magic header. Both
@@ -375,15 +352,8 @@ func hasDeleteUnsafeMembershipFilterExpr(expr *planpb.Expr) bool {
 		case *planpb.Expr_RoaringFilterExpr:
 			return false
 		case *planpb.Expr_CallExpr:
-			spec, ok := membershipSpecByName[e.CallExpr.GetFunctionName()]
-			if !ok {
-				return false
-			}
-			if spec.kind == membershipUnknown {
-				// Deferred unified call: kind unknown here, fail closed.
-				return true
-			}
-			return !spec.allowInDelete
+			// Deferred unified call: kind unknown here, fail closed.
+			return isMembershipFunctionName(e.CallExpr.GetFunctionName())
 		default:
 			return false
 		}

internal/parser/planparserv2/membership_match_test.go

@@ -32,7 +32,6 @@ import (
 	"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/paramtable"
-	"github.com/milvus-io/milvus/pkg/v3/util/roaringfilter"
 	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
 )
 
@@ -299,19 +298,19 @@ func TestMembershipMatchElementLevelGuards(t *testing.T) {
 func TestMembershipMatchPreflightChargesUnifiedOccurrences(t *testing.T) {
 	helper := newTestSchemaHelper(t)
 	pt := paramtable.Get()
-	_, blob := roaringBytesTemplate(t, 1, 2, 3)
-	values := map[string]*schemapb.TemplateValue{"rb": bytesTemplate(blob)}
-	body := len(blob) - roaringfilter.HeaderSize
+	template, blob := bloomBytesTemplate(t, 0.001, 1, 2, 3)
+	values := map[string]*schemapb.TemplateValue{"bf": template}
+	body := len(blob) - mbf1HeaderSize
 
 	// One occurrence fits; two occurrences of the same body exceed it.
 	pt.Save(pt.ProxyCfg.MaxMembershipFilterPlanSize.Key, strconv.Itoa(2*body-1))
 	defer pt.Reset(pt.ProxyCfg.MaxMembershipFilterPlanSize.Key)
 
-	_, err := ParseExpr(helper, "membership_match(Int64Field, {rb})", values)
+	_, err := ParseExpr(helper, "membership_match(Int64Field, {bf})", values)
 	require.NoError(t, err, "one occurrence fits")
 
 	_, err = ParseExpr(helper,
-		"membership_match(Int64Field, {rb}) and membership_match(Int64Field, {rb})", values)
+		"membership_match(Int64Field, {bf}) and membership_match(Int64Field, {bf})", values)
 	require.ErrorIs(t, err, merr.ErrParameterTooLarge)
 	require.ErrorContains(t, err, "before plan materialization")
 }

internal/parser/planparserv2/roaring_match.go

@@ -73,10 +73,10 @@ func validateRoaringBitmapBlob(blob []byte) (validatedRoaringBitmapBlob, error)
 	// top. Checked before Validate so an oversized blob is rejected without
 	// decoding it.
 	//
-	// A separate proxy.maxMembershipFilterPlanSize budget bounds the aggregate
-	// serialized plans before proto.Marshal. This per-blob gate rejects one
-	// oversized bitmap before decode; the aggregate gate limits repeated
-	// otherwise-valid occurrences across the request.
+	// A separate proxy.maxMembershipFilterPlanSize budget bounds both aggregate
+	// serialized plan bytes and aggregate estimated decoded bytes across the
+	// request. This per-blob gate rejects one oversized bitmap before decode;
+	// the request-wide gate limits repeated otherwise-valid occurrences.
 	if maxSize := paramtable.Get().ProxyCfg.MaxMembershipFilterSize.GetAsInt(); len(blob) > maxSize+roaringfilter.HeaderSize {
 		bodySize := len(blob) - roaringfilter.HeaderSize
 		if bodySize < 0 {

internal/parser/planparserv2/roaring_match_test.go

@@ -434,22 +434,68 @@ func TestRoaringMatchPreflightCachesValidationByTemplateName(t *testing.T) {
 
 func TestRoaringMatchPreflightBudgetsAggregateDecodedBytes(t *testing.T) {
 	helper := newTestSchemaHelper(t)
-	blob := compactHighContainerRoaringBlob(170_000)
+	blob := compactHighContainerRoaringBlob(220_000)
 	summary, err := serverroaring.Validate(blob)
 	require.NoError(t, err)
-	require.LessOrEqual(t, summary.EstimatedDecodedBytes, uint64(serverroaring.MaxEstimatedDecodedBytes))
-	require.Greater(t, summary.EstimatedDecodedBytes*2, uint64(serverroaring.MaxEstimatedDecodedBytes))
+	require.Greater(t, summary.EstimatedDecodedBytes, uint64(40*1024*1024))
+	require.Less(t, summary.EstimatedDecodedBytes, uint64(44*1024*1024))
+	require.LessOrEqual(t, summary.EstimatedDecodedBytes, uint64(serverroaring.MaxEstimatedDecodedBytes),
+		"each bitmap must still satisfy the fixed 64 MiB per-bitmap admission")
+	require.LessOrEqual(t, summary.EstimatedDecodedBytes*2,
+		uint64(paramtable.DefaultMaxMembershipFilterPlanSize))
+	require.Greater(t, summary.EstimatedDecodedBytes*3,
+		uint64(paramtable.DefaultMaxMembershipFilterPlanSize))
 
 	values := map[string]*schemapb.TemplateValue{"rb": bytesTemplate(blob)}
-	_, err = ParseExpr(helper, "membership_match(Int64Field, {rb}, type=roaring)", values)
-	require.NoError(t, err, "one admitted bitmap must stay valid")
+	const occurrence = "membership_match(Int64Field, {rb}, type=roaring)"
+	_, err = ParseExpr(helper, occurrence+" and "+occurrence, values)
+	require.NoError(t, err, "two approximately 40 MiB decoded bitmaps fit the default 128 MiB request budget")
 
-	_, err = ParseExpr(helper,
-		"membership_match(Int64Field, {rb}, type=roaring) and membership_match(Int64Field, {rb}, type=roaring)",
-		values)
+	_, err = ParseExpr(helper, occurrence+" and "+occurrence+" and "+occurrence, values)
 	require.ErrorIs(t, err, merr.ErrParameterTooLarge)
 	require.ErrorContains(t, err, "estimated decoded size")
+	require.ErrorContains(t, err, "proxy.maxMembershipFilterPlanSize")
 	require.ErrorContains(t, err, "before plan materialization")
+
+	t.Run("custom plan size controls decoded aggregate", func(t *testing.T) {
+		pt := paramtable.Get()
+		configuredLimit := summary.EstimatedDecodedBytes*2 - 1
+		require.LessOrEqual(t, uint64((len(blob)-roaringfilter.HeaderSize)*2), configuredLimit,
+			"the serialized-body preflight must fit so this exercises the decoded-total gate")
+		pt.Save(pt.ProxyCfg.MaxMembershipFilterPlanSize.Key,
+			strconv.FormatUint(configuredLimit, 10))
+		defer pt.Reset(pt.ProxyCfg.MaxMembershipFilterPlanSize.Key)
+
+		_, err := ParseExpr(helper, occurrence+" and "+occurrence, values)
+		require.ErrorIs(t, err, merr.ErrParameterTooLarge)
+		require.ErrorContains(t, err, "estimated decoded size")
+		require.ErrorContains(t, err, "proxy.maxMembershipFilterPlanSize")
+	})
+}
+
+func TestRoaringMatchDecodedBudgetIsSharedAcrossParses(t *testing.T) {
+	helper := newTestSchemaHelper(t)
+	blob := compactHighContainerRoaringBlob(220_000)
+	summary, err := serverroaring.Validate(blob)
+	require.NoError(t, err)
+
+	pt := paramtable.Get()
+	pt.Save(pt.ProxyCfg.MaxMembershipFilterPlanSize.Key,
+		strconv.FormatUint(summary.EstimatedDecodedBytes*2, 10))
+	defer pt.Reset(pt.ProxyCfg.MaxMembershipFilterPlanSize.Key)
+
+	values := map[string]*schemapb.TemplateValue{"rb": bytesTemplate(blob)}
+	const expr = "membership_match(Int64Field, {rb}, type=roaring)"
+	visitorArgs := &ParserVisitorArgs{MembershipBudget: NewMembershipPreflightBudget()}
+
+	_, err = parseExprInner(helper, expr, values, visitorArgs)
+	require.NoError(t, err)
+	_, err = parseExprInner(helper, expr, values, visitorArgs)
+	require.NoError(t, err, "the configured decoded budget admits its exact boundary")
+	_, err = parseExprInner(helper, expr, values, visitorArgs)
+	require.ErrorIs(t, err, merr.ErrParameterTooLarge,
+		"all expression parses in one request must share the decoded total")
+	require.ErrorContains(t, err, "proxy.maxMembershipFilterPlanSize")
 }
 
 // The occurrence budget is documented as a per-request ceiling. It used to be
@@ -458,21 +504,30 @@ func TestRoaringMatchPreflightBudgetsAggregateDecodedBytes(t *testing.T) {
 func TestRoaringMatchPreflightBudgetIsSharedAcrossParses(t *testing.T) {
 	helper := newTestSchemaHelper(t)
 	pt := paramtable.Get()
-	// A structurally valid blob: this test is about the budget, so the parse
-	// must not fail for any other reason.
-	template, blob := roaringBytesTemplate(t, 1, 2, 3)
+	// Use a bitmap body large enough that its fixed decoded-container overhead
+	// still fits below two body lengths. This test targets the serialized-body
+	// occurrence budget; a tiny bitmap would now hit the shared configurable
+	// decoded budget first.
+	members := make([]int64, 4097)
+	for i := range members {
+		members[i] = int64(i * 2)
+	}
+	template, blob := roaringBytesTemplate(t, members...)
 	values := map[string]*schemapb.TemplateValue{"rb": template}
 	expr := "membership_match(Int64Field, {rb}, type=roaring)"
 
 	// Room for exactly one occurrence across the whole request (body basis;
 	// the fixed MRB1 header rides on top of the budget).
 	body := len(blob) - roaringfilter.HeaderSize
+	summary, err := serverroaring.Validate(blob)
+	require.NoError(t, err)
+	require.LessOrEqual(t, summary.EstimatedDecodedBytes, uint64(2*body-1))
 	pt.Save(pt.ProxyCfg.MaxMembershipFilterPlanSize.Key, strconv.Itoa(2*body-1))
 	defer pt.Reset(pt.ProxyCfg.MaxMembershipFilterPlanSize.Key)
 
 	visitorArgs := &ParserVisitorArgs{MembershipBudget: NewMembershipPreflightBudget()}
 
-	_, err := parseExprInner(helper, expr, values, visitorArgs)
+	_, err = parseExprInner(helper, expr, values, visitorArgs)
 	require.NoError(t, err, "first parse fits in the budget")
 
 	_, err = parseExprInner(helper, expr, values, visitorArgs)

internal/parser/planparserv2/utils.go

@@ -269,9 +269,7 @@ func castValue(dataType schemapb.DataType, value *planpb.GenericValue) (*planpb.
 	// proxy, instead of fanning out a GenericValue kBytesVal that segcore's plan
 	// parser cannot evaluate.
 	if IsBytes(value) {
-		return nil, merr.WrapErrParameterInvalidMsg(
-			"a bytes template value can only be used as the membership filter argument " +
-				"of membership_match")
+		return nil, bytesTemplateValueError()
 	}
 	if typeutil.IsJSONType(dataType) {
 		return value, nil
@@ -311,6 +309,12 @@ func castValue(dataType schemapb.DataType, value *planpb.GenericValue) (*planpb.
 		"cannot cast value to %s: incompatible source type", dataType.String())
 }
 
+func bytesTemplateValueError() error {
+	return merr.WrapErrParameterInvalidMsg(
+		"a bytes template value can only be used as the membership filter argument " +
+			"of membership_match")
+}
+
 func combineBinaryArithExpr(op planpb.OpType, arithOp planpb.ArithOpType, arithExprDataType schemapb.DataType, columnInfo *planpb.ColumnInfo, operandExpr, valueExpr *planpb.ValueExpr) (*planpb.Expr, error) {
 	var err error
 	operand := operandExpr.GetValue()

internal/querynodev2/delegator/scalar_pruner.go

@@ -65,10 +65,10 @@ func (lbe *LogicalBinaryExpr) Eval(evalCtx *EvalCtx) *bitset.BitSet {
 
 	// 3. set true for possible nil expr
 	if leftRes == nil {
-		leftRes = evalCtx.allTrueBitSet
+		leftRes = evalCtx.allTrueBitSet.Clone()
 	}
 	if rightRes == nil {
-		rightRes = evalCtx.allTrueBitSet
+		rightRes = evalCtx.allTrueBitSet.Clone()
 	}
 
 	// 4. and/or left/right results
@@ -161,7 +161,7 @@ func (ure *UnaryRangeExpr) Eval(
 				localBst.Set(idx)
 			}
 		default:
-			return evalCtx.allTrueBitSet
+			return evalCtx.allTrueBitSet.Clone()
 		}
 	}
 	return localBst

internal/querynodev2/delegator/scalar_pruner_test.go

@@ -0,0 +1,99 @@
+// 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 delegator
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
+	"github.com/milvus-io/milvus/internal/storage"
+	"github.com/milvus-io/milvus/pkg/v3/proto/planpb"
+)
+
+func TestUnsupportedBloomLeavesStayConservative(t *testing.T) {
+	const clusteringKeyFieldID = int64(100)
+	const bloomFieldID = int64(101)
+
+	bloomLeaf := func() *planpb.Expr {
+		return &planpb.Expr{Expr: &planpb.Expr_BloomFilterExpr{
+			BloomFilterExpr: &planpb.BloomFilterExpr{
+				ColumnInfo: &planpb.ColumnInfo{
+					FieldId:  bloomFieldID,
+					DataType: schemapb.DataType_Int64,
+				},
+			},
+		}}
+	}
+	clusteringKeyEqualsOne := &planpb.Expr{Expr: &planpb.Expr_UnaryRangeExpr{
+		UnaryRangeExpr: &planpb.UnaryRangeExpr{
+			ColumnInfo: &planpb.ColumnInfo{
+				FieldId:  clusteringKeyFieldID,
+				DataType: schemapb.DataType_Int64,
+			},
+			Op: planpb.OpType_Equal,
+			Value: &planpb.GenericValue{Val: &planpb.GenericValue_Int64Val{
+				Int64Val: 1,
+			}},
+		},
+	}}
+	binary := func(op planpb.BinaryExpr_BinaryOp, left, right *planpb.Expr) *planpb.Expr {
+		return &planpb.Expr{Expr: &planpb.Expr_BinaryExpr{
+			BinaryExpr: &planpb.BinaryExpr{Op: op, Left: left, Right: right},
+		}}
+	}
+
+	// The first AND narrows its result to the first segment. The Bloom leaf in
+	// the outer OR is unsupported by clustering-key pruning and must therefore
+	// conservatively restore all segments instead of reusing that narrowed
+	// mutable bitmap.
+	predicate := binary(
+		planpb.BinaryExpr_LogicalOr,
+		binary(planpb.BinaryExpr_LogicalAnd, bloomLeaf(), clusteringKeyEqualsOne),
+		bloomLeaf(),
+	)
+	parsed, err := ParseExpr(
+		predicate,
+		NewParseContext(clusteringKeyFieldID, schemapb.DataType_Int64),
+	)
+	require.NoError(t, err)
+
+	segmentStats := []storage.SegmentStats{
+		*storage.NewSegmentStats([]storage.FieldStats{{
+			FieldID: clusteringKeyFieldID,
+			Type:    schemapb.DataType_Int64,
+			Min:     storage.NewInt64FieldValue(0),
+			Max:     storage.NewInt64FieldValue(5),
+		}}, 1),
+		*storage.NewSegmentStats([]storage.FieldStats{{
+			FieldID: clusteringKeyFieldID,
+			Type:    schemapb.DataType_Int64,
+			Min:     storage.NewInt64FieldValue(10),
+			Max:     storage.NewInt64FieldValue(20),
+		}}, 1),
+	}
+	filteredSegments := make(map[UniqueID]struct{})
+	PruneByScalarField(
+		parsed,
+		segmentStats,
+		[]UniqueID{1, 2},
+		filteredSegments,
+	)
+
+	require.Empty(t, filteredSegments)
+}

pkg/util/paramtable/component_param.go

@@ -57,9 +57,10 @@ const (
 	DefaultSessionTTL        = 15 // s
 	DefaultSessionRetryTimes = 30
 
-	// DefaultMaxMembershipFilterPlanSize is the aggregate serialized size budget for
-	// membership-filter-bearing plans in one Search, HybridSearch, Query, or
-	// complex Delete request. It is deliberately below the default 256 MiB proxy
+	// DefaultMaxMembershipFilterPlanSize is the request-wide budget for both the
+	// aggregate serialized size of membership-filter-bearing plans and the
+	// aggregate estimated decoded size of Roaring filters. The two totals are
+	// checked independently. It is deliberately below the default 256 MiB proxy
 	// gRPC client send limit so placeholders and the rest of the internal request
 	// retain ample headroom.
 	DefaultMaxMembershipFilterPlanSize = 128 * 1024 * 1024
@@ -2620,10 +2621,11 @@ For migration, enable streaming.splitChunkSN first, then disable proxy.splitChun
 		DefaultValue: strconv.Itoa(DefaultMaxMembershipFilterPlanSize),
 		FallbackKeys: []string{"proxy.maxBloomFilterPlanSize"},
 		Version:      "3.0.0",
-		Doc: "The maximum aggregate serialized byte size of membership-filter-bearing expression plans " +
-			"in one Search, HybridSearch, Query, or complex Delete request. The proxy checks the assembled plans " +
-			"with proto.Size before proto.Marshal, and hybrid sub-searches share the configured budget. Must " +
-			"be positive; invalid values fall back to 128 MiB.",
+		Doc: "The request-wide membership-filter budget in bytes. It independently limits both the aggregate " +
+			"serialized size of membership-filter-bearing expression plans and the aggregate estimated decoded " +
+			"size of Roaring filters in one Search, HybridSearch, Query, or complex Delete request. The proxy " +
+			"checks assembled plans with proto.Size before proto.Marshal, and hybrid sub-searches and scorer " +
+			"filters share both totals. Must be positive; invalid values fall back to 128 MiB.",
 		Export:       true,
 		PanicIfEmpty: true,
 		Formatter: func(v string) string {