rfc: distributed dynamic filtering by jayshrivastava · Pull Request #553 · datafusion-contrib/datafusion-distributed

jayshrivastava · 2026-07-14T19:11:58Z

Rendered View: https://github.com/datafusion-contrib/datafusion-distributed/blob/js/dynamic-filtering-design/rfcs/1-distributed-dynamic-filtering.md

Informs #528

introduces the proto converter to the PhysicalExtensionCodec trait which helps dedupe dynamic filters

gabotechs

This is looking very good! I really do not have any major comments, this seems to be going in the right direction.

gabotechs · 2026-07-24T08:24:55Z

+The proposed change is to implement dynamic filtering similarly to Trino.
+
+1. During execution, the dynamic filter producers (`HashJoinExec`, `SortExec`, `AggregateExec`) will send dynamic filter updates to the coordinator.
+2. The coordinator will union / merge the dynamic filter updates from all workers.
+3. The coordinator will send the unioned / merged dynamic filter updates to the consumers (`DataSourceExec`).


👍 Sounds good, with the current infrastructure, I don't think it would be terrible difficult to do this

gabotechs · 2026-07-24T08:33:54Z

+- Less overhead than ORing
+
+Cons:  
+- Is brittle. What if we have to support non-range and non-IN-LIST expressions? It would be nice if dynamic filters in vanilla datafusion natively implemented a `merge` or `union` operation


Why do you think it's brittle? For merging, I imagine you can just rely on Arrow comparison kernels for deduping IN LIST expressions.

Can you expand on this? It's easy to implement a merge operation if we assume the expressions to merge always look like a@0 >= 0 AND a@0 <= 5 OR IN LIST [10, 11]. We would downcast to InListExpr and RangeExpr and define a merge operation for those.

In the future, what if there's other kinds of expressions in there? Then maybe the merge operation would break.

Maybe @LiaCastaneda has more insights on this. I think she implemented something similar.

I imagine that not any arbitrary expression can be placed there. Even if it can, you still decide to optimize the most common cases, like IN LISTs or min/max ranges, and if there's anything else besides does two things just leave it OR-ed.

In the future, what if there's other kinds of expressions in there?

I think that rather than adding other kinds of expressions, there is probably a higher possibility that the overall dynamic filter shape changes from the current CASE statement to something else.
Regarding expression types, I think the other discussed expression that hasn’t been implemented yet is bloom filters pushdown as dynamic filters. In that case, deduping probably becomes awkward, I’m not really sure how you would merge that (or if its possible). But I think it’s a very niche use case, and as Gabriel said, it can be ORed anyway.

gabotechs · 2026-07-24T08:37:33Z

+
+// Read Updates
+pub fn current(&self) -> Result<Arc<dyn PhysicalExpr>>
+pub(crate) fn subscribe(&self) -> DynamicFilterSubscription


🤔 subscribe not being public might be a problem.

True. I think we can just use this instead https://github.com/apache/datafusion/blob/3a29d6bd8cc9ac2bf5efee9f070dcdeea9f97b32/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs?plain=1#L320 since we would wait for completion anyways.

gabotechs · 2026-07-24T08:41:11Z

+
+#### Options 1 & 2: Support new APIs In Vanilla DataFusion
+
+See https://github.com/apache/datafusion/issues/23814


apache/datafusion#23814 seems to be evolving in a good direction, and people seem engaged in the discussion, so I'd lean towards using whatever comes out of that.

Yup I will stay on top of that issue.

barbarj · 2026-08-06T22:47:36Z

+For correctness, the coordinator must wait for all dynamic filter updates from all workers before sending them to the consumers, otherwise, the
+consumers may prune rows incorrectly.


Is this
A) the coordinator must get an update from each worker at least once before sending the updated expression, and then can update eagerly going forward
B) the coordinator must get an update from each worker at least once per filter generation before sending the next generation?
C) or something else?

Relatedly, this is overly strict for SortExec's TopK dynamic filter, as well as AggregateExec's min/max filter, which can safely be updated eagerly. (I've no problem with correctness then optimization as the sequence for shipping this though)

Great point. For hash joins, eager updates aren't safe. The filter effectively contains the build side of the join. If the join is on different workers and each has a different build side, we need filter ORed together to get a global filter which represents the entire build side across workers.

However for sort and aggregate, I think eager updates are very safe. For an aggregate that has MIN(a) for example the aggregate on any worker can push down a < 10 for example. We can OR the expressions across workers ex. a < 10 OR a < 20 and the minimum value, 10 would win. This mechanism is similar for SortExec.

For now, I think we should ensure on correctness / safety in the initial implementation, so I'm proposing option A) for now. I think we can roll out eagerness later. This will mean the dynamic filters on SortExec and AggregateExec will still be pretty useless for now, but we will be able to unlock joins.

Since you worked on joins recently, maybe you can confirm if what I said makes sense

Yeah, after reading up a bit more on HashJoinExec's filter strategy, that makes sense, and A) is a reasonable general approach.

We'll want to consider how we can track each expression's producer node's type, as that will be necessary info for enabling eagerness correctly later.

We'll want to consider how we can track each expression's producer node's type, as that will be necessary info for enabling eagerness correctly later.

Agreed. At least, maybe some property to indicate if its safe to apply eager updates or not.

I pushed a short section in the RFC on eager updates so this is documented.

…he#22437) (apache#24018) ## Which issue does this PR close?  - Informs: apache#23814 - Informs: datafusion-contrib/datafusion-distributed#584 This change does not close the above issues because it does not implement a way to tell if a node is a producer dynamic filters. ## Rationale for this change See apache#23814 and datafusion-contrib/datafusion-distributed#553. To send dynamic filter updates across the network, there needs to be a way to get access to `PhysicalExpr` from `ExecutionPlan`. As discussed in apache#23814, the cleanest way to do this is to add `ExecutionPlan::apply_expressions`, which mirrors a similar method for logical plan nodes. ## What changes are included in this PR? There's 3 commits in this PR: Firstly, commit 1 re-applies the changes in apache#20337 (reverted in apache#22437). Some of the reasons for why the original PR was reverted include (a) `apply_expressions` is too complicated to implement and there's no concrete need to justify this complexity (b) there was no usage of `apply_expressions` inside this repo To address (a) - justification for adding this method is provided in apache#23814 - commit 2 in this PR adds helper methods `apply_expression_roots` and `apply_no_expressions` which abstract away the `TreeNodeRecursion` complexity from implementors. Now, `apply_expressions` very trivial to implement ex. ```rust fn apply_expressions( &self, f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>, ) -> Result<TreeNodeRecursion> { apply_expression_roots([&self.predicate_1], f) apply_expression_roots([&self.predicate_2], f) apply_expression_roots([&self.other_expression], f) } ``` - the method traverses over `&Arc<dyn PhysicalExpr>` rather than `&dyn PhysicalExpr` to reduce complexity around lifetimes To address (b): - commit 3 adds a usage of `apply_expressions` in `physical-plan/src/aggregates/mod.rs`. Previously, there was a hack that checked if a filter was pushed down using `Arc::strong_count(dyn_filter) > 1`. Now it uses `apply_expressions` - similarly, commit 4 removes `is_used` from dynamic filters which used to check Arc references counts to see if a filter was pushed down. Now, the hash join uses `apply_expressions` to find pushed down filters. ## Are these changes tested? Yes. ## Are there any user-facing changes? There's a new mandatory method `ExecutionPlan::apply_expressions()`. See the upgrading guide and documentation for details. --------- Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

shinzoxD

The new eager-update section leaves one liveness boundary to clarify.

shinzoxD · 2026-08-11T16:55:27Z

+#### Sorts  
+These operate similarly to Aggregates and can be applied eagerly.
+
+The proposal in this RFC is to take the safest approach and wait for all filters to complete before


The preceding paragraph says aggregate filters are never marked complete, so a literal wait for all discovered filters can never finish for a plan containing AggregateExec (and presumably SortExec). Could the initial scope explicitly define the required-completion set as join-produced filters only, and state that filter collection must not block query execution? Otherwise an implementation that registers every producer can either never publish these filters or deadlock while awaiting completion. Non-completing producers can then be added when eager updates land.

…he#22437) (apache#24018) ## Which issue does this PR close?  - Informs: apache#23814 - Informs: datafusion-contrib/datafusion-distributed#584 This change does not close the above issues because it does not implement a way to tell if a node is a producer dynamic filters. ## Rationale for this change See apache#23814 and datafusion-contrib/datafusion-distributed#553. To send dynamic filter updates across the network, there needs to be a way to get access to `PhysicalExpr` from `ExecutionPlan`. As discussed in apache#23814, the cleanest way to do this is to add `ExecutionPlan::apply_expressions`, which mirrors a similar method for logical plan nodes. ## What changes are included in this PR? There's 3 commits in this PR: Firstly, commit 1 re-applies the changes in apache#20337 (reverted in apache#22437). Some of the reasons for why the original PR was reverted include (a) `apply_expressions` is too complicated to implement and there's no concrete need to justify this complexity (b) there was no usage of `apply_expressions` inside this repo To address (a) - justification for adding this method is provided in apache#23814 - commit 2 in this PR adds helper methods `apply_expression_roots` and `apply_no_expressions` which abstract away the `TreeNodeRecursion` complexity from implementors. Now, `apply_expressions` very trivial to implement ex. ```rust fn apply_expressions( &self, f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>, ) -> Result<TreeNodeRecursion> { apply_expression_roots([&self.predicate_1], f) apply_expression_roots([&self.predicate_2], f) apply_expression_roots([&self.other_expression], f) } ``` - the method traverses over `&Arc<dyn PhysicalExpr>` rather than `&dyn PhysicalExpr` to reduce complexity around lifetimes To address (b): - commit 3 adds a usage of `apply_expressions` in `physical-plan/src/aggregates/mod.rs`. Previously, there was a hack that checked if a filter was pushed down using `Arc::strong_count(dyn_filter) > 1`. Now it uses `apply_expressions` - similarly, commit 4 removes `is_used` from dynamic filters which used to check Arc references counts to see if a filter was pushed down. Now, the hash join uses `apply_expressions` to find pushed down filters. ## Are these changes tested? Yes. ## Are there any user-facing changes? There's a new mandatory method `ExecutionPlan::apply_expressions()`. See the upgrading guide and documentation for details. --------- Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

## 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`

jayshrivastava added 3 commits July 7, 2026 15:32

update datafusion to 01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b

44d85d8

introduces the proto converter to the PhysicalExtensionCodec trait which helps dedupe dynamic filters

poke

a6c3268

wip

06ee41b

jayshrivastava changed the base branch from main to branch-55 July 14, 2026 19:12

jayshrivastava added 3 commits July 14, 2026 15:33

wip

d808d57

wip

0fefcb3

wip

47a0c51

jayshrivastava changed the title ~~Js/dynamic filtering design~~ rfc: distributed dynamic filtering Jul 14, 2026

jayshrivastava added 4 commits July 14, 2026 16:23

wip

4beaa3d

wip

9524b15

wip

e9c9601

wip

91f6467

jayshrivastava commented Jul 15, 2026

View reviewed changes

Comment thread rfcs/1-distributed-dynamic-filtering.md

wip

c03366c

jayshrivastava changed the title ~~rfc: distributed dynamic filtering~~ [WIP] rfc: distributed dynamic filtering Jul 20, 2026

updates

775fbb8

jayshrivastava mentioned this pull request Jul 22, 2026

make dynamic filters in ExecutionPlan nodes discoverable apache/datafusion#23814

Closed

final updates

e98ff14

jayshrivastava marked this pull request as ready for review July 22, 2026 19:49

jayshrivastava changed the title ~~[WIP] rfc: distributed dynamic filtering~~ rfc: distributed dynamic filtering Jul 22, 2026

jayshrivastava added 2 commits July 22, 2026 15:55

newlines

2caffbb

newlines

1465911

jayshrivastava mentioned this pull request Jul 22, 2026

store partitioning in dynamic filters and remove CASE hash(expr) apache/datafusion#23817

Open

add issue

c3d3d24

gabotechs reviewed Jul 24, 2026

View reviewed changes

This was referenced Aug 3, 2026

Reapply "Add ExecutionPlan::apply_expressions() (apache#20337)" (apache#22437) apache/datafusion#24018

Merged

feat: add dynamic filter store in coordinator #592

Closed

barbarj reviewed Aug 6, 2026

View reviewed changes

add section about eager updates

f715695

shinzoxD reviewed Aug 11, 2026

View reviewed changes

jayshrivastava mentioned this pull request Aug 11, 2026

coordinator: display consumer dynamic filters after execution #623

Merged

jayshrivastava force-pushed the branch-55 branch 2 times, most recently from 78b65f1 to b8eddac Compare August 12, 2026 17:08

gabotechs force-pushed the branch-55 branch from 55b8559 to f5aedab Compare August 13, 2026 11:07

jayshrivastava force-pushed the branch-55 branch 3 times, most recently from 300b5af to e114458 Compare August 19, 2026 21:41

Base automatically changed from branch-55 to main August 20, 2026 08:38

jayshrivastava mentioned this pull request Aug 23, 2026

[dynamic filtering] add docs #666

Open

jayshrivastava mentioned this pull request Sep 21, 2026

[EPIC] dynamic filtering #528

Open


		#### Options 1 & 2: Support new APIs In Vanilla DataFusion

		See https://github.com/apache/datafusion/issues/23814

		For correctness, the coordinator must wait for all dynamic filter updates from all workers before sending them to the consumers, otherwise, the
		consumers may prune rows incorrectly.

Conversation

jayshrivastava commented Jul 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

gabotechs left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

barbarj Aug 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jayshrivastava Aug 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shinzoxD left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

jayshrivastava commented Jul 14, 2026 •

edited

Loading

barbarj Aug 6, 2026 •

edited

Loading

jayshrivastava Aug 11, 2026 •

edited

Loading