Problem
v2's multi-fragment plans lose the probe-side dynamic-filter reduction once the
probe is partitioned: Velox's runtime DF (hash-join build → probe TableScan)
does not cross an exchange, so once the probe is shuffled the reduction is gone.
At 4 workers this is a real regression:
- q17 — subquery
lineitem scan runs the full 60M rows vs. the single-node
60M→61k reduction (sf10 +286%, sf1 +31%).
- q9_alt — sf10 +22%.
Scope
Cross-fragment only: the target scan is separated from the join by ≥1 exchange
(partitioned join, or broadcast join with an upstream shuffle). The same-fragment
case (single-node, co-located/broadcast with no upstream probe shuffle) is already
handled by Velox's adaptive pushdown and is untouched.
Design
Plan of record: DistributedDynamicFilters.md (added in #1564).
Split of responsibilities:
- Optimizer statically places each DF at plan time — producer join, consumer
scan (via a filter-pushdown-legality walk through projects / aggregations on
grouping keys / across the exchange), key mapping, bloom sizing from the total
build estimate, and a selectivity/cost gate. Emits DF descriptors on the plan
nodes and MultiFragmentPlan. One DF per join key.
- Runner (
LocalRunner) aggregates the build-side filters across tasks
(collectAll → OR-union for partitioned, collectAny → first for broadcast;
any opt-out → disable) and delivers the global filter to every pre-shuffle
probe scan. No task talks to another; the filter flows Task → Runner → Task.
- Velox exposes the per-task seam (tracked separately — see the Velox issue).
Work items
Acceptance
q17 and q9_alt at 4 workers, sf1/sf10: reductions restored (q17 lineitem
60M→~61k), results unchanged, total-CPU regression closed.
Problem
v2's multi-fragment plans lose the probe-side dynamic-filter reduction once the
probe is partitioned: Velox's runtime DF (hash-join build → probe
TableScan)does not cross an exchange, so once the probe is shuffled the reduction is gone.
At 4 workers this is a real regression:
lineitemscan runs the full 60M rows vs. the single-node60M→61k reduction (sf10 +286%, sf1 +31%).
Scope
Cross-fragment only: the target scan is separated from the join by ≥1 exchange
(partitioned join, or broadcast join with an upstream shuffle). The same-fragment
case (single-node, co-located/broadcast with no upstream probe shuffle) is already
handled by Velox's adaptive pushdown and is untouched.
Design
Plan of record: DistributedDynamicFilters.md (added in #1564).
Split of responsibilities:
scan (via a filter-pushdown-legality walk through projects / aggregations on
grouping keys / across the exchange), key mapping, bloom sizing from the total
build estimate, and a selectivity/cost gate. Emits DF descriptors on the plan
nodes and
MultiFragmentPlan. One DF per join key.LocalRunner) aggregates the build-side filters across tasks(
collectAll→ OR-union for partitioned,collectAny→ first for broadcast;any opt-out → disable) and delivers the global filter to every pre-shuffle
probe scan. No task talks to another; the filter flows Task → Runner → Task.
Work items
+ cross-fragment gate; emit DF descriptors on
MultiFragmentPlanand plannodes.
disable, deliver to consumer scans.
Taskcalls +TableScanwait(separate Velox issue).
(plan + results); e2e correctness (
LocalRunner, DF on == off == single-node);perf.
Acceptance
q17 and q9_alt at 4 workers, sf1/sf10: reductions restored (q17
lineitem60M→~61k), results unchanged, total-CPU regression closed.