Skip to content

feat: plan distributed dynamic filters - #634

Open
jayshrivastava wants to merge 6 commits into
mainfrom
js/2-forward-dynamic-filter-updates-to-coordinator
Open

feat: plan distributed dynamic filters#634
jayshrivastava wants to merge 6 commits into
mainfrom
js/2-forward-dynamic-filter-updates-to-coordinator

Conversation

@jayshrivastava

@jayshrivastava jayshrivastava commented Aug 13, 2026

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 <- you are here
  3. feat: forward remote dynamic filter updates to coordinator #635
  4. [do not review] coordinator: merge partial dynamic filters  #636
  5. [do not review] coordinator: forward merged dynamic filters to consumers #637
  6. [do not review] worker: apply merged dynamic filters during execution #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 jayshrivastava changed the title feat: forward completed dynamic filters to the coordinator forward dynamic filters from worker -> coordinator Aug 13, 2026
@jayshrivastava jayshrivastava changed the title forward dynamic filters from worker -> coordinator forward dynamic filters from workers -> coordinator Aug 13, 2026
@jayshrivastava jayshrivastava changed the title forward dynamic filters from workers -> coordinator worker: forward dynamic filters to coordinator Aug 13, 2026
@jayshrivastava jayshrivastava changed the title worker: forward dynamic filters to coordinator worker: forward partial dynamic filters to coordinator Aug 13, 2026
@jayshrivastava
jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from a3e5a8d to 6cb1c46 Compare August 13, 2026 19:27
Comment thread src/worker/impl_coordinator_channel.rs Outdated
@jayshrivastava
jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch 3 times, most recently from 8947cd1 to 1d75f65 Compare August 17, 2026 19:38
@jayshrivastava jayshrivastava changed the title worker: forward partial dynamic filters to coordinator feat: plan distributed dynamic filters Aug 18, 2026
@jayshrivastava
jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from 1d75f65 to c54c017 Compare August 18, 2026 18:15
@jayshrivastava
jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from c54c017 to 90cba3a Compare August 18, 2026 18:52
@jayshrivastava
jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from 90cba3a to ec0b443 Compare August 21, 2026 16:28
@jayshrivastava
jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch 4 times, most recently from acfb3b1 to 918839a Compare August 23, 2026 00:21
@jayshrivastava
jayshrivastava marked this pull request as ready for review August 23, 2026 19:28
nuno-faria pushed a commit to fornwall/datafusion that referenced this pull request Aug 24, 2026
…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>
@jayshrivastava
jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from 02092fc to f02eb8b Compare September 3, 2026 22:03
Base automatically changed from js/1-display-dynamic-filters to main September 8, 2026 12:56
jayshrivastava added a commit that referenced this pull request Sep 8, 2026
## 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/2-forward-dynamic-filter-updates-to-coordinator branch 2 times, most recently from 015deb6 to 30804a5 Compare September 8, 2026 18:12

#[derive(Default)]
pub(super) struct DynamicFilterRegistryState {
pub(super) filters: HashMap<u64, PlannedDynamicFilter>,

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.

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(

@jayshrivastava jayshrivastava Sep 9, 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.

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.

@jayshrivastava

Copy link
Copy Markdown
Collaborator Author

benchmarks run tpch/sf100

@gabot-0

gabot-0 commented Sep 9, 2026

Copy link
Copy Markdown

Requested by this comment.

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

@gabotechs

Copy link
Copy Markdown
Collaborator

benchmarks run tpch/sf100

@gabot-0

gabot-0 commented Sep 9, 2026

Copy link
Copy Markdown

Requested by this comment.

Benchmark results

Compared: PR base e5bca18b8016PR head 820f92a02296 · View exact source diff

=== 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 details

Job 52 captured both immutable revisions when the request was queued. The bot fetched and checked out each full commit SHA in detached HEAD, then built and deployed the datafusion-distributed-benchmarks --bin worker target from that checkout.

Identity PR base PR head
Source commit e5bca18b801652f4c79928f1536aed2e1d48e21f 820f92a02296ef4e378b92a6c4ff74c2cfdd3f1a
Phase Base PR head
Build and deployment 1m 21s 1m 20s
All benchmarks 7m 36s 8m 0s
Benchmark tpch/sf100 7m 36s 8m 0s

Workload: tpch/sf100 · all queries · 1 warmup + 5 measured iterations per query

Capacity: 12 c5n.2xlarge nodes for both revisions

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<()> {

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.

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.

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.

Given that PlannedDynamicFilter is mostly just a placeholder implementation for now, I think we should be good with just removing this test.

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.

Deleted 👍🏽

Comment thread src/dynamic_filtering/discovery.rs Outdated

#[test]
fn identifies_a_producer_without_a_local_consumer() -> Result<()> {
fn distinguishes_consumers_from_network_boundary_anchors() -> Result<()> {

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.

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.

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.

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>>,

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 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.

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 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.

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.

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.

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.

One side effect is that the insertion of anchors happens in prepare static/dynamic plan now instead of inject_network_boundaries

Comment on lines +165 to +167
self.dynamic_filter_registry
.register_task(&specialized, task_key)?;

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.

👍 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.
@jayshrivastava
jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from 820f92a to 9866970 Compare September 11, 2026 16:56
Comment thread src/test_utils/routing.rs
}
}

/// Colocates all tasks on the same worker by choosing a URL once and caching it.

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.

This is new, as a part of the rebase. Previously, this was in tests/dynamic_filtering/common.rs

@jayshrivastava
jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch 2 times, most recently from 55e34b7 to 7b28bdb Compare September 11, 2026 18:52
@jayshrivastava

Copy link
Copy Markdown
Collaborator Author

@gabotechs This is ready for another review.

@jayshrivastava
jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from 847ccf7 to f3c06e2 Compare September 11, 2026 20:40
@jayshrivastava
jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from f3c06e2 to 2b1213f Compare September 11, 2026 21:31
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants