feat: forward remote dynamic filter updates to coordinator by jayshrivastava · Pull Request #635 · datafusion-contrib/datafusion-distributed

jayshrivastava · 2026-08-13T13:26:58Z

Stack

This stack of PRs implements distributed dynamic filtering #528

Problem

The QueryCoordinator needs to receive partial dynamic filter updates from workers.

Solution

We introduce a new WorkerToCoordinatorMsg which

message ProducedDynamicFilter {
  uint64 expression_id = 1;
  // Serialized datafusion.proto.PhysicalExprNode.
  bytes expression_proto = 2;
}

In this PR makes each worker unconditionally send updates (via wait_update() and wait_complete()) to the coordinator for any dynamic_filter_remote_producer_ids in the SetPlanRequest. The purpose of dynamic_filter_remote_producer_ids is to exclude any dynamic filters who only have local consumers - these don't need to be forwarded to the coordinator.

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

Task-local consumer producer F ────────────────> consumer F shared expression; no RPC Cross-task consumer producer F ── boundary anchor F ──> SetPlanRequest.report_ids=[F] │ ├── wait_update() -----+ ├── wait_complete() ---+--> latest observed F --RPC--> coordinator └── query cancellation +--> stop Derive the report allowlist from producer IDs intersected with network-boundary anchor IDs. Workers observe only allowlisted producers, while DataFusion continues updating task-local consumers directly in memory. Watch-based updates may naturally coalesce, and completion remains observed separately because it does not advance the generation.

jayshrivastava · 2026-09-09T02:04:45Z

                    }
+                    // Runtime dynamic-filter reports are accepted by this transport change. A
+                    // later change in the stack will retain and merge them.
+                    WorkerToCoordinatorMsg::ProducedDynamicFilter(_) => {}


In this PR, we do nothing when receiving dynamic filter updates. In later PRs, we will consume these updates.

jayshrivastava · 2026-09-09T02:22:10Z

+                // naturally coalesce while this observer is busy. The stream promises the latest
+                // observed state rather than one message per generation; completion is awaited
+                // separately because it does not advance the generation.
+                let completed = tokio::select! {


This PR introduces some overhead. We forward these messages even though they aren't used yet. They will be used in future PRs.

jayshrivastava · 2026-09-09T03:19:17Z

benchmarks run tpch/sf100

gabot-0 · 2026-09-09T03:19:41Z

Requested by this comment.

Benchmark job 51 failed for tpch/sf100. Full details are available in the controller journal.

gabotechs · 2026-09-09T07:49:48Z

+struct SpecializedTaskPlan {
+    plan: Arc<dyn ExecutionPlan>,
+    work_unit_feed_declarations: Vec<WorkUnitFeedDeclaration>,
+    dynamic_filter_remote_producer_ids: Vec<u64>,
+}
+


For this struct, It might be worth to:

Move it below in this file, so that the comment above still applies to StageCoordinator

Name it TaskSpacializedPlan to match the method's name

gabotechs · 2026-09-09T07:53:46Z

+        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
+
+        let local_filter = dynamic_filter();
+        let local_probe = Arc::new(FilterExec::try_new(local_filter.clone(), empty(&schema))?)
+            as Arc<dyn ExecutionPlan>;
+        let local_join = join_with_filter(&schema, local_probe, Arc::clone(&local_filter))?;
+        assert!(dynamic_filter_remote_producer_ids(&local_join)?.is_empty());
+
+        let remote_filter = dynamic_filter();
+        let repartition = Arc::new(RepartitionExec::try_new(
+            empty(&schema),
+            Partitioning::Hash(vec![Arc::new(Column::new("a", 0))], 1),
+        )?) as Arc<dyn ExecutionPlan>;
+        let remote_probe = Arc::new(
+            NetworkShuffleExec::try_new(repartition, 1)?
+                .with_dynamic_filter_anchors(vec![remote_filter.clone()]),
+        ) as Arc<dyn ExecutionPlan>;


I see this being a pattern with the tests: plans get constructed manually by chaining operators together.

It's pretty hard to see what's happening and what's getting chained with what to a human eye. Do you think there's a chance we can just use normal SQL for these tests?

The main reason for having the weather and flights datasets committed to the codebase is so that we can just use SQL for building plans, rather than constructing them manually.

gabotechs · 2026-09-09T08:05:12Z

+        let expression = Arc::clone(&dynamic_filter) as Arc<dyn PhysicalExpr>;
+        let (_cancel_tx, cancel_rx) = watch::channel(false);
+        let task_ctx = SessionContext::new().task_ctx();
+        let mut stream = produced_dynamic_filter_stream(expression_id, expression, cancel_rx);


I really think we should change the approach to testing in general in this PR.

The produced_dynamic_filter_stream is a private function of this module, and therefore an implementation detail subject to change. By testing at this level, we are testing the implementation detail rather than the overall functionality, and the tests tend to be very verbose.

Here's one idea for change the way we approach testing in this PR: for anything related to dynamic filter collection and over-the-wire transfer, we can add some nice DataFusion metrics at the coordinator level, and during integration testing, we can perform assertion on those metrics.

With that, we'd win two things:

Reliable tests that survive changes to the implementation details.

Runtime metrics that give visibility around what happened.

jayshrivastava changed the title ~~plan distributed hash join filters~~ coordinator: plan distributed dynamic filters Aug 13, 2026

jayshrivastava force-pushed the js/3-plan-dynamic-filters branch from 60dc39a to 7bee97e Compare August 13, 2026 19:27

jayshrivastava force-pushed the js/3-plan-dynamic-filters branch from 7bee97e to 53be7d0 Compare August 17, 2026 18:54

jayshrivastava force-pushed the js/3-plan-dynamic-filters branch 2 times, most recently from 4fd1433 to c29bcf1 Compare August 17, 2026 19:38

jayshrivastava changed the title ~~coordinator: plan distributed dynamic filters~~ feat: forward remote dynamic filter updates to coordinator Aug 18, 2026

jayshrivastava force-pushed the js/3-plan-dynamic-filters branch from c29bcf1 to a7bb152 Compare August 18, 2026 18:15

jayshrivastava force-pushed the js/3-plan-dynamic-filters branch from a7bb152 to 8bc2dc9 Compare August 18, 2026 18:52

jayshrivastava force-pushed the js/3-plan-dynamic-filters branch 2 times, most recently from e1a6b47 to 77c9528 Compare August 21, 2026 16:48

jayshrivastava force-pushed the js/3-plan-dynamic-filters branch from 77c9528 to dd39b56 Compare August 21, 2026 20:58

jayshrivastava force-pushed the js/3-plan-dynamic-filters branch from dd39b56 to fceebde Compare August 22, 2026 15:00

jayshrivastava force-pushed the js/3-plan-dynamic-filters branch 2 times, most recently from 5912188 to a60f964 Compare August 23, 2026 15:27

jayshrivastava force-pushed the js/3-plan-dynamic-filters branch from a60f964 to 8d666d6 Compare August 23, 2026 15:35

jayshrivastava marked this pull request as ready for review August 23, 2026 19:28

jayshrivastava force-pushed the js/3-plan-dynamic-filters branch from 51e5eda to 323cbf2 Compare September 3, 2026 22:03

jayshrivastava force-pushed the js/3-plan-dynamic-filters branch from 323cbf2 to 336d392 Compare September 8, 2026 13:28

jayshrivastava and others added 3 commits September 9, 2026 02:01

comments

f3d7dc3

test: provide context to worker message encoder

8a3f76f

jayshrivastava force-pushed the js/3-plan-dynamic-filters branch from 336d392 to 8a3f76f Compare September 9, 2026 02:03

jayshrivastava commented Sep 9, 2026

View reviewed changes

clean up producer cancellation

8be5482

use maybe encoded

22137b1

gabotechs reviewed Sep 9, 2026

View reviewed changes

Navigation Menu

Uh oh!

feat: forward remote dynamic filter updates to coordinator - #635

feat: forward remote dynamic filter updates to coordinator#635
jayshrivastava wants to merge 5 commits into
js/2-forward-dynamic-filter-updates-to-coordinatorfrom
js/3-plan-dynamic-filters

jayshrivastava commented Aug 13, 2026 •

edited

Loading

Uh oh!

jayshrivastava Sep 9, 2026

Uh oh!

jayshrivastava Sep 9, 2026

Uh oh!

jayshrivastava commented Sep 9, 2026

Uh oh!

gabot-0 commented Sep 9, 2026 •

edited

Loading

Uh oh!

gabotechs Sep 9, 2026

Uh oh!

gabotechs Sep 9, 2026

Uh oh!

gabotechs Sep 9, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

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

Uh oh!

Stack

Problem

Solution

Uh oh!

jayshrivastava Sep 9, 2026

Choose a reason for hiding this comment

Uh oh!

jayshrivastava Sep 9, 2026

Choose a reason for hiding this comment

Uh oh!

jayshrivastava commented Sep 9, 2026

Uh oh!

gabot-0 commented Sep 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gabotechs Sep 9, 2026

Choose a reason for hiding this comment

Uh oh!

gabotechs Sep 9, 2026

Choose a reason for hiding this comment

Uh oh!

gabotechs Sep 9, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

jayshrivastava commented Aug 13, 2026 •

edited

Loading

gabot-0 commented Sep 9, 2026 •

edited

Loading