Add RUNTIME_FILTER predicate primitive (Bloom) for MSE runtime filter

[UOETianleZhang]

Summary

Issue: #200
Leaf-stage primitive for the upcoming MSE hash-join runtime filter (dynamic filter pushdown). This PR adds only the predicate plumbing and an in-process Bloom filter wrapper — no rule registration, no operator wiring, and the predicate has no producer yet. Independent of the broker-config scaffolding PR; either order works.

What's added

pinot-common

• Predicate.Type.RUNTIME_FILTER — umbrella predicate type for the runtime-filter feature, inclusive semantics, allows false positives but never false negatives. The concrete filter form is encoded inside the carried RuntimeFilter so additional forms (min/max, IN-list) slot in without a new Predicate.Type.
• RuntimeFilter interface + Kind enum — abstracts the filter form (Kind.BLOOM for v1; min/max and IN-list are future Kinds). Carries the column DataType so the evaluator can validate the wiring at construction.
• RuntimeFilterPredicate — carries a RuntimeFilter and an ExpressionContext. Uses base-class reference equality (a runtime filter is identified by the build-side instance that produced it, not by the bit pattern).
• BloomRuntimeFilter — in-memory RuntimeFilter implementation backed by Guava's BloomFilter, mirroring the funnel choices in BloomFilterIdSet (INT/FLOAT share an int funnel via raw-bits encoding, LONG/DOUBLE share a long funnel via raw-bits encoding, plus STRING and BYTES funnels). Lives in pinot-common so predicate classes can reference it. Serialization for cross-worker transport is intentionally deferred.

pinot-core

• RuntimeFilterPredicateEvaluatorFactory — umbrella dispatcher. Routes both raw-value and dictionary-based evaluation to a form-specific factory based on RuntimeFilter.Kind. Adding a new Kind means adding one case arm here plus the per-form factory — no churn to PredicateEvaluatorProvider or FilterPlanNode.
• BloomRuntimeFilterEvaluatorFactory — Bloom-specific factory. Raw-value path: per-DataType inner evaluators (INT/LONG/FLOAT/DOUBLE/STRING/BYTES), each delegating applySV to the matching BloomRuntimeFilter.mightContain overload. Dictionary-based path: probes the column dictionary once at construction time, populates an IntSet of matching dict ids, and answers applySV(int) from that set; sets _alwaysTrue/_alwaysFalse based on the count of matching dict ids. Fails fast on dataType mismatch on both paths.
• PredicateEvaluatorProvider — dispatch updated so both branches route RUNTIME_FILTER to the umbrella factory.

Supported data types (v1)

The Bloom filter form supports the following stored data types:

Supported in v1	Funnel
INT, FLOAT	int funnel (FLOAT via raw IEEE-754 bits)
LONG, DOUBLE	long funnel (DOUBLE via raw IEEE-754 bits)
STRING	unencoded-chars funnel
BYTES	byte-array funnel

Any other stored type (BIG_DECIMAL, BOOLEAN, TIMESTAMP, JSON, MAP/LIST/complex) fails fast at BloomRuntimeFilter construction. TODO: evaluate whether BOOLEAN/TIMESTAMP (stored as INT/LONG) and BIG_DECIMAL are worth supporting once the end-to-end feature lands; they are intentionally out of scope for now.

Type enforcement lives in the planner rule, not here. This primitive only fails fast as a backstop. The eligibility check that decides whether a join key's data type can carry a runtime filter belongs in the RuntimeFilterInsertRule (per review on #195) so unsupported types never reach the leaf stage.

NULL handling (v1)

NULLs are out of scope for v1 and are not modeled: the build side adds only non-null join keys and the probe side treats NULLs as non-matching (safe for inner equi-joins, where a NULL key never matches). The authoritative contract and a TODO(NULL support) for the end-to-end definition (code + user docs) live on the RuntimeFilter interface; the planner rule must not push a runtime filter where NULLs must survive the scan.

Out of scope (follow-up changes)

• QueryContext injection at the leaf-stage boundary
• HashJoinOperator build-side Bloom construction
• Probe-side RuntimeFilterOperator
• Calcite planner rule + physical runtime-filter node
• Wire format / cross-worker transport of the Bloom payload

Testing Done

• BloomRuntimeFilterTest (13 cases): constructor rejects unsupported data types; reports Kind.BLOOM; no false negatives for INT/LONG/FLOAT/DOUBLE/STRING/BYTES; observed false-positive rate within 5× of configured fpp; byte[] content equality hashes identically; raw-bits float encoding pinned down (including a non-canonical NaN).
• BloomRuntimeFilterEvaluatorFactoryTest (18 cases): factory rejects type mismatch and unsupported types on both raw-value and dictionary paths; evaluator metadata correct (predicate-type, data-type, dictionary-based, exclusive); no false negatives across all six SV types on the raw-value path; INT false-positive rate within tolerance; base-class default applySV overloads throw for the wrong type so callers cannot accidentally call applySV(long) on an INT-typed bloom; MV inherits correct semantics via the base loop; dictionary-based evaluator returns the correct matching dict ids (sorted ascending), short-circuits to _alwaysFalse on an empty dictionary, short-circuits to _alwaysTrue when every dict entry passes the Bloom; PredicateEvaluatorProvider routes raw-value and dictionary-based RUNTIME_FILTER to the factory.
• RuntimeFilterPredicateEvaluatorFactoryTest (2 cases): umbrella dispatcher routes Kind.BLOOM to the Bloom factory on both raw-value and dictionary code paths.
• mvn -pl pinot-common test -Dtest=BloomRuntimeFilterTest → 13/13
• mvn -pl pinot-core test -Dtest=BloomRuntimeFilterEvaluatorFactoryTest,RuntimeFilterPredicateEvaluatorFactoryTest → 20/20
• 33 new tests total; no existing tests touched.

[Copilot]

Pull request overview

Adds the BLOOM_MEMBERSHIP predicate primitive and in-process Bloom filter wrapper needed to support upcoming MSE hash-join runtime filtering, plus raw-value predicate evaluation plumbing and unit tests.

Changes:

• Introduces Predicate.Type.BLOOM_MEMBERSHIP, BloomMembershipPredicate, and RuntimeBloomFilter in pinot-common.
• Adds BloomMembershipPredicateEvaluatorFactory and routes raw-value evaluation through PredicateEvaluatorProvider (dictionary-based explicitly unsupported for now).
• Adds unit tests covering Bloom filter behavior and evaluator/provider wiring.

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/PredicateEvaluatorProvider.java	Routes raw-value BLOOM_MEMBERSHIP to the new evaluator factory; rejects dictionary-based usage.
pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactory.java	Implements per-type raw-value evaluators delegating to RuntimeBloomFilter.mightContain(...).
pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactoryTest.java	Adds coverage for factory wiring, per-type membership, overload behavior, MV behavior, and provider dispatch.
pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/Predicate.java	Adds new Predicate.Type.BLOOM_MEMBERSHIP enum value and documents semantics.
pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomMembershipPredicate.java	Adds a predicate that carries a RuntimeBloomFilter with reference-based equality semantics.
pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilter.java	Adds an in-memory Bloom wrapper mirroring BloomFilterIdSet funnel choices and raw-bits encoding.
pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilterTest.java	Adds tests for supported types, false-negative behavior, and bytes hashing semantics.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[siddharthteotia]

Can this be reviewed and merged independently of PR 197. Is this a pre-cursor to that ?

[UOETianleZhang]

Can this be reviewed and merged independently of PR 197. Is this a pre-cursor to that ?

Yes this one is independent of 197

[siddharthteotia]

I think we should not strictly tie this to Bloom Filter.

BLOOM_MEMBERSHIP makes this very narrow / specific. In future, the same foundation can be used for min-max, IN LIST as well.

So I suggest using something like RUNTIME_FILTER to prevent churn in future -- repeating boiler plate work later on for other filter types that can be dynamically pushed down into leaf.

[siddharthteotia]

Where are we relying on reference equality check ?

[UOETianleZhang]

we weren't actually relying on it anywhere, the defensive equals / hashCode override here was boilerplate without a real consumer.

[siddharthteotia]

Is this call being made to intentionally keep the scope limited (while we know that it will certainly be a useful improvement) OR we don't know if dictionary will be useful for the runtime path.

I am ok with the former as we initially have a bigger fish to fry for the overall design of this feature. However if it's the latter then I am curious why we think it won't be useful generally.

We are building the bloom on raw values of the build side of the join and then sending that to leaf stage as a filter for the large fact table scan so that we read fewer fact rows into the JOIN.

Say the fact has 1 billion rows and the join key column (for which bloom has been pushed down) is dictionary encoded

The non dictionary based evaluator will do something like

for each of 1B rows {
   dictId   = column.readDictId(docID)     --> 1B reads of forward index 
   rawValue = dictionary.lookup(dictId)   -->  1B lookup of dictionary 
   if bloom.mightContain(rawValue).        --> 1B reads of bloom         
        // do something ....
}

Now if you leverage the dictionary instead, let's assume cardinality of 1M for the join key column in fact, we can potentially do something like ..

matchingDictIds = empty bitmap
for each of 1M dict entries.                        --> 1M reads of dictionary
    rawValue = dictionary.getValue(i).         --> retrieve raw value from dictionary
    if bloom.mightContain(rawValue):         --> 1M reads of bloom 
        matchingDictIds.set(i)
        project  ....

The difference is obvious ....

FWIW, the basic building blocks to write a dictionary based predicate evaluator are already there for the usual predicates so it should not be that much of a net new work but I don't want to force include it in the scope.

[UOETianleZhang]

Yes it was planned in future PRs but since you have mentioned this - I have added the implementation so it can be self-contained. Feel free to review the latest version!

[UOETianleZhang]

I think we should not strictly tie this to Bloom Filter.

BLOOM_MEMBERSHIP makes this very narrow / specific. In future, the same foundation can be used for min-max, IN LIST as well.

So I suggest using something like RUNTIME_FILTER to prevent churn in future -- repeating boiler plate work later on for other filter types that can be dynamically pushed down into leaf.

Sounds good, it totally makes sense. I have added some factories and interfaces to make it able to be extended.

[siddharthteotia]

We should add a TODO for NULL Support. By "Support" I mean the contract should be documented in code and user docs. For now, it is fine to not worry about NULLs and focus on rest of the feature.

[siddharthteotia]

Please update the PR description clearly calling out the types that are supported in V1 of this feature and add a TODO for types we are not planning as of now.

Per my comment on #195 regarding planner rule, I think the type enforcement / check should happen in the rule.

[siddharthteotia]

The enum itself does not go on the wire right ?

[UOETianleZhang]

Correct. Predicate.Type does not go on the wire. It's a server-local, query-engine-internal representation.

[UOETianleZhang]

RUNTIME_FILTER in particular is never parsed from a query at all. it's injected in-process into the leaf filter tree, so it has no wire representation by construction.

[UOETianleZhang]

We should add a TODO for NULL Support. By "Support" I mean the contract should be documented in code and user docs.

Done in db0960f.

[UOETianleZhang]

Please update the PR description clearly calling out the types that are supported in V1 ... I think the type enforcement / check should happen in the rule.

Updated the PR description with a "Supported data types (v1)" table (INT/LONG/FLOAT/DOUBLE/STRING/BYTES) and a TODO for the rest (BOOLEAN/TIMESTAMP/BIG_DECIMAL/JSON/complex).