[Design] Runtime Filter (Dynamic Filter Pushdown) for MSE Hash Joins

[UOETianleZhang]

Runtime Filter (Dynamic Filter Pushdown) for MSE Hash Joins

TL;DR

For inner hash joins in the Multi-Stage Query Engine (MSE), materialize a compact Bloom filter over the build-side join keys and push it down to the probe-side segment scan. Non-matching rows are pruned during I/O instead of being fetched, shuffled across the network, and dropped at the join operator. This is a well-known MPP optimization (Trino, StarRocks, Apache Doris, Spark, DuckDB). Pinot doesn't have it today; this issue proposes the design.

Design follows the Trino / StarRocks / Doris pattern: a per-query, broker-mediated runtime-filter registry that producers (the HashJoinOperator in the intermediate stage) write to on build completion, and consumers (leaf-stage segment scans) poll between row batches. Timing is asynchronous and best-effort — late rows are pruned, rows already in flight when the filter arrives are not.

Revised on 2026-05-28 based on review feedback from @siddharthteotia (issue comments). The earlier "same-OpChain hand-off via OpChainExecutionContext" sketch was incorrect: build (intermediate stage) and probe leaf are always different OpChains, so cross-stage transport is now first-class instead of a follow-up.

1. Status & Scope

• In scope (v1): Inner hash joins with single equi-condition; Bloom filter built on the build side; cross-stage transport via a broker-mediated registry; pushed into the leaf-stage segment scan as a new BLOOM_MEMBERSHIP predicate.
• Timing model: Trino-style asynchronous / best-effort. Snowflake schema (FACT × N-DIM, multi-producer to one fact scan) supported day one.
• Out of scope (follow-ups, no longer including cross-stage transport): Semi-join and anti-join runtime filters; multi-stage runtime-filter chains beyond simple snowflake; non-equi joins; Spark-style synchronous scheduling barrier; dictionary-encoded leaf path (pre-materialized matching dict-ID set).

2. Background

2.1 JOIN basic knowledge

A JOIN combines rows from two tables that share a value in some column (the join key). Example: a 10-billion-row orders table joined against a small customers table on customer_id to find VIP orders.

2.2 What's a hash join?

There are two phases for hash join:

1. Build phase — load the smaller side (the 3 VIPs) into an in-memory hash table keyed on customer_id.

2. Probe phase — stream the bigger side row-by-row, look each customer_id up in the hash table; emit matches, discard misses.

2.3 What's the problem?

In a distributed engine like MSE, the probe side and the build side often run on different workers. Every probe-side row gets read from storage and sent across the network, only to be discarded at the join operator if its key isn't in the hash table. For a selective join (small build side, large probe side), that's a lot of wasted I/O and bandwidth.

2.4 What's a "runtime filter"?

A small "did the build side maybe contain this key?" check that the build side ships to the probe side at runtime, after the hash table is built. The probe side applies it as early as possible — ideally inside the segment scan, so non-matching rows are skipped during I/O. The check is a Bloom filter, which has zero false negatives (so the join stays correct) and a tunable false-positive rate (so it stays compact).

This is the same trick Spark, Trino, Presto, and DuckDB use. Pinot already does a related thing for SEMI joins via PinotJoinToDynamicBroadcastRule (broadcasts an IN-list); the new feature does it for INNER hash joins via a Bloom filter, transported via a registry rather than baked into the physical plan.

---

3. Design

3.1 Three layers + a control plane

Layer	Module	Responsibility
Planner	pinot-query-planner	Detect eligible inner joins (single equi-condition, build-side row count below threshold). Insert a physical runtime-filter node that names the registry key (requestId, joinColumn).
Runtime	pinot-query-runtime	HashJoinOperator builds the Bloom on finishBuildingRightTable() and publishes to the registry. Leaf-scan OpChain polls the registry between row batches.
Leaf	pinot-core	Accept the Bloom as a BLOOM_MEMBERSHIP predicate inside the segment scan via FilterPlanNode.combineWithRuntimeBloomFilters (already landed in #197).
Control plane (new)	broker	Hosts the per-query RuntimeBloomFilterRegistry keyed by (requestId, columnName). Producers write, consumers read. Cleaned up at query end.

3.2 Key design decisions

1. Inner joins, single equi-condition first. Mirrors the restriction in the existing PinotJoinToDynamicBroadcastRule. Composite keys and outer joins are out of scope for v1.
2. Bloom filter as the filter form for v1. Compact (KB-level), zero false negatives. Implementation reuses Guava's BloomFilter (matching BloomFilterIdSet). The registry shape is polymorphic so IN-list / min-max can be added later without rework (cf. StarRocks / Doris multi-form pattern in §3.4).
3. New Predicate.Type.BLOOM_MEMBERSHIP. Dedicated predicate keeps evaluator routing clean and avoids regressions in the existing IN path.
4. QueryContext carries runtime Bloom filters per column. Leaf-stage FilterPlanNode AND-combines a synthetic BLOOM_MEMBERSHIP predicate per column into the segment-scan filter tree. Queries without runtime filters allocate nothing.
5. Broker-mediated registry, not in-process hand-off. OpChainId = (requestId, virtualServerId, stageId) is stage-bound, every PhysicalExchange creates a stage boundary, and build / probe always live in different OpChains. There is no in-process channel that survives the stage boundary. The registry lives on the broker, producers publish(requestId, column, bloom), consumers get(requestId, column).
6. Asynchronous / best-effort timing. Probe-side leaf scans start immediately. The leaf execution loop polls the registry between batches; when an entry appears, it merges into QueryContext.runtimeBloomFilters and the next batch is filtered. Rows already in flight when the filter arrives are not retroactively filtered — this is the same trade-off Trino / StarRocks / Doris accept. We instrument the "leak" (rows emitted before the filter arrived) so a future decision to add a synchronous barrier is data-driven.
7. Multi-producer to single consumer day one. Snowflake schemas (FACT × N-DIM) work without rework: each producer HashJoin writes its own column key into the same per-query registry, the fact scan reads all columns it touches. The consumer-side Map<String, RuntimeBloomFilter> already supports this (Inject runtime filters at the leaf-stage filter boundary #197).
8. Automatic triggering, hint-overridable. The optimizer rule fires when build-side RelMetadataQuery.getRowCount() is below pinot.broker.mse.runtime.filter.build.cardinality.threshold (default 1M). Per-join enable_runtime_filter hint can force on/off.
9. Flag-gated, disabled by default. Same rollout pattern as the recent MSE fingerprint change. No behavior change for existing queries until the flag flips.

3.3 Data flow

The diagram below was sketched against the original same-OpChain design and is partially superseded — the new control-plane path (registry between intermediate-stage HashJoin and leaf-stage scan) replaces the in-OpChain hand-off shown here. The textual description in §3.2 is authoritative.

3.4 Industry comparison

Surveyed along the dimensions that matter operationally:

System	Filter forms	Build trigger	Transport	Timing	Probe app point
Trino / Presto	Discrete value set, range, Bloom (recent)	Cost-based; per-type size thresholds	Coordinator-mediated registry (DynamicFilterService)	Async / late, default ~5s wait	Row filter + connector pushdown (Iceberg / Hive / Delta)
Spark DPP (SPARK-25540)	Discrete partition values	Plan rule (join column = partition column)	Subquery materialization	Plan-time barrier	Partition prune at file scan
Spark RBF (SPARK-32268, 3.3+)	Bloom	Cost-based, off by default	BroadcastExchange synchronous broadcast	Plan-time barrier	Row filter
DuckDB	Hash-table membership (perfect)	Automatic	In-memory shared (single-node)	Pipeline barrier	Row filter
ClickHouse	IN-list via GLOBAL JOIN	Manual keyword	Distributed broadcast	Late / serial	Row filter (plus static on-disk indices)
StarRocks / Doris	IN-list, MIN/MAX, Bloom — in parallel from same join	Cost-based, per-type thresholds	FE-mediated registry (global), BE-local	Async / late, default 1s wait	Row filter + segment-level zone map / Bloom
Pinot today (SEMI JOIN only via PinotJoinToDynamicBroadcastRule)	IN-list	Hint + cardinality threshold	Plan rewrite + mailbox broadcast	Plan-time	Segment-scan IN predicate
Pinot proposed v1	Bloom (extensible)	Cost-based + per-join hint	Broker-mediated registry	Async / late	Segment-scan BLOOM_MEMBERSHIP predicate (PR #196 + #197)

Three takeaways:

• Async / registry is industry consensus for general-purpose distributed MPP engines. Trino, StarRocks, Doris all converged here independently. Our proposal sits in the same row.
• Spark's synchronous barrier is cheap only because Catalyst already has plan-time barriers (DynamicPruningSubquery, BroadcastExchange with synchronous materialization). MSE has neither. Replicating Spark's coverage in Pinot is not a planner-rule patch — it would require building an OpChain-level scheduling primitive, on the order of quarters of work. That's a v2 decision, not a v1 default.
• No SOTA system tries to retroactively prune the early-row leak. They all manage it with timeouts and instrumentation. We follow suit.

3.5 Why async (Trino-style) not synchronous (Spark-style)

	Trino-style (proposed v1)	Spark-style (deferred to v2)
Coverage	Late rows pruned; early rows leak	~100% (probe blocks on Bloom)
Change to MSE scheduling	None	Significant — new OpChain-level barrier
Multi-producer (snowflake)	Falls out for free	Requires plan-shape rewrite per dim
Cost to add	Weeks	Quarters
Risk	Low — additive to existing pipeline	High — new scheduling primitive
Decision driver to v2	Measured leak rate in production	—

We instrument the leak (rows-emitted-before-Bloom-arrived vs. after) on Phase 7 so the v2 decision is data-driven.

---

4. Implementation plan

Total: 9 phases. PRs #195, #196, #197 are already up.

#	Phase	Status	What lands
0	Broker config + join hint	Done · #195	pinot.broker.mse.runtime.filter.enabled (default false), build-cardinality threshold, enable_runtime_filter hint.
1	BLOOM_MEMBERSHIP predicate primitive	Done · #196	New predicate type, BloomMembershipPredicate, RuntimeBloomFilter, per-DataType evaluator factory, provider dispatch.
2	QueryContext map + leaf injection	In review · #197	Map<String, RuntimeBloomFilter> on QueryContext; FilterPlanNode.combineWithRuntimeBloomFilters AND-injects BLOOM_MEMBERSHIP predicates. PR description to be revised to drop the OpChainExecutionContext framing.
3	RuntimeBloomFilterRegistry (new)	To do	Per-query, broker-resident registry keyed by (requestId, columnName). Producers call publish(...), consumers call get(...). Cleaned up at query end.
4	HashJoinOperator publish hook	To do	In finishBuildingRightTable(): iterate the PrimitiveLookupTable keys, build a RuntimeBloomFilter sized to the build cardinality, publish to the registry. Replaces the discarded "publish via OpChainExecutionContext" idea.
5	Leaf-scan consumer (best-effort poll)	To do	Between row batches in the leaf execution loop: poll the registry for columns touched by the scan, merge into QueryContext.runtimeBloomFilters. Phase 2's injection logic does the rest. Instrument rows-emitted-before-Bloom-arrived vs. after.
6	RuntimeFilterInsertRule planner rule	To do	Calcite rule detecting eligible inner equi-joins; inserts a physical node carrying the registry key. Mirrors PinotJoinToDynamicBroadcastRule's shape (sibling rule), registered in PhysicalOptRuleSet.
7	Snowflake integration test + benchmark	To do	FACT × 2-DIM test per @siddharthteotia's example. pinot-perf harness measures end-to-end latency and leak rate. Decision driver for Phase 8.
8	Spark-style synchronous barrier	Tracked, not committed	OpChain-level barrier so leaf scans wait on the registry. Closes the leak entirely but invasive. Only if Phase 7 numbers warrant.

4.1 Out of scope for v1 (follow-up issues to file once v1 lands)

• Dictionary-encoded leaf path for BLOOM_MEMBERSHIP (pre-materialized matching dict-ID set)
• Semi-join and anti-join runtime filters
• Multi-stage runtime-filter chains beyond simple snowflake
• Min/max range and IN-list filter forms (registry shape supports them — just deferring implementation)
• Phase 8: Spark-style synchronous scheduling barrier

---

5. Testing strategy

• Unit tests at every layer — predicate evaluator across SV types (Add RUNTIME_FILTER predicate primitive (Bloom) for MSE runtime filter #196); leaf injection no-op + injection cases (Inject runtime filters at the leaf-stage filter boundary #197); registry publish/get correctness; build-side publish (no false negatives); probe-side consumer polling; planner rule trigger / non-trigger matrix.
• Planner golden plans — PhysicalOptimizerPlans.json cases: INNER triggers, OUTER does not, build-too-large does not, hint-disabled does not, hint-forced does, snowflake (multi-producer) produces correct registry keys.
• Snowflake integration test — TPC-H-shaped FACT × 2-DIM in pinot-integration-tests: two HashJoins in different intermediate stages both publish to the same fact-scan registry, leaf scan pulls both Blooms. Assert leaf-stage emitted row count drops vs. flag-off baseline.
• Performance benchmark with leak instrumentation — pinot-perf measures latency delta and the (rows-before-Bloom / total-rows) ratio. Target ≥ 20% latency improvement on a representative selective join; leak ratio is the v2 decision driver.

---

6. Rollout

1. Land all phases behind pinot.broker.mse.runtime.filter.enabled=false (no behavior change for existing queries).
2. Enable in a staging cluster, monitor latency, CPU, and the leak ratio on join-heavy workloads.
3. Enable cluster-wide once stable; consider per-table overrides.
4. If staging leak ratio is acceptable (TBD threshold), stay on async. If unacceptable on representative workloads, open the Phase 8 discussion.

[siddharthteotia]

Same-worker (same OpChain) hand-off first. When build and probe live in the same OpChain the bloom is published through OpChainExecutionContext. Cross-worker transport (serialized blocks across the mailbox) is a follow-up — it adds protocol risk and isn't required for the first useful win.

Can you elaborate on this a bit more ? Whose OpChainContext will it be published from ? I don't think this is even possible in the current architecture.

Let's take the simplest STAR Schema FACT to DIMENSION JOIN example

  SELECT f.user_id, f.amount, d.region
  FROM   fact_orders f
  JOIN   dim_users d ON f.user_id = d.user_id
  WHERE  d.tier = 'VIP' 
  Sizes: fact_orders = 1 B rows, dim_users = 1 M rows, VIPs = 10 K, matching orders = 10 K.

Current State without dynamic bloom filter injection

         ┌─────────────────────── STAGE 2 (intermediate) ──────────────────────────┐
         │                                                                          │
         │   HashJoinOperator                                                       │
         │   ┌──── build ────┐         ┌────────── probe ──────────┐                │
         │   │ 10 K VIP rows │         │ 1 000 000 000 rows in     │                │
         │   │  in hash table│         │   999 999 990 thrown away │                │
         │   └──────▲────────┘         └────────────▲──────────────┘                │
         │          │ 10 K                          │ 1 B  ← network + CPU burned   │
         │     MailboxReceive                   MailboxReceive                      │
         └──────────┼──────────────────────────────┼─────────────────────────────────┘
                    │                              │
          ┌─────────┴──────────┐        ┌──────────┴──────────────────────────┐
          │ STAGE 1 (leaf)     │        │ STAGE 1 (leaf)                       │
          │ server A           │        │ servers B,C,D (fact segments)        │
          │ scan dim_users     │        │ scan fact_orders                     │
          │ WHERE tier='VIP'   │        │ no predicate on user_id              │
          │ → 10 K rows        │        │ → 1 000 000 000 rows read + shipped  │
          └────────────────────┘        └──────────────────────────────────────┘

We want to get to something like this to reduce the number of records we read from FACT

 ┌─────────────────────── STAGE 2 (intermediate) ──────────────────────────┐
         │   HashJoinOperator                                                       │
         │   ┌──── build (10 K) ───┐    ┌────── probe (10 K) ──────┐                │
         │   │ finishBuildingRight │    │ 10 K rows in, all match  │                │
         │   │ → build Bloom on    │    │ 0 wasted probes          │                │
         │   │   user_id keys      │═╗  └────────────▲─────────────┘                │
         │   └─────────────────────┘ ║               │                              │
         │          ▲                ║               │ 10 K                         │
         │          │ 10 K           ║          MailboxReceive                      │
         │     MailboxRecv           ║               │                              │
         └──────────┼────────────────╫───────────────┼──────────────────────────────┘
                    │                ║ ~8 KB Bloom   │
                    │                ║ broadcast mailbox                            │
                    │                ║               │
          ┌─────────┴──────────┐  ┌──╨───────────────┴───────────────────────────┐
          │ STAGE 1 (leaf)     │  │ STAGE 1 (leaf) — fact_orders                  │
          │ scan dim_users     │  │   QueryContext.runtimeBloomFilters[user_id]   │
          │ → 10 K rows        │  │     ↓ injected into FilterPlanNode (PR #197)  │
          │                    │  │   FilterPlanNode = AND(                       │
          │                    │  │       baseFilter,                              │
          │                    │  │       BLOOM_MEMBERSHIP(user_id))               │
          │                    │  │   → segment scan reads only ~10 K + FP rows   │
          │                    │  │   → ships 10 K, not 1 B                       │
          └────────────────────┘  └───────────────────────────────────────────────┘

Couple of facts from current implementation of exchanges

• OpChainId = (requestId, virtualServerId, stageId) and every exchange in MSE creates a stage boundary (See PlanFragmentAndMailboxAssignment ).
• Build and probe-side leaf scans are always different stages → different OpChains.
• Every exchange creates a stage boundary → different stage → different OpChainId → different OpChainExecutionContext.
• "Same worker" ≠ "same OpChain." Two stages can run on the same physical server and still be two distinct OpChains with isolated execution contexts.
• There is no existing facility for one OpChain's runtime state to land inside another OpChain's QueryContext.

Based on this I don't see how through same OpChainContext we can even activate the dynamic bloom filter on the leaf FACT scan. To visualize this, I have labeled the OpChains in the diagram above and we can see that there is no way to share OpChainContext and publish the bloom.

   ┌─ STAGE 2 ─────────────────────────────────────────────────────────────┐
     │                                                                        │
     │   OpChain  (reqId, vsrv=W1, stageId=2)     ← ONE OpChain               │
     │   ┌───────────────────────────────────────────────────────────────┐    │
     │   │  MailboxRecv(dim) ─→ HashJoin.build ─→ Bloom built here  ★    │    │
     │   │  MailboxRecv(fact)─→ HashJoin.probe                           │    │
     │   └───────────────────────────────────────────────────────────────┘    │
     │              ▲                          ▲                              │
     └──────────────│──────────────────────────│──────────────────────────────┘
                    │ mailbox                  │ mailbox
     ┌─ STAGE 1 ────┼─────────────┐   ┌────────┼──────────────────────────┐
     │ OpChain      │             │   │ OpChain│  OpChain   OpChain        │
     │ (req, vsrv=A,│ stage=1)    │   │ (req,B,│   (req,C,   (req,D,        │
     │ ┌────────────┴────────┐    │   │  st=1) │    st=1)    st=1)          │
     │ │ LeafOp scan dim     │    │   │ ┌──────┴──┐ ┌────────┐ ┌────────┐  │
     │ │ WHERE tier='VIP'    │    │   │ │LeafOp   │ │LeafOp  │ │LeafOp  │  │
     │ │ → MailboxSend       │    │   │ │fact seg │ │fact seg│ │fact seg│  │
     │ └─────────────────────┘    │   │ └─────────┘ └────────┘ └────────┘  │
     └────────────────────────────┘   └─────────────────────────────────────┘
          dim_users server                  fact_orders servers B, C, D

[siddharthteotia]

More Feedback / Questions:

In the first version, the design should not be tailor made for a simple STAR Schema query. A Snowflake schema JOIN query (e.g FACT to 2 dimension table JOINS) should also be supported as that will influence how the dynamic filter is built and transported into the leaf stage (FilterPlanNode) for FACT Scans

Something like this...

  SELECT f.amount
  FROM   fact_orders f
  JOIN   dim_users     d1 ON f.user_id    = d1.user_id    WHERE d1.tier     = 'VIP'
  JOIN   dim_products  d2 ON f.product_id = d2.product_id WHERE d2.category = 'Electronics'
  Sizes: fact_orders = 1 B, VIPs = 10 K, Electronics products = 50 K, final matches = 10 K.

                        ┌──────── STAGE 3 ────────┐
                         │   HashJoin #2            │
                         │   probe ← join#1 result  │
                         │   build ← dim_products   │
                         │   finish → Bloom₂        │═══╗
                         └──▲────────────────▲──────┘    ║
                            │ J1 result      │ 50 K       ║
                            │                MailboxRecv  ║
                ┌───────────┴──────┐                      ║
                │   STAGE 2        │                      ║
                │   HashJoin #1    │                      ║
                │   build ← d1     │                      ║
                │   finish → Bloom₁│═════════╗            ║
                │   probe ← fact   │         ║            ║
                └──▲──────────▲────┘         ║            ║
                   │ 10 K     │ 10 K (post-filter)        ║
              MailboxRecv  MailboxRecv                    ║
                   │          │                ║          ║
     ┌─────────────┴───┐  ┌───┴────────────────╨──────────╨───────────────────────────┐
     │ STAGE 1 (leaf)  │  │ STAGE 1 (leaf) — fact_orders                                │
     │ scan dim_users  │  │   QueryContext.runtimeBloomFilters = {                      │
     │ WHERE VIP       │  │     user_id    : Bloom₁,                                    │
     │ → 10 K rows     │  │     product_id : Bloom₂                                     │
     └─────────────────┘  │   }                                                          │
                          │   FilterPlanNode = AND(                                      │
     ┌─────────────────┐  │       baseFilter,                                            │
     │ STAGE 1 (leaf)  │  │       BLOOM_MEMBERSHIP(user_id),    ← from Bloom₁           │
     │ scan dim_products│ │       BLOOM_MEMBERSHIP(product_id)  ← from Bloom₂           │
     │ WHERE category= │  │   )                                                          │
     │  'Electronics'  │  │   → ~10 K + FP rows scanned + shipped (was 1 B)             │
     │ → 50 K rows     │  └──────────────────────────────────────────────────────────────┘
     └─────────────────┘

This is also important in the context of my previous comment on OpChain:

• Two producer OpChains must both publish into one consumer OpChain per fact server.
• The two producers live in different intermediate stages, on potentially different workers. They both want to write to QueryContext.runtimeBloomFilters on every Stage 1 fact-scan OpChain.
• I see that in PR Inject runtime filters at the leaf-stage filter boundary #197's Map<String, RuntimeBloomFilter> you are supporting on the consumer side.
• The missing piece is the transport that allows multiple upstream OpChains to deposit entries into the same downstream
  OpChain's QueryContext.

[siddharthteotia]

[ More Feedback / Questions ]

Have we done some research on timing ? How do other SOTA (state of the art ) systems in industry handle timing. IIRC, Trino / Presto does some best effort

MSE leaf scans don't wait — they start producing rows the moment the OpChain is scheduled. The Bloom is born after finishBuildingRightTable(), which is after the fact leaf has already streamed a large fraction of its rows. So if we put the event on a timeline, it's highly like that by the time bloom is ready (and gets shipped), the fact leaf has shipped most rows.

I think the design choice here will have a considerable impact on the speed-ups. I suggest reading up how Trino does this (deadline based, best effort) v/s what is done dynamically in Spark - DPP or Spark Runtime Bloom Filter features. The Spark one's are more architecturally cleaner and may require a lot of heavy lifting but will deliver large order of magnitude wins.

We need to make a conscious call on this and plan the design (and it's subsequent evolution) accordingly.

[siddharthteotia]

I don't think predicate type should be specifically tied to bloom filter. It's more than just a naming thing

See my comment here - #196 (comment)

[siddharthteotia]

If we have done some comparison across systems, I would like to see some data on the following dimensions. Systems to consider -- DuckDB, Clickhouse, Spark Trino etc.

• filter forms supported (Bloom / IN-list / min-max)
• build-side trigger (size threshold, cost-based, hint-only)
• transport (mailbox / broadcast / coordinator-mediated)
• timing (blocking / late / plan-time)
• probe-side application point (segment scan / row filter / partition prune)

[UOETianleZhang]

Thanks @siddharthteotia! These three points land, and they reshape the design more than I'd appreciated. Let me re-state what I now believe is correct, then propose where I think we should go.

1. About the OpChain assumption

You're right: OpChainId is stage-bound, every PhysicalExchange becomes a stage boundary (PlanFragmentAndMailboxAssignment confirms this), and the fact-scan OpChain is always a different OpChain from the HashJoin OpChain in the intermediate stage. That means for the INNER hash join we're actually targeting, we still need to do cross-stage transport.

2. About multi-producer use case

Agreed. Was focusing on 1 x 1 join - it would be definately helpful for JOIN of multiple tables.

[UOETianleZhang]

About the timing thing - thanks for flagging this. I also agreed. My understanding of the timeline is:

t=0     fact leaf scheduled → starts emitting rows immediately
t=0     dim  leaf scheduled → scans dim
t=t1    dim leaf done, ships to STAGE 2 HashJoin
t=t2    STAGE 2 finishes build → Bloom is born
t=t3    Bloom ships to fact leaf  ← by now some rows are already on the wire

Surveying what other distributed engines do:

System	Filter forms	Build-side trigger	Transport	Timing	Probe-side application point
Trino / Presto	Discrete value set, range (min/max), Bloom (recent versions)	Cost-based; per-type configurable size threshold (e.g. dynamic-filtering.large.max-distinct-values-per-driver)	Coordinator-mediated registry (DynamicFilterService); per-query in-memory	Async / late — probe starts immediately, polls supplier; configurable max wait (default ~5s)	Row filter at TableScanOperator + pushed into the source connector (Iceberg manifest / Parquet row-group stats / Hive partition prune)
Spark DPP	Discrete value set of partition keys (subquery result)	Plan rule: only fires when join column = partition column	Materialized subquery via Catalyst plan rewrite	Plan-time barrier — subquery must resolve before file scan reads partitions	Partition prune at the file scan layer (no row-level filter)
Spark RBF	Bloom	Cost-based; off by default; configurable size threshold	BroadcastExchange (synchronous broadcast through ExecSubqueryExpression)	Plan-time barrier — broadcast must materialize before probe scan starts	Row filter at scan (no source connector pushdown beyond what Spark already supports)
DuckDB	Hash-table membership ("perfect filter" — the table itself is the filter)	Automatic for every hash join	In-memory shared (single-node, no transport)	Pipeline barrier — probe pipeline depends on build pipeline in the morsel-driven scheduler	Row filter; in some cases pushed into the columnar scan via late materialization
ClickHouse	IN-list (GLOBAL JOIN materializes right side as a Set / temporary table)	Manual: user writes GLOBAL JOIN in SQL (or implicit via distributed_product_mode)	Distributed broadcast — right side materialized on coordinator, shipped to every shard	Late / serial — right side fully materialized before left scans	Row filter; on top of MergeTree's static storage-level Bloom and skip indices (not the same thing — those are pre-built at merge time)
StarRocks / Apache Doris	IN-list, MIN/MAX, AND Bloom — all three can be produced from the same join and applied in parallel	Cost-based, per-type size thresholds (e.g. runtime_filter_max_in_num default 1024 for IN-list)	FE-mediated for global RF (cross-BE), BE-local for local RF	Async / late — configurable wait timeout (runtime_filter_wait_timeout_ms default 1000ms)	Row filter + segment-level (zone map / on-disk bloom index match) — closest in spirit to what we're building
Pinot proposed	Bloom (registry shape is polymorphic — can host IN-list / min-max later)	Cost-based via RelMetadataQuery.getRowCount() + per-join enable_runtime_filter hint	Broker-mediated registry keyed by (requestId, column)	Async / late (Trino-style); fact scan polls registry between batches	Segment-scan BLOOM_MEMBERSHIP predicate (PR #196) injected by FilterPlanNode (PR #197)

Some observations from the survey:

1. Async best-effort is the industry consensus for general-purpose distributed engines. Trino, StarRocks, Doris all picked the same pattern independently. The architectural shape they converged on (a coordinator-side registry that producers publish to and consumers poll) is exactly what we'd build for MSE.
2. Spark's "sync" cheapness is a Spark-specific accident. DPP and RBF are cheap to add in Spark because Catalyst already has plan-time barriers. MSE has neither. Replicating the win in MSE is not a planner-rule patch - it's building an entire scheduling primitive, that may need quarter-level effort.
3. No SOTA system tries to retroactively prune the leak. Trino, StarRocks, Doris all accept "rows shipped before the filter arrived are lost". They manage it with timeouts and instrumentation, not with row buffering or scan rewind. This is a settled trade-off in the field, not a corner we'd be cutting.

[UOETianleZhang]

Given above, I would lean on the aync style (Trino, StarRocks, Doris all chose this). Let me know what you think?

I will update the design as we discussed!

[siddharthteotia]

Thanks @siddharthteotia! These three points land, and they reshape the design more than I'd appreciated. Let me re-state what I now believe is correct, then propose where I think we should go.

1. About the OpChain assumption

You're right: OpChainId is stage-bound, every PhysicalExchange becomes a stage boundary (PlanFragmentAndMailboxAssignment confirms this), and the fact-scan OpChain is always a different OpChain from the HashJoin OpChain in the intermediate stage. That means for the INNER hash join we're actually targeting, we still need to do cross-stage transport.

2. About multi-producer use case

Agreed. Was focusing on 1 x 1 join - it would be definately helpful for JOIN of multiple tables.

I think you have designed it in a way that supports multiple consumers (the code shows that) and that's one of the good aspects of the design. The Map<String, RuntimeBloomFilter> keyed by column is the right shape for multi-producer fan-in (two upstream joins both writing their respective columns into the same map). Similarly the single-AND-bundle wrap in FilterPlanNode is the right shape for composing the pushed down dynamic filter. So this is good.

We need to decide the multi-producer and the bloom filter transport -- I suggest studying PinotJoinToDynamicBroadcastRule and see if we can piggy back on it. Spark DPP does something similar too.

[siddharthteotia]

I would lean on the aync style (Trino, StarRocks, Doris all chose this). Let me know what you think?

Did you do some research and capture trade-offs ? I want us to understand and call out the trade-offs. I can chime in more on this a bit later.

[UOETianleZhang]

Yes I added trade off in comment

[UOETianleZhang]

I think you have designed it in a way that supports multiple consumers (the code shows that) and that's one of the good aspects of the design. The Map<String, RuntimeBloomFilter> keyed by column is the right shape for multi-producer fan-in (two upstream joins both writing their respective columns into the same map). Similarly the single-AND-bundle wrap in FilterPlanNode is the right shape for composing the pushed down dynamic filter. So this is good.

We need to decide the multi-producer and the bloom filter transport -- I suggest studying PinotJoinToDynamicBroadcastRule and see if we can piggy back on it. Spark DPP does something similar too.

Yes will take sometime today to design this part. It is also closely related to the next PR after #195, #196 and #197