Skip to content

[do not review] coordinator: forward merged dynamic filters to consumers - #637

Open
jayshrivastava wants to merge 2 commits into
js/4-merge-dynamic-filtersfrom
js/5-forward-dynamic-filters
Open

[do not review] coordinator: forward merged dynamic filters to consumers#637
jayshrivastava wants to merge 2 commits into
js/4-merge-dynamic-filtersfrom
js/5-forward-dynamic-filters

Conversation

@jayshrivastava

@jayshrivastava jayshrivastava commented Aug 13, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

@jayshrivastava jayshrivastava changed the title feat: forward merged dynamic filters to consumers coordinator: forward merged dynamic filters to consumers Aug 13, 2026
This was referenced Aug 13, 2026
Merged
Open
Open
Open
Open
@jayshrivastava
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from 3229903 to ad77338 Compare August 13, 2026 19:27
@jayshrivastava
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from ad77338 to 15e6602 Compare August 17, 2026 18:54
@jayshrivastava
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch 2 times, most recently from 56fee53 to 1572836 Compare August 17, 2026 19:38
@jayshrivastava
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from 1572836 to 296026d Compare August 18, 2026 18:15
@jayshrivastava
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch 2 times, most recently from e04ceac to e720301 Compare August 21, 2026 16:28
@jayshrivastava
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from e720301 to 7a74a70 Compare August 21, 2026 16:48
@jayshrivastava
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from 7a74a70 to c1b6398 Compare August 21, 2026 20:58
@jayshrivastava
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from c1b6398 to 6a70c73 Compare August 22, 2026 15:00
@jayshrivastava
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch 2 times, most recently from 0a68111 to 42cb2ac Compare August 23, 2026 15:27
@jayshrivastava
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from 42cb2ac to f9c7c5d Compare August 23, 2026 15:35
@jayshrivastava jayshrivastava mentioned this pull request Aug 23, 2026
Open
@jayshrivastava
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from f9c7c5d to 9a949f2 Compare August 23, 2026 19:07
@jayshrivastava
jayshrivastava marked this pull request as ready for review August 23, 2026 19:28
@jayshrivastava
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from d0694d9 to e77fd82 Compare September 3, 2026 22:03
jayshrivastava added a commit that referenced this pull request Sep 8, 2026
@jayshrivastava
coordinator: display consumer dynamic filters after execution (#623)
e5bca18 
## 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
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from e77fd82 to 0abbbff Compare September 8, 2026 13:28
@jayshrivastava
feat: forward merged dynamic filters to consumers
a490525 
Route each completed coordinator merge over the existing query-scoped
task stream. Deliver updates at most once, keep runtime filtering fail-open,
and release retained senders before signaling query EOS.

    producer reports --> coordinator merge
                              |
                    +---------+---------+
                    v                   v
              consumer task 0     consumer task 1
@jayshrivastava
remove docs
d027946
@jayshrivastava
jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from 0abbbff to d027946 Compare September 9, 2026 02:03
@jayshrivastava jayshrivastava changed the title coordinator: forward merged dynamic filters to consumers [do not review] coordinator: forward merged dynamic filters to consumers Sep 9, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

1 participant

@jayshrivastava