feat: plan distributed dynamic filters - #634
Conversation
a3e5a8d to
6cb1c46
Compare
8947cd1 to
1d75f65
Compare
1d75f65 to
c54c017
Compare
c54c017 to
90cba3a
Compare
90cba3a to
ec0b443
Compare
acfb3b1 to
918839a
Compare
…che#24601) ## Which issue does this PR close? - Related to apache#18856 - Informs datafusion-contrib/datafusion-distributed#634 Does not close apache#18856: `PushedDown::No` still conflates "I will not use this filter" with "I will use it, but not for exact row-level filtering". This PR only stops that ambiguity from forcing a runtime decision. ## Rationale for this change `HashJoinExec` decides whether to compute a dynamic filter inside `execute()`, by walking the probe subtree looking for a node that holds the filter expression: ```rust // Only compute a dynamic filter when the probe subtree contains a consumer. let enable_dynamic_filter_pushdown = ... .map(|id| plan_contains_expression_id(&self.right, id)) ``` Whether a consumer exists is a planning-time property. Deciding it at execution time breaks any consumer that rewrites the plan after optimization. The concrete case is a distributed planner splitting the optimized plan into stages: ``` worker 1 HashJoinExec (dynamic filter) NetworkShuffleExec worker 2 DataSourceExec (consumes the dynamic filter) ``` At execution time on worker 1 the probe subtree ends at the network boundary, so the traversal finds nothing and the filter is silently never produced — even though the pushdown had found a consumer while the plan was still whole. Working around this requires the shuffle node to hold "anchor" references to filters it never evaluates, purely so the traversal sees them. The check itself is well motivated (apache#17527: skip build-side bounds accumulation when nothing will read the result). Its placement in `execute()` is a leftover from apache#19546, which implemented it as `Arc::strong_count`, a signal only meaningful once the whole plan is assembled. Since apache#24018 replaced refcounting with `expression_id` + `apply_expressions`, that constraint is gone — and `AggregateExec` already makes the same decision at planning time. ## What changes are included in this PR? - `HashJoinExec::handle_child_pushdown_result` runs the consumer check and only attaches the dynamic filter if the probe subtree contains a consumer, mirroring `AggregateExec::handle_child_pushdown_result`. - `HashJoinExec::execute` reduces to `self.dynamic_filter.is_some()`. - Documents the resulting contract on `HashJoinExec::with_dynamic_filter_expr`: holding a dynamic filter is what makes the join compute one, so a caller wiring one up by hand owns the consumer check. This is safe because the Post phase `FilterPushdown` rule is the last rule that mutates the plan (only `SanityCheckPlan` follows, which changes nothing), and the optimizer calls `handle_child_pushdown_result` on the node with its post-pushdown children already in place. The decision then travels as node state, surviving `replace_children` and the proto round trip. No new API, no new `PushedDown` state. As before, the discriminant is not consulted, because a node replying `PushedDown::No` may still retain the filter for statistics pruning. ## Are these changes tested? Yes. - `test_hashjoin_dynamic_filter_pushdown_is_used` is renamed to `test_hashjoin_dynamic_filter_requires_probe_consumer` (the old name referred to the now-deprecated `is_used()`) and strengthened: with no consumer the join now produces no dynamic filter at all, rather than producing one nothing reads. - New `test_hashjoin_dynamic_filter_survives_probe_subtree_replacement` reproduces the stage split — it runs filter pushdown, replaces the probe subtree with an equivalent scan that does not hold the filter, executes, and asserts the build-side bounds were still published. Both fail without the `exec.rs` change. Full workspace extended tests, sqllogictest, and `./dev/rust_lint.sh` pass. ## Are there any user-facing changes? One behavior change worth calling out: a `HashJoinExec` given a dynamic filter outside the filter pushdown rule (via the public `with_dynamic_filter_expr`) now computes it, where previously the runtime traversal could silently disable it. That is the point of the change — it is what lets a plan rewritten after optimization keep producing filters — but it does change the meaning of a public API, so this may warrant the `api change` label. A minor side effect: `gather_filters_for_pushdown` only pushes a self filter when `dynamic_filter.is_none()`, so on a plan with no consumer a repeated Post-phase run now creates and pushes a fresh filter instead of finding one already attached. Same result, slightly more work in replan loops. ## Note on overlapping work @jayshrivastava raised this in apache#18856 (comment) and has apache#24528 open, which adds a third `PushedDown` state to reach the same goal. This is the smaller alternative: it removes the runtime check without changing the pushdown protocol. It is also only possible because of the `apply_expressions` work in apache#24018. Happy to close this in favour of that approach if preferred. --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com> Co-authored-by: Jayant Shrivastava <jshrivastava03@gmail.com>
02092fc to
f02eb8b
Compare
## Stack This stack of PRs implements distributed dynamic filtering #528 1. #623 <- you are here 2. #634 3. #635 4. #636 5. #637 6. #639 Closes: #529 ## Problem Post df-55 upgrade, dynamic filters should work in the worker-local case. There's no way to observe them working other than looking at metrics. ``` ┌───── Stage 2 ── tasks=1 │ AggregateExec: Final COUNT(*) │ [Stage 1] => NetworkCoalesceExec └────────────────────────────────────────────────── ┌───── Stage 1 ── tasks=2 │ HashJoinExec: orders.customer_id = selected_customers.customer_id │ DistributedLeafExec: | ... │ DistributedLeafExec: │ t0: DataSourceExec: predicate=DynamicFilter [ empty ] │ t1: DataSourceExec: predicate=DynamicFilter [ empty ] └──────────────────────────────────────────────── ``` Ideally we want the final filters visible when displaying plans. ## Solution This PR adds a new protocol which is basically identical to the metrics protocol. Even the `MetricsStore` is now just `Store` and is generic over `TaskMetrics` and `TaskCompletedDynamicFilters` (contains completed dynamic filters for a task). ```rust pub(crate) type MetricsStore = Store<TaskMetrics>; pub(crate) type CompletedDynamicFilterStore = Store<TaskCompletedDynamicFilters>; ``` Similar to the metrics protocol, workers now collect completed dynamic filters and send them back to the coordinator. ``` Coordinator Worker ----------- ------ Create independent display copies | +-- SetPlan(task 0, filter IDs) -------> Decode plan | | | | execute | | | | | | | | | | task finishes | v |<----- TaskDynamicFilters ----- Serialize completed filters from the consumers | v ``` Then, at display time, we call `apply_reports_to_distributed_leaves` which traverses the `plan_for_viz` and updates the dynamic filters for all the variants: ``` DistributedLeafExec task 0: DynamicFilter [ key@0 >= 1 AND key@0 <= 10 ] task 1: DynamicFilter [ empty ] ``` ## Notes ### Duplicate RPC Messages We will eventually have more dynamic filter RPCs which manage the worker -> coordinator -> merge -> worker flow mentioned in #553. In theory, the coordinator will know at `merge` time what the completed filters are, making the `TaskCompletedDynamicFilters` and final worker -> coordinator message in this PR irrelevant. However, I think having these mechanisms be separate is good because a) it helps us validate that the dynamic filter coordinator -> worker flow work using external "oracle", and b) there's no guarantee that the coordinator -> worker propagation happens before the query is done (ex. the `DataSourceExec` may not block execution waiting for dynamic filters), so it's good to have a separate way to know if the final `DataSourceExec` applied a filter or not. ### `AND true` and empty filters ``` DynamicFilter [ sr_returned_date_sk@0 >= 2451545 AND sr_returned_date_sk@0 <= 2451910 AND true ] AND DynamicFilter [ empty ] ``` In this filter `AND true` occurs because of apache/datafusion#24277. The first `DynamicFilter` is active but we lose the `HashTableLookupExpr` when serializing it to send back to the coordinator. The 2nd filter is `DynamicFilter [ empty ]` because this is a dynamic filter produced by a remote producer, which does not get propagated to this node yet. This will be fixed later. ### Displaying Dynamic Filters Protocol is as similar to the metrics protocol as possible. Due to double wrapping (`MetricsWrapperExec` wraps `DistributedLeafExec`, it's tricky to do the dynamic filter rewrite after doing the metrics rewrite. So `rewrite_distributed_plan_with_dynamic_filters` has to be called **first**. ```rust let plan = rewrite_distributed_plan_with_dynamic_filters(plan).await?; let plan = rewrite_distributed_plan_with_metrics(plan, DistributedMetricsFormat::Aggregated).await?; println!("{}", display_plan_ascii(plan.as_ref(), true)); ``` ## Testing - Tests in `tests/dynamic_filtering.rs`
015deb6 to
30804a5
Compare
|
|
||
| #[derive(Default)] | ||
| pub(super) struct DynamicFilterRegistryState { | ||
| pub(super) filters: HashMap<u64, PlannedDynamicFilter>, |
There was a problem hiding this comment.
In future PRs, we add more mutex protected state in this struct.
| // one in-memory dynamic filter. This ends up being a race between two writers and two readers. | ||
| // For a partitioned hash join, a producer may update its task-local consumer in memory, but | ||
| // updates must not cross task boundaries. | ||
| pub(crate) fn maybe_roundtrip_plan_to_sever_in_memory_dynamic_filter_relationships( |
There was a problem hiding this comment.
I added more failure modes here and an integration test, colocated_local_dynamic_filters to cover this. I think it's safest to roundtrip any plan that has dynamic filters.
|
benchmarks run tpch/sf100 |
|
Requested by this comment. Benchmark job 50 failed for |
|
benchmarks run tpch/sf100 |
|
Requested by this comment. Benchmark resultsCompared: PR base === Comparing tpch/sf100 results 'datafusion-benchmark-base' [prev] with 'datafusion-benchmark-head' [new] === TASKS: prev=1286.0, new=1286.0, diff=no change (sum of per-query averages) TOTAL: prev=73065 ms, new=75487 ms, diff=1.03 slower ✖ Show full query output q1: prev=3030 ms, new=2677 ms, diff=1.13 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
q2: prev=1679 ms, new=1686 ms, diff=1.00 slower ✖, tasks: prev=94.0, new=94.0, diff=no change
q3: prev=3205 ms, new=3232 ms, diff=1.01 slower ✖, tasks: prev=56.0, new=56.0, diff=no change
q4: prev=1339 ms, new=1530 ms, diff=1.14 slower ✖, tasks: prev=48.0, new=48.0, diff=no change
q5: prev=4114 ms, new=4267 ms, diff=1.04 slower ✖, tasks: prev=83.0, new=83.0, diff=no change
q6: prev=1538 ms, new=1564 ms, diff=1.02 slower ✖, tasks: prev=12.0, new=12.0, diff=no change
q7: prev=4935 ms, new=4995 ms, diff=1.01 slower ✖, tasks: prev=83.0, new=83.0, diff=no change
q8: prev=5106 ms, new=5299 ms, diff=1.04 slower ✖, tasks: prev=100.0, new=100.0, diff=no change
q9: prev=6699 ms, new=6790 ms, diff=1.01 slower ✖, tasks: prev=102.0, new=102.0, diff=no change
q10: prev=5237 ms, new=5664 ms, diff=1.08 slower ✖, tasks: prev=69.0, new=69.0, diff=no change
q11: prev=1076 ms, new=1127 ms, diff=1.05 slower ✖, tasks: prev=65.0, new=65.0, diff=no change
q12: prev=2024 ms, new=2187 ms, diff=1.08 slower ✖, tasks: prev=48.0, new=48.0, diff=no change
q13: prev=1746 ms, new=1817 ms, diff=1.04 slower ✖, tasks: prev=44.0, new=44.0, diff=no change
q14: prev=2054 ms, new=2026 ms, diff=1.01 faster ✔, tasks: prev=28.0, new=28.0, diff=no change
q15: prev=4273 ms, new=4212 ms, diff=1.01 faster ✔, tasks: prev=50.0, new=50.0, diff=no change
q16: prev= 824 ms, new= 844 ms, diff=1.02 slower ✖, tasks: prev=53.0, new=53.0, diff=no change
q17: prev=4979 ms, new=5123 ms, diff=1.03 slower ✖, tasks: prev=40.0, new=40.0, diff=no change
q18: prev=5782 ms, new=5858 ms, diff=1.01 slower ✖, tasks: prev=68.0, new=68.0, diff=no change
q19: prev=1976 ms, new=2372 ms, diff=1.20 slower ✖, tasks: prev=28.0, new=28.0, diff=no change
q20: prev=2767 ms, new=3077 ms, diff=1.11 slower ✖, tasks: prev=65.0, new=65.0, diff=no change
q21: prev=7862 ms, new=8238 ms, diff=1.05 slower ✖, tasks: prev=86.0, new=86.0, diff=no change
q22: prev= 820 ms, new= 902 ms, diff=1.10 slower ✖, tasks: prev=40.0, new=40.0, diff=no change
Verification and run detailsJob
Workload: Capacity: 12 Other timings: Queue 1s · Dataset validation 0s · Total 18m 24s |
|
|
||
| // Test that we correctly register producers and consumers while ignoring anchors. | ||
| #[test] | ||
| fn registers_dynamic_filters_by_expression_id() -> Result<()> { |
There was a problem hiding this comment.
This test seems a bit too big and verbose, and I seems a bit coupled to the specific implementation detail of the DynamicFilterRegistry.
I think we should be able to cover this with an integration test, that tests this just through the public API instead.
There was a problem hiding this comment.
Given that PlannedDynamicFilter is mostly just a placeholder implementation for now, I think we should be good with just removing this test.
|
|
||
| #[test] | ||
| fn identifies_a_producer_without_a_local_consumer() -> Result<()> { | ||
| fn distinguishes_consumers_from_network_boundary_anchors() -> Result<()> { |
There was a problem hiding this comment.
This is one of the issues with testing at the private module level: as the code evolves, the test needs to evolve with it, introducing maintenance burden and making the tests inefective for future contributions.
As dynamic filter integration evolves, it could make sense to think about moving these to just integration tests that stay stable across contributions.
There was a problem hiding this comment.
Great point. I pushed a change which I think you will like.
The tests in src/dynamic_filtering/discovery.rs are now gone in favor of a test in tests/dynamic_filtering/discovery.rs where we use SQL and snapshot the plan.
| pub(crate) properties: Arc<PlanProperties>, | ||
| pub(crate) input_stage: Stage, | ||
| pub(crate) worker_connections: WorkerConnectionPool, | ||
| pub(crate) dynamic_filter_anchors: Vec<Arc<dyn PhysicalExpr>>, |
There was a problem hiding this comment.
I see the anchors are present at every time, however, I think it might make sense to move them to Stage::Remote { .., dynamic_filter_anchors } instead.
The reason is that, if the input_stage is Stage::Local, execution plan traversal will work as normal, and you should not need to resource to artifical anchors for discovering dynamic filters.
However, if input_stage == Stage::Remote, the plan got broken down into fragments, and the fragment below is not recursible. This seems like the only moment we really need to resource to dynamic filter anchors as a workaround for discovering dynamic filters.
There was a problem hiding this comment.
I bet that placing dynamic_filter_anchors would also provide a cleaner implementation, with less repetition, and better encapsulation of the fact that we are capturing the anchors specifically because of the fact that we are transitioning a Stage to Stage::Remote, losing access to the plan below.
There was a problem hiding this comment.
I've pushed this in the last commit. Let me know what you think. The difference in lines of code is about the same.
I agree that it makes sense for the anchors to live in Stage::Remote though. That way we keep the NetworkBoundary implementations clean.
I haven't updated the PR description yet in case we would like to revert that.
There was a problem hiding this comment.
One side effect is that the insertion of anchors happens in prepare static/dynamic plan now instead of inject_network_boundaries
| self.dynamic_filter_registry | ||
| .register_task(&specialized, task_key)?; | ||
|
|
There was a problem hiding this comment.
👍 very naturally handled
Before stage split Producer worker plan
HashJoin producer F HashJoin producer F
└── NetworkShuffle └── NetworkShuffle
└── Local stage ├── Remote stage
└── consumer F └── anchor F
|
apply_expressions() finds F
Register task-specialized producer and consumer topology independently from completed-consumer display state. Preserve consumers crossing a stage boundary as non-evaluating network-boundary anchors so both static and dynamic planning retain the information needed to identify remote filters.
820f92a to
9866970
Compare
| } | ||
| } | ||
|
|
||
| /// Colocates all tasks on the same worker by choosing a URL once and caching it. |
There was a problem hiding this comment.
This is new, as a part of the rebase. Previously, this was in tests/dynamic_filtering/common.rs
55e34b7 to
7b28bdb
Compare
|
@gabotechs This is ready for another review. |
847ccf7 to
f3c06e2
Compare
f3c06e2 to
2b1213f
Compare
Stack
This stack of PRs implements distributed dynamic filtering #528
Goal
The coordinator should know what dynamic filters exist and where to route updates.
Details
1. Dynamic Filter Registry
Each
PlannedDynamicFilterstoresThis will be used in future PRs to store incoming dynamic filter updates from workers and determine how/where to forward the updates.
Implementation
In the
StageCoordinator, we send every task to the registry and extract dynamic filters.2. Network Anchors
In this situation, the hash join does an
execute()-time check to determine if it should update its dynamic filter. It checks to see if the filter is used by any children usingapply_expressions(beforeapply_expressionswas added upstream, it was an Arc pointer strong count check to see if there were multiple references).The join sees that no plan nodes below it use the filter, so it decides not to update it.
Ideally, the hash join decides at optimization time, before distributed planning. I've opened a discussion here about it: apache/datafusion#18856 (comment). While that issue is being resolved, I propose this workaround:
We create an "anchor" to make it seem like the
NetworkShuffleExecuses the filter.Network Anchors Implementation
The implementation adds serialization overhead but is simpler. In static and dynamic planning, we recursively propagate all anchors upwards in the plan to all the network boundaries. We can revisit this implementation in future iterations. This recursive implementation is in
inject_network_boundaries.This means we serialize 8 filters in total.
However, the minimal anchors you need are like this:
In this plan, we would serialize 6 filters.
For 1 dynamic filter, the minimum filters you need to serialize are 1 (producer) + N (consumers) + 1 (network boundary). In this implementation, we serialize 1 (producer) + N (consumers) + M (all network boundaries above the consumer)
Other Notes
See #528. During dynamic planning, the sampler on the probe side of a hash join may overreport rows / cost because dynamic filters aren't being applied yet.