feat: plan distributed dynamic filters by jayshrivastava · Pull Request #634 · datafusion-contrib/datafusion-distributed

jayshrivastava · 2026-08-13T13:25:52Z

Stack

This stack of PRs implements distributed dynamic filtering #528

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

…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 · 2026-09-08T18:24:44Z

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


In future PRs, we add more mutex protected state in this struct.

More? I already see one Mutex in DynamicFilterRegistry.state, do you think we'll need more than that?

jayshrivastava · 2026-09-09T01:51:18Z

+// 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(


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 · 2026-09-09T03:09:41Z

benchmarks run tpch/sf100

gabot-0 · 2026-09-09T03:09:45Z

Requested by this comment.

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

gabotechs · 2026-09-09T06:50:03Z

benchmarks run tpch/sf100

gabot-0 · 2026-09-09T06:50:25Z

Requested by this comment.

Benchmark results

Compared: PR base e5bca18b8016 → PR 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

gabotechs · 2026-09-09T07:05:39Z

+
+    // Test that we correctly register producers and consumers while ignoring anchors.
+    #[test]
+    fn registers_dynamic_filters_by_expression_id() -> Result<()> {


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.

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

Deleted 👍🏽

gabotechs · 2026-09-09T07:10:28Z


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


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.

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.

gabotechs · 2026-09-09T07:14:02Z

    pub(crate) properties: Arc<PlanProperties>,
    pub(crate) input_stage: Stage,
    pub(crate) worker_connections: WorkerConnectionPool,
+    pub(crate) dynamic_filter_anchors: Vec<Arc<dyn PhysicalExpr>>,


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.

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.

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.

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

gabotechs · 2026-09-09T07:17:12Z

+        self.dynamic_filter_registry
+            .register_task(&specialized, task_key)?;
+


👍 very naturally handled

jayshrivastava · 2026-09-11T18:59:17Z

@gabotechs This is ready for another review.

gabotechs · 2026-09-21T07:42:38Z

+/// An anchor is an artificial dynamic filter consumer injected into network boundaries
+/// to keep consumer references alive when they are moved across network boundaries.
+///
+/// TODO(#697): remove anchors in df-56.
+#[derive(Clone)]
+pub struct DiscoveredDynamicFilterAnchor {
+    pub id: u64,
+    pub expression: Arc<dyn PhysicalExpr>,
+}
+
+pub struct DiscoveredDynamicFilterConsumers {
+    // Real consumers, ordered by expression id.
+    pub consumers: Vec<DiscoveredDynamicFilter>,
+    // Artificial consumers. Dynamic filters in network boundaries. Also ordered by expression id.
+    pub anchors: Vec<DiscoveredDynamicFilterAnchor>,
+}
+
+/// Finds dynamic-filter consumers and network-boundary anchors in `plan`, deduplicated by
+/// expression ID within each category.
+pub fn discover_dynamic_filter_consumers(
    plan: &Arc<dyn ExecutionPlan>,


I see the public exposure here is getting a bit out of hand. Do we really need all these to be part of the public API?

My intuition tells me that none of these should be part of the public API

These are actually gated by #[cfg(any(feature = "integration", test))]

The reason is because I moved to integration style tests in /home/bits/datafusion-distributed/tests/, so these helpers need to be pub.

However, I think we can probably move those integration-style tests to this file to avoid the import hacks.

gabotechs · 2026-09-21T07:47:08Z

+/// Isolates all shared, in-memory dynamic filter state from this plan if it contains
+/// any dynamic filter producers or consumers.
+///
+/// Plan nodes *within* this plan will share in-memory dynamic filter state. However, they
+/// will not share state with plan nodes outside of this plan, such as in parent stages.
+///
+/// # Correctness: Task-Local Dynamic Filters are Safe to Apply
+///
+/// Claim: It is correct for a producer to *always* update consumers within the same task.
+///
+/// Proof:
+///
+/// Invariant: If the task-local consumer filters out a row, removing that row must not change the output
+/// of the task-local producer.


This doc comment is a bit too ong and LLM-y, If we can shorten it out a bit that'd be great.

I actually wrote it myself 😭

I'll shorten it out though. I just wanted to have this written down somewhere

I added diagrams, which is more lines but is better than a wall of text.

gabotechs · 2026-09-21T07:49:17Z

+        .is_empty();
+
+    if has_producers || has_consumers {
        roundtrip_pb(plan, task_ctx)


This roundtrip_pb function is serializing to bytes and decoding those bytes back. The byte serialization is unnecessary, it's enough to just convert to prost-generated struct and back, without paying for serialization here.

Done. Good observation

gabotechs · 2026-09-21T08:19:51Z

+            let consumers = discover_dynamic_filter_consumers(plan)?;
+            let discovered = DynamicFilterIds {
+                consumers: consumers
+                    .consumers
+                    .into_iter()
+                    .map(|consumer| consumer.id)
+                    .collect(),
+                anchors: consumers
+                    .anchors
+                    .into_iter()
+                    .map(|anchor| anchor.id)
+                    .collect(),
+                producers: discover_dynamic_filter_producers(plan)?
+                    .into_iter()
+                    .map(|producer| producer.id)
+                    .collect(),
+            };
+            render(plan.as_ref(), 1, &discovered, &mut normalizer, &mut output)?;


I don't think we want to expose this API to the public:

discover_dynamic_filter_consumers

DiscoveredDynamicFilterConsumers

DiscoveredDynamicFilterAnchor

DiscoveredDynamicFilter

DiscoveredDynamicFilterProducer

These are internal struct that we invented inside datafusion-distributed, and users should not be able to interact with them.

I get the feeling that the previous unit tests where moved here as "integration" tests, but these tests are still coupled to internal implementation details of the dyn filter discovery machinery.

Some options here:

Move this tests to src/dynamic_fitlering/discovery.rs, and remove all the public exposure of the internal structs

Create a test_utils/dynamic_filters.rs helper that exposes a properly encapsulated display_dynamic_filter_discovery() function, keeping private all the other internal structs.

Keep the tests here as real integration tests, using exclusively the public API

jayshrivastava · 2026-09-21T17:04:24Z

Ready for another review! I'll rebase before merging. I don't want to rebase and overwrite the history so that my diffs are easy to review.

gabotechs

👍 Looks good!

gabotechs · 2026-09-21T17:40:48Z

-    let encoded = encode_execution_plan(plan, task_ctx)?;
-    decode_execution_plan(&encoded, task_ctx)
+    let encode_codec = DistributedCodec::new_combined_with_user(task_ctx.session_config());
+    let encode_converter = new_proto_converter();
+    let proto = encode_converter.execution_plan_to_proto(&plan, &encode_codec)?;
+    let decode_codec = DistributedCodec::new_combined_with_user(task_ctx.session_config());
+    let decode_ctx = PhysicalPlanDecodeContext::new(task_ctx, &decode_codec);
+    let decode_converter = new_proto_converter();
+    decode_converter.proto_to_execution_plan(&proto, &decode_ctx)


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.

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

## Stack This stack of PRs implements distributed dynamic filtering #528 1. #623 2. #634 3. #635 4. #636 <- you are here 5. #637 6. #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.

## What Bump the datafusion-distributed ref so we stay current. It picks up distributed dynamic filters (datafusion-contrib/datafusion-distributed#634, datafusion-contrib/datafusion-distributed#635), reduced producer fan-out in NetworkShuffleExec (datafusion-contrib/datafusion-distributed#717), the AQE worker-assignment retry (datafusion-contrib/datafusion-distributed#723), fractional desired task counts (datafusion-contrib/datafusion-distributed#714), and the datafusion-iceberg refactor (datafusion-contrib/datafusion-distributed#742). The introduction of distributed dynamic filters introduces a large hash join regression when used over our transport layer, so this PR disables their use until we get out a fix. ## Tests CI's green. Once benchmarks run we should be able to verify this _does not_ introduce the aforementioned perf regression. --------- Co-authored-by: paradedb-github-bot[bot] <282009505+paradedb-github-bot[bot]@users.noreply.github.com>

jayshrivastava changed the title ~~feat: forward completed dynamic filters to the coordinator~~ forward dynamic filters from worker -> coordinator Aug 13, 2026

jayshrivastava changed the title ~~forward dynamic filters from worker -> coordinator~~ forward dynamic filters from workers -> coordinator Aug 13, 2026

jayshrivastava changed the title ~~forward dynamic filters from workers -> coordinator~~ worker: forward dynamic filters to coordinator Aug 13, 2026

jayshrivastava changed the title ~~worker: forward dynamic filters to coordinator~~ worker: forward partial dynamic filters to coordinator Aug 13, 2026

jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from a3e5a8d to 6cb1c46 Compare August 13, 2026 19:27

jayshrivastava commented Aug 17, 2026

View reviewed changes

Comment thread src/worker/impl_coordinator_channel.rs Outdated

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 changed the title ~~worker: forward partial dynamic filters to coordinator~~ feat: plan distributed dynamic filters Aug 18, 2026

jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from 1d75f65 to c54c017 Compare August 18, 2026 18:15

jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from c54c017 to 90cba3a Compare August 18, 2026 18:52

jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from 90cba3a to ec0b443 Compare August 21, 2026 16:28

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 mentioned this pull request Aug 23, 2026

Make filter pushdown API more precise for their purpose apache/datafusion#18856

Closed

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

adriangb mentioned this pull request Aug 24, 2026

fix: decide hash join dynamic filter production at planning time apache/datafusion#24601

Merged

jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from 015deb6 to 30804a5 Compare September 8, 2026 18:12

jayshrivastava commented Sep 8, 2026

View reviewed changes

jayshrivastava commented Sep 9, 2026

View reviewed changes

gabotechs reviewed Sep 9, 2026

View reviewed changes

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

jayshrivastava requested a review from gabotechs September 11, 2026 18:58

jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch 2 times, most recently from f3c06e2 to 2b1213f Compare September 11, 2026 21:31

datafusion-contrib deleted a comment from jayshrivastava Sep 21, 2026

gabotechs reviewed Sep 21, 2026

View reviewed changes

jayshrivastava requested a review from gabotechs September 21, 2026 17:03

gabotechs approved these changes Sep 21, 2026

View reviewed changes

jayshrivastava force-pushed the js/2-forward-dynamic-filter-updates-to-coordinator branch from 32a5dc2 to 8dba6b2 Compare September 21, 2026 19:21

wrap comments in code blocks

3582c10

jayshrivastava merged commit d450623 into main Sep 21, 2026
34 checks passed

jayshrivastava deleted the js/2-forward-dynamic-filter-updates-to-coordinator branch September 21, 2026 20:01

barbarj mentioned this pull request Sep 24, 2026

chore: Bump df-d to snapshot-main-2026-09-24-211026 paradedb/paradedb#6475

Closed

barbarj mentioned this pull request Sep 28, 2026

chore: bump df-d snapshot and disable remote dynamic filters paradedb/paradedb#6534

Merged

		self.dynamic_filter_registry
		.register_task(&specialized, task_key)?;

Conversation

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

Uh oh!

Stack

Goal

Details

1. Dynamic Filter Registry

Implementation

2. Network Anchors

Network Anchors Implementation

Other Notes

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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

Benchmark results

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!

Choose a reason for hiding this comment

Uh oh!

jayshrivastava commented Sep 11, 2026

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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jayshrivastava commented Sep 21, 2026

Uh oh!

gabotechs left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jayshrivastava commented Aug 13, 2026 •

edited

Loading

jayshrivastava Sep 9, 2026 •

edited

Loading

gabot-0 commented Sep 9, 2026 •

edited

Loading

gabot-0 commented Sep 9, 2026 •

edited

Loading