Repository navigation
feat: add zero-wait distributed runtime filtering to AQE - #2749
wirybeaver wants to merge 3 commits into
Conversation
github-actions
Bot
added
the
documentation
ff642c2 to
957f55a
Compare
|
@andygrove The TPC-H benchmark shows ~10% improvement for the Q19. |
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Disclaimer: this review was generated by an LLM (Claude Code) at my request, against 957f55ae6. File/line references were checked against the code and the planner behaviour below was reproduced with scratch planner tests, but please treat the conclusions as suggestions and push back where it's wrong.
Thanks for this, and for gating it behind a default-off flag and posting TPC-H numbers. I couldn't find a wrong-result path:
- Filtering either input is safe for inner and semi joins.
- NULL-safe equality keeps NULL keys only when the build saw one.
- Filters never go below a fetch or a non-value-preserving operator.
- A payload is bound only when every successful producer task reported.
- The fixed-seed foldhash state hashes the same on every executor.
It also merges cleanly with current main (after #2682 and #2671), and the scheduler lib tests pass on the merged tree.
My main question is where the cross-stage path fires today.
1. The partitioned fallback never gets the filter. With stage_build_side on (the default), a build side that measures too large leaves a pending probe exchange that is created after the build has finished. That is the one case where a complete filter can cut the probe's shuffle with no waiting. But discover_pair returns as soon as the build exchange is shuffle_created() (runtime_filter_planner.rs:113), so the probe stage is planned without an apply and the build tap's work is wasted. Reproduced with MemTable scans, a filter on both sides and broadcast_join_threshold_bytes = 500:
pass 1, stage 0:
SortShuffleWriterExec: partitioning=Hash([id@0], 4)
RuntimeFilterBuildExec: id=0, key=id@0
FilterExec: val@1 > 0
DataSourceExec: partitions=4, partition_sizes=[1, 1, 1, 1]
build measured too large, pass 2, stage 1:
SortShuffleWriterExec: partitioning=Hash([id@0], 4)
FilterExec: val@1 > 0, projection=[id@0]
DataSourceExec: partitions=4, partition_sizes=[1, 1, 1, 1]
When the build already has a registered filter, binding a not-yet-planned probe exchange to it looks safe, and prepare would find the summary complete at first dispatch.
2. Sibling stages mostly bypass. When both sides are shuffled at once, the apply is planned in the probe stage. It only binds if the build finishes before the probe's first dispatch, which is unusual when slots are free. prepare also freezes the plan when bind_tasks_with merely fetches the stage (cluster/mod.rs:754), even if no task is bound in that round. Until the waiting follow-up lands, would it be worth not tapping builds whose only consumer is a sibling probe?
So today the path that reliably fires is a broadcast join whose probe scan sits in the same task as the build reader (Q19). It would help to say that in the docs section.
3. Bloom sizing. With 8 KiB and k=4, the false-positive rate is about:
- 0.4% at Q19's 4,754 keys
- 4% at 10k
- 50% at ~30k
- 99% at 100k
RuntimeFilterApplyExec only skips at full saturation (~191k distinct keys), so in between every probe row is hashed for little pruning. prepare already holds the merged Bloom, so a cheap guard is to bypass when its fill ratio implies a high false-positive rate. The MinMax bounds aren't used for row filtering either; a range check is cheap and stays selective after the Bloom saturates.
4. Tests. runtime_filters_preserve_adaptive_planning (stage_build_side.rs:171) can't fail. Its StatsExec tables aren't DataSourceExec, so no filter is ever placed; I checked, and there is no tap or apply in either pass. assert_plan! snapshots of the instrumented stage plans over MemTable scans would pin the plan shape the way the other AQE rules' tests do. Cover the broadcast, fallback and sibling flows. That would also cover the EXPLAIN diff I asked for on #2701. Plain EXPLAIN renders the static planner's stages, so these operators only appear under EXPLAIN ANALYZE; an excerpt for Q19 in the PR body would help.
5. Benchmark details. Could you add the cluster setup, the settings used for A and B (AQE, prefer_hash_join, stage_build_side), and the run count? With three runs per side the ±3% rows are hard to read; Q19 looks real. The numbers predate #2671, whose distinct counts feed the join estimates, so a re-run on main would be good. TPC-DS star joins would exercise this more than TPC-H.
Smaller things:
- Each successful build task keeps an 8–9 KB payload per filter in
TaskInfofor the life of the job, so 10k tasks is about 80 MB per filter. Worth a line in the docs, as #2671 did for HLL. instrumentunwraps the boundary's stage id (runtime_filter_planner.rs:182-188). An internal error would be safer in the scheduler.- A consumer stage that re-runs after a rollback keeps the filter it first bound, because
preparedis never cleared. That's fine for deterministic builds; worth a comment, or rebinding on a new attempt. - configs.md asks users to set
prefer_hash_join = true, butDynamicJoinSelectionExec::to_actual_joinsays AQE deliberately ignores it. Is it needed?
|
Thanks for the review. To clarify, each query has 3 measured runs per arm, thus the total measure number is 3*3 = 9. I’ve updated the PR body to make this explicit. Will evaluate in this PR
I previously removed a broader speculative multi-join producer guard after SF10 experiments did not establish a clear benefit. The narrower sibling-only case still needs evaluation. Already deferred to personal PRs |
wirybeaver
changed the title
Attach taps to executable stage copies without changing AQE planning or stage selection. At first dispatch, freeze only summaries with complete current-attempt coverage and bypass unavailable producers. Keep saturated Bloom hashing bounded while continuing NULL and MinMax collection.
Benchmark Result
The four latency rows above are custom directed SQL scenarios on TPC-DS SF10 data, not standard TPC-DS queries or suite totals. Negative change means lower latency. Four independent local A/Bs after rebasing Task 2: A = unchanged zero-wait Task 2; B = A plus one candidate, with runtime filtering enabled in both. Candidates are not yet integrated into this PR. Environment: one Mac (10 CPUs, 24 GiB), two executors per arm (4 vcores / 4 GiB each), 16 partitions, batch 8192, release without LTO. Each comparison covers 13 unmodified TPC-DS queries (listed below) plus its directed case. Process: one DataFusion-verified preflight per query/arm, one complete-suite warmup, then five full-suite rounds. Pair A → B in rounds 1/3/5 and B → A in 2/4; each arm gets one short warmup immediately followed by one timed request. Servers stay running between visits and restart between candidate comparisons. Latency is client SQL execution through final result collection, excluding registration. No A/A, retries or dropped samples. All 112 preflight checks matched and 560 timed requests completed. Every standard-query change stayed below +3% (maximum +2.92%); five samples do not establish zero regression. C1/C3 show benefits in their directed cases. C1 bound a ready filter in B: 5/5 visits versus A: 0/5. C2 was scheduling-dependent (A: 1/5, B: 2/5 ready), with no clear median gain. C4 removed the producer without improving latency. These results are not additive or whole-suite speedups. Per-visit results — all timed standard and directed queries
SQL for the two accelerated directed cases D1 — selective build followed by partitioned fallback: SELECT SUM(s.ss_ext_sales_price * c.c_birth_day) AS revenue
FROM customer c JOIN store_sales s ON c.c_customer_sk = s.ss_customer_sk
WHERE c.c_birth_month = 1 AND c.c_birth_day = 1
AND s.ss_quantity BETWEEN 10 AND 90;D3 — ready, nearly saturated Bloom: SELECT SUM(s.ss_ext_sales_price * c.c_birth_day) AS revenue
FROM customer c JOIN store_sales s ON c.c_customer_sk = s.ss_customer_sk
WHERE CAST(c.c_birth_month AS VARCHAR) IN ('1', '2')
AND s.ss_quantity BETWEEN 10 AND 90;These forced-plan settings were identical on A and B:
D1 forces the measured build above the broadcast limit, exposing the missing fallback consumer. D3 uses a pessimistic filter estimate to stage the build, then broadcasts its measured result; both arms receive a complete Bloom, and C3 bypasses the near-full one. Common settings: staged execution, AQE and runtime filtering enabled; |
957f55a to
65162ce
Compare
Motivation
Part of #2700 (Task 2), building on the runtime filter operators and task reports.
Use completed build-side summaries to reduce probe-side row processing without adding a runtime-filter wait. Existing broadcast dependencies, as in Q19, can make summaries available before probe execution. Optional nonzero waiting is scoped to Task 4 (personal PR #13).
Modification
Attach build/apply operators only to executable stage copies after AQE chooses stage boundaries. A job-scoped hub merges reports from complete successful producer coverage through the existing task-status lifecycle.
At stage preparation, bind complete summaries already available or bypass unavailable filters, then freeze the stage plan. Scheduling proceeds without waiting for reports; independently runnable sibling probes may therefore execute without filtering.
The feature is disabled by default and covers the first supported direct-column equality key of inner and semi hash joins, including NULL-safe equality. Fully saturated Blooms bypass hashing while NULL and MinMax collection continue.
Benchmarking Result
A = the upstream base of this PR; B = the current PR HEAD with runtime filtering enabled. This is one fresh comparison covering 21 TPC-H SF10 shapes (Q16 excluded), 13 TPC-DS SF10 shapes, and two custom shapes using TPC-DS data. Every row is an individual query median, not a suite total. All 72 preflight checks matched DataFusion, and all 360 timed requests completed.
TPC-H Q3 and Q6 crossed the +3% median-latency screening threshold in this run; these observations are retained and their causes are not established. Q6 has identical executed plans and no runtime-filter operators. The two custom rows are not standard TPC-DS queries or default-configuration scores.
One host; two executors per arm (4 vcores / 4 GiB each), 16 partitions, batch 8192. After one preflight and one complete-suite warmup, five rounds traversed all 36 shapes: A→B within each query in rounds 1/3/5, B→A in 2/4. Each arm ran one short warmup immediately followed by one timed request. Servers stayed running between visits. Latency covers client SQL execution through final result collection, excluding session/table registration. No retries or sample exclusions.
Custom D1 — partitioned fallback
Both arms set
ballista.optimizer.broadcast_join_threshold_bytes=4096andballista.optimizer.stage_build_side_max_estimate_multiple=1024to exercise fallback after the build completes.Custom D3 — near-saturated Bloom
Both arms use
ballista.optimizer.broadcast_join_threshold_bytes=1500000,ballista.optimizer.stage_build_side_max_estimate_multiple=1024,datafusion.optimizer.default_filter_selectivity=100, and DataFusion'shash_join_single_partition_threshold/hash_join_single_partition_threshold_rowsset to0to exercise a staged build followed by broadcast.Common settings: staged execution, AQE enabled,
stage_build_side=true,prefer_hash_join=true; coalescing, skew optimization and local dynamic filter pushdown disabled. Only B enablesballista.planner.runtime_filter.enabled, which is absent from the upstream base.Bloom bypass threshold: 100% vs 99%
Both arms enable runtime filtering; this diagnostic comparison isolates the probe-side bypass threshold (fraction of set Bloom bits), rather than upstream base vs HEAD. At 99%, D3 bypasses low-value Bloom checks while D6 still filters approximately 8.62 million rows. The PR keeps the fixed 99% threshold, with no query option.
D6 uses the D3 SQL above with
CAST(c.c_birth_month AS VARCHAR) = '1',broadcast_join_threshold_bytes=4096,stage_build_side_max_estimate_multiple=65536, andstage_build_side_min_probe_ratio=1(all underballista.optimizer); other settings match D3.Same binary and persistent cluster for both arms, with the executor sizing above. D3 used five paired rounds; D6 used nine: A→B on odd rounds, B→A on even rounds. Each arm had one preflight and one global warmup; each visit had one short warmup immediately followed by one timed query. All four preflight checks matched DataFusion and all 28 timed queries completed, without retries or sample exclusions.
Threshold experiment: per-visit results
Upstream base vs fixed 99% — D3 and D6
This separate paired rerun uses the actual upstream base (no runtime filtering) as A and the PR's fixed-99% implementation as B. It is not an on/off comparison within the PR binary. The D3/D6 SQL and settings are unchanged.
Same host and executor sizing as above, with a persistent, version-matched cluster per arm. After one preflight and one global warmup per arm/query, nine rounds ran D3 then D6, A→B on odd rounds and B→A on even rounds. Each visit had one short warmup followed immediately by one timed query. All four preflight checks matched DataFusion and all 36 timed queries completed; no retries or sample exclusions.
Upstream-base comparison: per-visit results