yashmayya opened a new pull request, #18848:
URL: https://github.com/apache/pinot/pull/18848
## What
Adds **probe-side runtime filters for equi-`INNER` joins** in the
multi-stage query engine (MSE). When
the build (right) side of a hash join is small or selective, the planner
builds a reducer from its
distinct join keys and pushes it down to the probe (left) **leaf scan**, so
the probe table drops rows
that cannot possibly match *before* they are shuffled across the network
into the join.
This is the `INNER`-join counterpart of the existing `SEMI`-join dynamic
broadcast
(`PinotJoinToDynamicBroadcastRule`). It is **disabled by default**.
## Why
For the classic fact ⋈ dim shape — a large fact table joined to a small
dimension table (or a heavily
filtered build side) — the MSE today hash-shuffles the **entire** probe
(fact) side into the join stage,
even though only the rows whose join key appears on the (tiny) build side
can contribute to the result.
That wastes scan, serialization, and network bandwidth proportional to the
*whole* fact table rather than
to the matching subset.
The `SEMI`-join path already solves the analogous problem by replacing the
join with a leaf `IN` filter,
but that rewrite is only legal for semi-joins (which emit left columns
only). An inner join projects
columns from both sides, so the join must still run. This PR therefore makes
the filter **additive**:
the real hash join is left completely intact, and we only *add* a reducer on
the probe leaf.
## How it works
After exchange insertion (`POST_LOGICAL`),
`PinotJoinToInnerRuntimeFilterRule` rewrites an eligible
inner join:
```
[ Inner Join ] (unchanged — still hash-shuffles both sides)
/ \
[xChange L] [xChange R]
/ \
[RuntimeFilter] [build subtree]
/ \
[probe leaf] [PIPELINE_BREAKER xChange]
|
[ build keys: Project(rightKeys) -> Filter(IS NOT NULL) ->
limit(maxBuildRows + 1) ]
```
- The join and **both** of its HASH exchanges are kept verbatim — execution
and results are identical to
before; the filter is purely additive.
- A new `RuntimeFilterRel` / `RuntimeFilterNode` is grafted on top of the
probe leaf subtree
(`input[0]` = probe pipeline, pass-through; `input[1]` = a
`PIPELINE_BREAKER` mailbox carrying the
distinct build-side join keys). The pipeline breaker runs the build side
first and ships its keys to
the probe-leaf worker, reusing the same mechanism as the SEMI dynamic
broadcast.
- At the probe leaf, `ServerPlanRequestUtils.attachRuntimeFilter` ANDs a
**tiered, no-false-negative**
reducer onto the V1 leaf query:
- **Exact `IN`** for small key sets (≤ a distinct-value threshold), or for
multi-key / `BIG_DECIMAL`
keys. This is index-accelerated and drives segment pruning.
- **Bloom filter** (`IN_ID_SET`) above the threshold, plus a `BETWEEN(min,
max)` range predicate for
numeric keys to enable cheap range-based segment pruning. Bloom keeps
the wire/heap footprint bounded
for high-cardinality build sides.
- Because the real hash join is the source of truth, the reducer can be
**abandoned at any point**
(empty/over-cap build, oversized bloom, unsupported subtree, mixed-version
cluster) with no effect on
results. Bloom false positives are simply re-checked and discarded by the
join.
This mirrors runtime/dynamic filtering in Trino, Impala, and Spark's
`InjectRuntimeFilter`.
## When it helps
- Large fact table joined to a **small or selectively-filtered**
dimension/build side.
- The probe side is a leaf scan (table scan, optionally with single-input
`Project`/`Filter`), so the
filter can be pushed all the way down to segment scan.
It is **not** beneficial (and is best left off) when the build side is large
or non-selective, or the
probe is cheap — there is no cost-based gate yet, so enablement is opt-in
(see below).
## How to use
Per-join hint (selects the reducer mode):
```sql
SELECT /*+ joinOptions(runtime_filter='auto') */ ...
FROM fact JOIN dim ON fact.key = dim.key
WHERE dim.attr = 'x'
```
`runtime_filter` accepts `off` | `in` | `bloom` | `auto` (exact `IN` below
the threshold, else bloom).
Cluster-wide default (enable/disable only; defaults to `auto` when on):
```
pinot.broker.enable.runtime.filter.join=true
```
Per-query override: `SET runtimeFilterJoin='on'` (or `off`).
## Correctness & safety
- **No false negatives.** Exact `IN` is exact; a bloom never reports
present-as-absent and its false
positives are discarded by the real join; the `BETWEEN(min, max)` bounds
cover every build key.
- **Null keys** are excluded (they cannot match an inner equi-join) both at
the planner (`IS NOT NULL`)
and defensively at the leaf.
- **`NaN`** float/double build keys keep the bloom membership but skip the
range predicate (a finite
range would wrongly drop probe `NaN` rows).
- **Truncation-safe.** The build-key stage is capped at `maxBuildRows + 1`;
if the cap is hit the key set
is incomplete, so the filter is abandoned (the planner cap and the leaf
abandon use the same constant).
- **Mixed-version.** The only wire change is the new `RuntimeFilterNode`
proto variant; the default-off
flag is the guard. Enabling the flag (or using the hint)
mid-rolling-upgrade can fail queries on
not-yet-upgraded servers — documented on the config constant.
- The feature is engine-agnostic: it lives entirely in the planner + Java
leaf, so both the default and
DataFusion worker runtimes honor it with no native changes.
## Testing
- `PinotJoinToInnerRuntimeFilterRuleTest` — rule firing/plan shape,
probe-key/build-key value alignment,
multi-key, hint/flag/query-option gating, pipeline-breaker distribution,
negative cases.
- `ServerPlanRequestUtilsTest` — exact-IN, bloom + range-prune, AUTO
tiering, multi-key, empty/all-null
build, null-key skip, `NaN` range omission, `maxBytes`/`maxBuildRows`
abandon, existing-filter merge.
- `PlanNodeDeserializerTest` — mixed-version graceful failure on the new
proto variant.
- `RuntimeFilterJoinIntegrationTest` — end-to-end cluster self-joins
asserting results are **identical**
with the filter on (in/bloom/auto) and off, across
INT/LONG/DOUBLE/STRING/mixed-type/null/multi-key/
empty-build cases.
- Full `pinot-query-planner` (1321) and `pinot-query-runtime` (4431) suites
pass — no regressions.
## Limitations / future work
- No cost-based auto-enablement yet (opt-in via hint/flag); a future change
can gate it on cardinality/
selectivity stats.
- The build side is materialized a second time for the key broadcast; a
shared spool would avoid this.
- Bloom is single-key (composite-key tuple-encoding deferred); partitioned
(both-sides-hash) joins use
the broadcast path. Only the logical (HEP) planner is wired; the
cost-based physical optimizer is a
follow-up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
