proto: serialize and dedupe dynamic filters v2 by jayshrivastava · Pull Request #21807 · apache/datafusion

jayshrivastava · 2026-04-23T17:58:13Z

Which issue does this PR close?

Informs: datafusion-contrib/datafusion-distributed#180
Closes: #20418

Rationale for this change

Consider you have a plan with a HashJoinExec and DataSourceExec

HashJoinExec(dynamic_filter_1 on a@0)
  (...left side of join)
  ProjectionExec(a := Column("a", source_index))
    DataSourceExec
      ParquetSource(predicate = dynamic_filter_2)

You serialize the plan, deserialize it, and execute it. What should happen is that the dynamic filter should "work", meaning:

When you deserialize the plan, both the HashJoinExec and DataSourceExec should have pointers to the same DynamicFilterPhysicalExpr
The DynamicFilterPhysicalExpr should be updated during execution by the HashJoinExec and the DataSourceExec should filter out rows

This does not happen today for a few reasons, a couple of which this PR aims to address

DynamicFilterPhysicalExpr is not survive round-tripping. The internal exprs get inlined (ex. it may be serialized as Literal) due to the PhysicalExpr::snapshot() API
Even if DynamicFilterPhysicalExpr survives round-tripping, the one pushed down to the DataSourceExec often has different children. In this case, you have two DynamicFilterPhysicalExpr which
do not survive deduping, causing referential integrity to be lost.

What changes are included in this PR?

This PR aims to fix those problems by:

Removing the snapshot() call from the serialization process
Adding protos for DynamicFilterPhysicalExpr so it can be serialized and deserialized
Removing Arc-based deduplication. We now only dedupe on
expression_id if the PhysicalExpr reports a expression_id.
After this change, only DynamicFilterPhysicalExpr reports an expression_id
to be deduped.
expression_id is now just a random u64. Since a given query likely
only has a few DynamicFilterPhysicalExpr instances, the odds of a
collision are very low
There's no need for a DedupingSerializer anymore since the
expression_id is already stored in the dynamic filter proto itself

Future work:

Serialize dynamic filters in HashJoinExec, AggregateExec and SortExec
Add tests which actually execute plans after deserialization and assert that dynamic filtering is functional
Add proto converters to the PhysicalExtensionCodec trait so implementors can utilize deduping logic

Are these changes tested?

adds tests which roundtrip dynamic filters and assert that referential
integrity is maintained
removes tests that test Arc-based deduplication and session id
rotation since we don't support that anymore

Are there any user-facing changes?

The default codec does not call snapshot() on PhysicalExpr during serialization anymore. This means that DynamicFilterPhysicalExpr are now serialized and deserialized without snapshotting.
All PhysicalExpr are not deduped anymore. Only DynamicFilterPhysicalExpr is

adriangb

This is looking very nice @jayshrivastava 👍🏻

Lets leave this open for a day to see if anyone has thoughts on the public APIs being tweaked.

cc @stuhood iirc you mentioned you also plan to ser/de dynamic filters across process (but not node) boundaries

jayshrivastava · 2026-04-24T19:43:33Z

@adriangb I've addressed the comments in the last commit. Let me know what you think!

jayshrivastava · 2026-04-24T19:44:48Z

Oh TYFR! That was fast

Lets leave this open for a day to see if anyone has thoughts on the public APIs being tweaked.

No problem. I can merge some time next week 👍🏽

adriangb · 2026-04-24T19:47:51Z

FYI I hit rebase

stuhood

Thanks!

stuhood · 2026-04-24T20:48:51Z

+    pub is_complete: bool,
+}
+
+// TODO: Include expression_id in debug output.


Ideally even in the DisplayAs representation for plans, I would think. It's actually possibly the most important thing to have in the plan when rendering a dynamic filter?

I wanted to keep this PR small, so I hid the expression_id to avoid 1 test that was failing. I have code ready to address this TODO which I plan to publish as a next step after this PR is merged. This PR is just an incremental step.

// See https://github.com/apache/datafusion/issues/20418. Currently, plan nodes // like `HashJoinExec`, `AggregateExec`, `SortExec` do not serialize their // dynamic filter. This causes round trips to fail on the `expression_id` // because it is regenerated on deserialization.

stuhood · 2026-04-24T20:50:29Z

+    /// Rebuild a `DynamicFilterPhysicalExpr` from its stored parts. Used by
+    /// proto deserialization to preserve `expression_id` across a roundtrip
+    /// rather than minting a fresh one.
+    ///
+    /// **Warning:** intended only for `datafusion-proto` (de)serialization.
+    /// Not a stable API.


Could this be generic via the shared state proposal? Not sure what is up with that. Ditto fn inner.

The shared state discussion seems promising! I'm hoping it will be an easy migration. I imagine that we would end up storing all Inners in the TaskContext instead of in the actual DynamicFilterPhysicalExpr and looking up the Inners via the expression_id. It's still an ongoing discussion though so I can't be sure how the migration will look.

Yep that would make sense to me

stuhood · 2026-04-24T20:52:36Z

 itertools = { workspace = true, features = ["use_std"] }
 parking_lot = { workspace = true }
 petgraph = "0.8.3"
+rand = { workspace = true }


Not sure what the conventions are, but: introducing random order to the physical-expr crate seems like a big deal? Is this being used to generate something that could be deterministic from some sort of context instead?

This is a good question! I was thinking about a few alternatives 😄

We could hash all the exprs in the DynamicFilterPhysicalExpr to get an id. However, this wouldn't solve the shared Inner state linking problem. We don't want this type of identifier - we want an identifier for the inner state specifically.

We could use the Arc address of the Inner struct, but it's a bit of a smell to rely on pointer addresses - for example, IDs derived from Arc pointers are only valid until the Arc is dropped. This is what the old code used and something I used as well in initial versions of this PR

A rand u64 is not bad. Realistically, we just need to not have a collision between distinct dynamic filters in a plan. I figure that the probability of more than 2 or 3 distinct dynamic filters in a query must be very low already. And the probability of 2 or 3 rand u64s colliding is negligible.

Lmk what you think!

I'm less worried about collisions, and more worried about non-determinism (causing flaky tests, different plans to be generated randomly, etc). The other annoying aspect of random ids is that they are huge. If you go ahead and actually render this everywhere, a random ID is going to take up a lot more space than one generated on a context (...starting from 0, etc).

Right now the ids are also random (generated from a mashup of arc pointer address, process id, etc.). So in that sense it's no better or worse. But I do agree something deterministically generated from context would be better.

The only alternative that occurs to me is a process level atomic. Not sure if that might cause some locking, etc. Do you have any other suggestions?

Lastly: we can always change this. As long as there is no API contract on what this number is going to be we could replace it at any point if it becomes a problem.

Informs: datafusion-contrib/datafusion-distributed#180 Closes: apache#20418 Consider you have a plan with a `HashJoinExec` and `DataSourceExec` ``` HashJoinExec(dynamic_filter_1 on a@0) (...left side of join) ProjectionExec(a := Column("a", source_index)) DataSourceExec ParquetSource(predicate = dynamic_filter_2) ``` You serialize the plan, deserialize it, and execute it. What should happen is that the dynamic filter should "work", meaning: 1. When you deserialize the plan, both the `HashJoinExec` and `DataSourceExec` should have pointers to the same `DynamicFilterPhysicalExpr` 2. The `DynamicFilterPhysicalExpr` should be updated during execution by the `HashJoinExec` and the `DataSourceExec` should filter out rows This does not happen today for a few reasons, a couple of which this PR aims to address 1. `DynamicFilterPhysicalExpr` is not survive round-tripping. The internal exprs get inlined (ex. it may be serialized as `Literal`) due to the `PhysicalExpr::snapshot()` API 2. Even if `DynamicFilterPhysicalExpr` survives round-tripping, the one pushed down to the `DataSourceExec` often has different children. In this case, you have two `DynamicFilterPhysicalExpr` which do not survive deduping, causing referential integrity to be lost. This PR aims to fix those problems by: 1. Removing the `snapshot()` call from the serialization process 2. Adding protos for `DynamicFilterPhysicalExpr` so it can be serialized and deserialized 3. Removing `Arc`-based deduplication. We now only dedupe on `expression_id` if the `PhysicalExpr` reports a `expression_id`. After this change, only `DynamicFilterPhysicalExpr` reports an `expression_id` to be deduped. 4. `expression_id` is now just a random u64. Since a given query likely only has a few `DynamicFilterPhysicalExpr` instances, the odds of a collision are very low 5. There's no need for a `DedupingSerializer` anymore since the `expression_id` is already stored in the dynamic filter proto itself Testing - adds tests which roundtrip dynamic filters and assert that referential integrity is maintained - removes tests that test `Arc`-based deduplication and session id rotation

adriangb · 2026-04-30T04:36:43Z

@stuhood I think based on the convo your review is not blocking but just wanted to check before we go ahead with this? @jayshrivastava do you think you could ping @LiaCastaneda to take a critical eye at this? We're on iteration 2 of this whole system so I'm really hoping we can get it right this time 🙏🏻

LiaCastaneda · 2026-04-30T07:31:10Z

👋 Sorry, I’ve been a bit detached from dynamic filtering. I’ll make some time to take a look at this today.

LiaCastaneda

The approach makes sense to me, It's nice that we don't need two expression_ids just for dynamic filters, and DynamicFilterPhysicalExpr is handled like any other expression during serialization. I have a few API style questions, but I think the overall approach is much cleaner now. Thanks for the hard work Jay and Adrian!

LiaCastaneda · 2026-04-30T09:46:59Z

+/// **Warning:** exposed publicly solely so that proto (de)serialization in
+/// `datafusion-proto` can read and rebuild this state. Do not treat this type
+/// or its layout as a stable API.
+#[derive(Clone)]
+pub struct Inner {


Should we hide this somehow instead of using a warning comment? Exposing this leaves the door open to callers messing with the inner dynamic filter's Arc count -- which isn't great either. It's strange to me that Rust doesn't provide a way to do this currently.

Alas, yes, I hate exposing this. But as far as I can tell there is no way around it: #21835.

There's a huge, ugly PR up at #21949.

Would appreciate any input on the issue or PR if you have time.

LiaCastaneda · 2026-04-30T09:59:22Z

 impl Inner {
    fn new(expr: Arc<dyn PhysicalExpr>) -> Self {
        Self {
+            expression_id: random::<u64>(),


Would it be better to use something more standard like uuid or an atomic counter across all expressions?

LiaCastaneda · 2026-04-30T10:49:59Z

+  // When serializing, this is set via `PhysicalExpr::expression_id`. When deserializing,
+  // this id is used by the `DeduplicatingProtoConverter` to preserve referential
+  // integrity across serde roundtrips for different expressions with the same id.
  optional uint64 expr_id = 30;


I didn't remember we had this already 🤔 we had an id in the proto but not in the trait, I guess this suggests the original intent was always to have expressions carry their own identity in the trait as well

LiaCastaneda · 2026-04-30T11:00:33Z

would it make sense to remove snapshot() api now that it's not used? (maybe as a follow up) i think the only expression that implements it is the dynamic filter.

It's still used in Parquet pruning. But I agree it never served as many purposes as we thought it would, it's probably best to remove it and add a downcast for DynamicFilterPhysicalExpr or something.

I can do this as a quick follow up. Will post a PR momentarily.

I opened this PR #21975. The base is main, so only the last commit is relevant.

adriangb · 2026-04-30T13:05:44Z

@jayshrivastava since both Lía and Stu suggested the same thing (use a deterministic incrementing counter) wdyt of adding something like:

use std::sync::atomic::{AtomicUsize, Ordering};

struct ExpressionIdCounter {
    inner: AtomicUsize,
}

impl ExpressionIdCounter {
    pub const fn new(initial_value: usize) -> Self {
        Self {
            inner: AtomicUsize::new(initial_value),
        }
    }

  pub fn get_next(&self) -> usize {
          self.inner.fetch_add(1, Ordering::Relaxed)
      }
}

static EXPR_ID: ExpressionIdCounter = ExpressionIdCounter::new(0);

I'd keep it private for now, but if we want to re-use it we just make it pub and move it.

I don't see any other blockers to merging this!

stuhood · 2026-04-30T17:27:03Z

@stuhood I think based on the convo your review is not blocking but just wanted to check before we go ahead with this?

Definitely not blocking! Still building context.

…jayshrivastava/datafusion into js/dedupe-dynamic-filter-inner-state-v2

jayshrivastava · 2026-05-01T14:20:20Z

@adriangb I added deterministic expression ids 👍🏽

github-actions · 2026-05-01T14:23:56Z

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning origin/main
    Building datafusion-physical-expr v53.1.0 (current)
       Built [  31.711s] (current)
     Parsing datafusion-physical-expr v53.1.0 (current)
      Parsed [   0.047s] (current)
    Building datafusion-physical-expr v53.1.0 (baseline)
       Built [  25.680s] (baseline)
     Parsing datafusion-physical-expr v53.1.0 (baseline)
      Parsed [   0.046s] (baseline)
    Checking datafusion-physical-expr v53.1.0 -> v53.1.0 (no change; assume patch)
     Checked [   0.411s] 222 checks: 222 pass, 30 skip
     Summary no semver update required
    Finished [  59.932s] datafusion-physical-expr
    Building datafusion-physical-expr-common v53.1.0 (current)
       Built [  20.531s] (current)
     Parsing datafusion-physical-expr-common v53.1.0 (current)
      Parsed [   0.022s] (current)
    Building datafusion-physical-expr-common v53.1.0 (baseline)
       Built [  20.170s] (baseline)
     Parsing datafusion-physical-expr-common v53.1.0 (baseline)
      Parsed [   0.022s] (baseline)
    Checking datafusion-physical-expr-common v53.1.0 -> v53.1.0 (no change; assume patch)
     Checked [   0.264s] 222 checks: 222 pass, 30 skip
     Summary no semver update required
    Finished [  42.538s] datafusion-physical-expr-common
    Building datafusion-proto v53.1.0 (current)
       Built [  55.046s] (current)
     Parsing datafusion-proto v53.1.0 (current)
      Parsed [   0.144s] (current)
    Building datafusion-proto v53.1.0 (baseline)
       Built [  54.815s] (baseline)
     Parsing datafusion-proto v53.1.0 (baseline)
      Parsed [   0.145s] (baseline)
    Checking datafusion-proto v53.1.0 -> v53.1.0 (no change; assume patch)
     Checked [   2.199s] 222 checks: 221 pass, 1 fail, 0 warn, 30 skip

--- failure enum_variant_added: enum variant added on exhaustive enum ---

Description:
A publicly-visible enum without #[non_exhaustive] has a new variant.
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#enum-variant-new
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.47.0/src/lints/enum_variant_added.ron

Failed in:
  variant ExprType:DynamicFilter in /home/runner/work/datafusion/datafusion/datafusion/proto/src/generated/prost.rs:1393
  variant ExprType:DynamicFilter in /home/runner/work/datafusion/datafusion/datafusion/proto/src/generated/prost.rs:1393

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [ 115.358s] datafusion-proto

See apache#21807 `PhysicalExpr::snapshot` isn't required because it only applies to dynamic filters. The only place that its used is in `pruning_predicate.rs`. Since there's only one use case, we can just downcast to `DynamicFilterPhysicalExpr` and call `current()`. Remove `PhysicalExpr::snapshot` and associated code. Yes, the existing tests should cover the exact same functionality. Filter pushdown still workers with dynamic filters; we use `current()` instead of `snapshot()` to capture the expression. Yes. `PhysicalExpr::snapshot()` no longer exists. Users should downcast to `DynamicFilterPhysicalExpr` and call `current()` intead of calling `snapshot()`. Similarly, in `datafusion/ffi`, `FFI_PhysicalExpr` no longer has a `snapshot()` method.

adriangb · 2026-05-01T17:21:49Z

Thanks @jayshrivastava !

See apache#21807 `PhysicalExpr::snapshot` isn't required because it only applies to dynamic filters. The only place that its used is in `pruning_predicate.rs`. Since there's only one use case, we can just downcast to `DynamicFilterPhysicalExpr` and call `current()`. Remove `PhysicalExpr::snapshot` and associated code. Yes, the existing tests should cover the exact same functionality. Filter pushdown still workers with dynamic filters; we use `current()` instead of `snapshot()` to capture the expression. Yes. `PhysicalExpr::snapshot()` no longer exists. Users should downcast to `DynamicFilterPhysicalExpr` and call `current()` intead of calling `snapshot()`. Similarly, in `datafusion/ffi`, `FFI_PhysicalExpr` no longer has a `snapshot()` method.

github-actions Bot added physical-expr Changes to the physical-expr crates proto Related to proto crate physical-plan Changes to the physical-plan crate labels Apr 23, 2026

jayshrivastava force-pushed the js/dedupe-dynamic-filter-inner-state-v2 branch from 004aa52 to 23fc7f1 Compare April 23, 2026 23:33

github-actions Bot removed the physical-plan Changes to the physical-plan crate label Apr 23, 2026

jayshrivastava commented Apr 23, 2026

View reviewed changes

Comment thread datafusion/proto/src/physical_plan/mod.rs

jayshrivastava force-pushed the js/dedupe-dynamic-filter-inner-state-v2 branch 2 times, most recently from 64b5889 to c042c08 Compare April 24, 2026 00:14

jayshrivastava changed the title ~~Js/dedupe dynamic filter inner state v2~~ proto: serialize and dedupe dynamic filters Apr 24, 2026

jayshrivastava marked this pull request as ready for review April 24, 2026 00:17

jayshrivastava mentioned this pull request Apr 24, 2026

[DISCUSSION] Future of Dynamic Filters Sync #21207

Open

jayshrivastava changed the title ~~proto: serialize and dedupe dynamic filters~~ proto: serialize and dedupe dynamic filters v2 Apr 24, 2026

adriangb requested changes Apr 24, 2026

View reviewed changes

jayshrivastava commented Apr 24, 2026

View reviewed changes

Comment thread datafusion/physical-expr/src/expressions/dynamic_filters.rs

jayshrivastava commented Apr 24, 2026

View reviewed changes

Comment thread datafusion/physical-expr/src/expressions/dynamic_filters.rs

adriangb approved these changes Apr 24, 2026

View reviewed changes

Comment thread datafusion/physical-expr-common/src/physical_expr.rs

Comment thread datafusion/physical-expr/src/expressions/dynamic_filters.rs

adriangb force-pushed the js/dedupe-dynamic-filter-inner-state-v2 branch from 586648b to 09c2d34 Compare April 24, 2026 19:47

adriangb mentioned this pull request Apr 24, 2026

Split proto serialization to encapsulate private state #21835

Open

stuhood reviewed Apr 24, 2026

View reviewed changes

jayshrivastava added 2 commits April 28, 2026 20:52

move expression_id to Inner

0766648

jayshrivastava force-pushed the js/dedupe-dynamic-filter-inner-state-v2 branch from 1e0ec7b to 0766648 Compare April 28, 2026 21:02

adriangb mentioned this pull request Apr 29, 2026

Split proto serialization to encapsulate private state (#21835) #21929

Draft

Merge branch 'main' into js/dedupe-dynamic-filter-inner-state-v2

8c96885

adriangb mentioned this pull request Apr 30, 2026

Split proto serialization to encapsulate private state (cleanup-first take on #21835) #21949

Draft

LiaCastaneda approved these changes Apr 30, 2026

View reviewed changes

jayshrivastava mentioned this pull request Apr 30, 2026

proto: serialize and dedupe dynamic filters #20416

Closed

jayshrivastava added 2 commits May 1, 2026 14:18

add deterministic expr ids

0b75476

Merge branch 'js/dedupe-dynamic-filter-inner-state-v2' of github.com:…

64253e5

…jayshrivastava/datafusion into js/dedupe-dynamic-filter-inner-state-v2

typo

9c4b437

jayshrivastava mentioned this pull request May 1, 2026

physical-expr-common: remote PhysicalExpr::snapshot method #21975

Open

adriangb approved these changes May 1, 2026

View reviewed changes

adriangb added this pull request to the merge queue May 1, 2026

Merged via the queue into apache:main with commit 948cd09 May 1, 2026
40 checks passed

Conversation

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

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

adriangb left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

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

Uh oh!

Uh oh!

jayshrivastava commented Apr 24, 2026

Uh oh!

adriangb commented Apr 24, 2026

Uh oh!

stuhood left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

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!

adriangb commented Apr 30, 2026

Uh oh!

LiaCastaneda commented Apr 30, 2026

Uh oh!

LiaCastaneda left a comment

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!

adriangb commented Apr 30, 2026

Uh oh!

stuhood commented Apr 30, 2026

jayshrivastava commented Apr 23, 2026 •

edited

Loading

jayshrivastava commented Apr 24, 2026 •

edited

Loading

github-actions Bot commented May 1, 2026 •

edited

Loading