Skip to content

coordinator: merge partial dynamic filters - #636

Merged
jayshrivastava merged 4 commits into
mainfrom
js/4-merge-dynamic-filters
Sep 25, 2026
Merged

coordinator: merge partial dynamic filters #636
jayshrivastava merged 4 commits into
mainfrom
js/4-merge-dynamic-filters

Conversation

@jayshrivastava

@jayshrivastava jayshrivastava commented Aug 13, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Stack

This stack of PRs implements distributed dynamic filtering #528

  1. coordinator: display consumer dynamic filters after execution #623
  2. feat: plan distributed dynamic filters #634
  3. feat: forward remote dynamic filter updates to coordinator #635
  4. coordinator: merge partial dynamic filters  #636 <- you are here
  5. coordinator: forward merged dynamic filters to consumers #637
  6. worker: apply merged dynamic filters during execution #639

Details

This PR adds machinery around merging dynamic filters in the DynamicFilterRegistry.

    register_task(stage1, task1)                                                                                              
    register_task(stage1, task2)       ───────────┐                                                                              
    register_task(stage1, task3)                  │          ┌───────────────────────┐      merge() when                         
         seal_stage(stage1)                       ├─────────▶│ DynamicFilterRegistry │───▶  - stage is sealed; and               
                                                  │          └───────────────────────┘      - there's enough partial filter      
                                                  │                                         updates                              
record_dynamic_filter_update(stage1, task1)   ────┘                                                                              
record_dynamic_filter_update(stage2, task1)                                                                                           
record_dynamic_filter_update(stage3, task1)

The query coordinator calls register_task for each task in a stage. Concurrently, any running task from any stage can send a dynamic filter update to the registry via record_dynamic_filter_update. The registry needs to detect when all the updates are present and merge() the partial dynamic filters. To help detect this, the coordinator is responsible for calling seal_stage once all the tasks have commenced so we know that no tasks will be added in the future.

The PR implements the above. In the next 2 PRs, we will actually forward the merged filters to consumers.

@jayshrivastava
jayshrivastava changed the base branch from main to js/3-plan-dynamic-filters August 13, 2026 13:27
@jayshrivastava jayshrivastava changed the title Js/4 merge dynamic filters merge dynamic filters in Aug 13, 2026
@jayshrivastava jayshrivastava changed the title merge dynamic filters in coordinator: merge partial dynamic filters Aug 13, 2026
This was referenced Aug 13, 2026
Merged
Merged
Merged
Open
Open
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch from 2cc815b to 7f6de9a Compare August 13, 2026 19:27
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch 2 times, most recently from 400db1c to ad3ea34 Compare August 17, 2026 19:05
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch 2 times, most recently from b798e6c to 876ac94 Compare August 18, 2026 18:15
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch from 876ac94 to d83e14d Compare August 18, 2026 18:52
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch from d83e14d to 593f037 Compare August 21, 2026 16:28
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch from 593f037 to 1548203 Compare August 21, 2026 16:48
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch from 1548203 to 21b4e38 Compare August 21, 2026 20:58
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch from 21b4e38 to 4e36655 Compare August 22, 2026 15:00
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch from 4e36655 to 441ba38 Compare August 23, 2026 00:21
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch from 8af79a1 to 40c52ae Compare September 9, 2026 02:03
@jayshrivastava jayshrivastava changed the title coordinator: merge partial dynamic filters [do not review] coordinator: merge partial dynamic filters Sep 9, 2026
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch from 40c52ae to 5528a7a Compare September 18, 2026 20:49
Base automatically changed from js/3-plan-dynamic-filters to main September 21, 2026 20:01
jayshrivastava added a commit that referenced this pull request Sep 21, 2026
@jayshrivastava
feat: plan distributed dynamic filters (#634)
d450623 
## Stack

This stack of PRs implements distributed dynamic filtering #528 
1. #623
2. #634
<- you are here
3. #635
4. #636
5. #637
6. #639

## Goal

The coordinator should know what dynamic filters exist and where to
route updates.

## Details

### 1. Dynamic Filter Registry

```
  QueryCoordinator
  └── DynamicFilterRegistry
      └── filters: Map<expression_id, PlannedDynamicFilter>
```

Each `PlannedDynamicFilter` stores
- the producers and their stage/tasks
- the consumers and their stage/tasks

This 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 using `apply_expressions` (before
`apply_expressions` was added upstream, it was an Arc pointer strong
count check to see if there were multiple references).
```
worker 1
HashJoinExec  (dynamic_filter_predicate)
    NetworkShuffleExec

worker 2
DataSourceExec (dynamic_filter_predicate)
```
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 `NetworkShuffleExec` uses
the filter.
```
worker 1
HashJoinExec  (dynamic_filter_predicate)
    NetworkShuffleExec  (anchor: dynamic_filter_predicate)

worker 2
DataSourceExec (dynamic_filter_predicate)
```

#### 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`.
```
stage3:
    HashJoinExec <- producer of filter1
        NetworkShuffleExec  (anchors: filter1, filter2)

stage2:
    RepartitionExec
        AggregateExec <- producer of filter2
            NetworkShuffleExec   (anchors: filter1, filter2)

stage1:
        DataSourceExec (consumer: filter1, filter2)
```
This means we serialize 8 filters in total.

However, the minimal anchors you need are like this:
```
stage3:
    HashJoinExec <- producer #1
        NetworkShuffleExec  (anchors: filter2)

stage2:
    RepartitionExec
        AggregateExec <- producer #2
            NetworkShuffleExec   (anchors: filter2)

stage1:
        DataSourceExec (consumer: filter1, filter2)
```
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.
jayshrivastava added a commit that referenced this pull request Sep 21, 2026
@jayshrivastava
feat: forward remote dynamic filter updates to coordinator (#635)
e2f04db 
## Stack

This stack of PRs implements distributed dynamic filtering #528 
1. #623
2. #634
3. #635
<- you are here
4. #636
5. #637
6. #639


## 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.
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch from 75c7432 to 172cecb Compare September 21, 2026 20:15
@jayshrivastava jayshrivastava changed the title [do not review] coordinator: merge partial dynamic filters coordinator: merge partial dynamic filters Sep 21, 2026
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch 4 times, most recently from 3f6f3d6 to 1004f74 Compare September 23, 2026 01:22
gabotechs
gabotechs reviewed Sep 23, 2026
View reviewed changes

@gabotechs gabotechs left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looking good! left some small comments, nothing structural.

Comment thread src/coordinator/dynamic_filter_registry.rs Outdated
Comment thread src/coordinator/dynamic_filter_registry.rs Outdated
Comment thread src/coordinator/dynamic_filter_registry.rs Outdated
Comment thread src/coordinator/dynamic_filter_registry.rs
/// Latest accepted snapshot from each producer task.
pub(super) producer_filters: HashMap<TaskKey, PhysicalDynamicFilterNode>,
/// Full dynamic filter containing the merged predicate and its completion state.
pub(super) merged: Option<PhysicalDynamicFilterNode>,

@gabotechs gabotechs Sep 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see this is working in terms of protobuf messages rather than raw DataFusion expressions.

This means that proto conversion would happen even in a fully in-memory setup. Is it easy to avoid and just work with normal DataFusion types here rather than protobuf messages?

@jayshrivastava jayshrivastava Sep 23, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The main reason is honestly that is_complete isn't public on DynamicFilterPhysicalExpr. So there's actually no way to know if it's complete or not here.

We could detect it on the worker here and send a is_complete bit in the `ProducedDynamicFilter. Do you think making that change is worth it?

@gabotechs gabotechs Sep 25, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤔 I don't think there should be any reason for is_complete() to be private upstream, even the mark_complete() method is public.

Just put a PR for that upstream (apache/datafusion#25738). In the meantime, it's fine to keep what you have here 👍

@jayshrivastava jayshrivastava Sep 25, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice!

Comment thread src/coordinator/dynamic_filter_registry.rs
gabotechs
gabotechs reviewed Sep 23, 2026
View reviewed changes
Comment thread src/coordinator/dynamic_filter_registry.rs Outdated
@jayshrivastava
feat: merge distributed dynamic filter updates
8cab600 
Merge the latest TopK and MIN/MAX producer bounds incrementally. Keep complete-producer coverage for partitioned joins and the complete-replica fast path for CollectLeft joins. Preserve the full dynamic filter and distinguish update readiness from global completion.
@jayshrivastava
pr comments
fadceb8
@jayshrivastava
pr comments
5c0f781
@jayshrivastava
deflake build_side_fetch_is_preserved_by_broadcast by adding an order…
52342b3 
… before the limit
@jayshrivastava
jayshrivastava force-pushed the js/4-merge-dynamic-filters branch from cd6fe19 to 52342b3 Compare September 25, 2026 14:35
@jayshrivastava
jayshrivastava merged commit 7e4c463 into main Sep 25, 2026
35 checks passed
@jayshrivastava
jayshrivastava deleted the js/4-merge-dynamic-filters branch September 25, 2026 14:52
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@gabotechs gabotechs gabotechs approved these changes

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

2 participants

@jayshrivastava @gabotechs