wip

Author: jayshrivastava

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

@@ -18,7 +18,7 @@
 use std::any::Any;
 use std::fmt;
 use std::fmt::{Debug, Display, Formatter};
-use std::hash::{DefaultHasher, Hash, Hasher};
+use std::hash::{Hash, Hasher};
 use std::sync::Arc;
 
 use crate::utils::scatter;
@@ -439,51 +439,14 @@ pub trait PhysicalExpr: Any + Send + Sync + Display + Debug + DynEq + DynHash {
         ExpressionPlacement::KeepInPlace
     }
 
-    /// Returns a composite identifier for a [`PhysicalExpr`]. Note that if the expression
-    /// is dropped, then the returned id is no longer valid.
-    fn expr_id(self: Arc<Self>, salt: &[u64]) -> Option<PhysicalExprId> {
-        Some(PhysicalExprId::new(expr_id_from_arc(&self, salt), None))
+    /// Returns a process-local identifier used during serde round-trips to
+    /// preserve referential integrity for special expressions like dynamic
+    /// filters. Ordinary expressions return `None` by default.
+    fn expression_id(&self) -> Option<u64> {
+        None
     }
 }
 
-/// A composite identifier for [`PhysicalExpr`].
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
-pub struct PhysicalExprId {
-    exact: u64,
-    shallow: Option<u64>,
-}
-
-impl PhysicalExprId {
-    /// Create a new [`PhysicalExprId`]. Both ids must be globally unique within
-    /// a process.
-    pub fn new(exact: u64, shallow: Option<u64>) -> Self {
-        Self { exact, shallow }
-    }
-
-    /// Returns the identifier for the full expression tree, including children.
-    pub fn exact(&self) -> u64 {
-        self.exact
-    }
-
-    /// Returns the identifier for just the expression root, ignoring children.
-    pub fn shallow(&self) -> Option<u64> {
-        self.shallow
-    }
-}
-
-/// Computes a unique identifier for a type contained within an [`Arc`]. It hashes
-/// the [`Arc`] pointer to create a process-local identifier that remains valid
-/// only while that allocation is still alive.
-pub fn expr_id_from_arc<T: ?Sized>(expr: &Arc<T>, salt: &[u64]) -> u64 {
-    let mut hasher = DefaultHasher::new();
-    let ptr = Arc::as_ptr(expr) as *const () as u64;
-    ptr.hash(&mut hasher);
-    for &salt in salt {
-        salt.hash(&mut hasher);
-    }
-    hasher.finish()
-}
-
 #[deprecated(
     since = "50.0.0",
     note = "Use `datafusion_expr_common::dyn_eq` instead"

datafusion/physical-expr/Cargo.toml

@@ -55,6 +55,7 @@ indexmap = { workspace = true }
 itertools = { workspace = true, features = ["use_std"] }
 parking_lot = { workspace = true }
 petgraph = "0.8.3"
+rand = { workspace = true }
 recursive = { workspace = true, optional = true }
 tokio = { workspace = true }
 half = { workspace = true }
@@ -64,7 +65,6 @@ arrow = { workspace = true, features = ["test_utils"] }
 criterion = { workspace = true }
 datafusion-functions = { workspace = true }
 insta = { workspace = true }
-rand = { workspace = true }
 rstest = { workspace = true }
 
 [[bench]]

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

@@ -16,7 +16,14 @@
 // under the License.
 
 use parking_lot::RwLock;
-use std::{fmt::Display, hash::Hash, sync::Arc};
+use std::{
+    fmt::Display,
+    hash::Hash,
+    sync::{
+        Arc, LazyLock,
+        atomic::{AtomicU64, Ordering},
+    },
+};
 use tokio::sync::watch;
 
 use crate::PhysicalExpr;
@@ -26,9 +33,8 @@ use datafusion_common::{
     tree_node::{Transformed, TransformedResult, TreeNode},
 };
 use datafusion_expr::ColumnarValue;
-use datafusion_physical_expr_common::physical_expr::{
-    DynHash, PhysicalExprId, expr_id_from_arc,
-};
+use datafusion_physical_expr_common::physical_expr::DynHash;
+use rand::random;
 
 /// State of a dynamic filter, tracking both updates and completion.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -59,6 +65,7 @@ impl FilterState {
 /// [Dynamic Filters: Passing Information Between Operators During Execution for 25x Faster Queries blog]: https://datafusion.apache.org/blog/2025/09/10/dynamic-filters
 #[derive(Debug)]
 pub struct DynamicFilterPhysicalExpr {
+    expression_id: u64,
     /// The original children of this PhysicalExpr, if any.
     /// This is necessary because the dynamic filter may be initialized with a placeholder (e.g. `lit(true)`)
     /// and later remapped to the actual expressions that are being filtered.
@@ -93,6 +100,7 @@ struct Inner {
 /// An atomic snapshot of a [`DynamicFilterPhysicalExpr`] used to reconstruct the expression during
 /// serialization / deserialization.
 pub struct DynamicFilterSnapshot {
+    expression_id: u64,
     children: Vec<Arc<dyn PhysicalExpr>>,
     remapped_children: Option<Vec<Arc<dyn PhysicalExpr>>>,
     // Inner state.
@@ -103,13 +111,15 @@ pub struct DynamicFilterSnapshot {
 
 impl DynamicFilterSnapshot {
     pub fn new(
+        expression_id: u64,
         children: Vec<Arc<dyn PhysicalExpr>>,
         remapped_children: Option<Vec<Arc<dyn PhysicalExpr>>>,
         generation: u64,
         inner_expr: Arc<dyn PhysicalExpr>,
         is_complete: bool,
     ) -> Self {
         Self {
+            expression_id,
             children,
             remapped_children,
             generation,
@@ -122,6 +132,10 @@ impl DynamicFilterSnapshot {
         &self.children
     }
 
+    pub fn expression_id(&self) -> u64 {
+        self.expression_id
+    }
+
     pub fn remapped_children(&self) -> Option<&[Arc<dyn PhysicalExpr>]> {
         self.remapped_children.as_deref()
     }
@@ -143,7 +157,8 @@ impl Display for DynamicFilterSnapshot {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "DynamicFilterSnapshot {{ children: {:?}, remapped_children: {:?}, generation: {}, inner_expr: {:?}, is_complete: {} }}",
+            "DynamicFilterSnapshot {{ expression_id: {}, children: {:?}, remapped_children: {:?}, generation: {}, inner_expr: {:?}, is_complete: {} }}",
+            self.expression_id,
             self.children,
             self.remapped_children,
             self.generation,
@@ -156,6 +171,7 @@ impl Display for DynamicFilterSnapshot {
 impl From<DynamicFilterSnapshot> for DynamicFilterPhysicalExpr {
     fn from(snapshot: DynamicFilterSnapshot) -> Self {
         let DynamicFilterSnapshot {
+            expression_id,
             children,
             remapped_children,
             generation,
@@ -171,6 +187,7 @@ impl From<DynamicFilterSnapshot> for DynamicFilterPhysicalExpr {
         let (state_watch, _) = watch::channel(state);
 
         Self {
+            expression_id,
             children,
             remapped_children,
             inner: Arc::new(RwLock::new(Inner {
@@ -193,6 +210,7 @@ impl From<&DynamicFilterPhysicalExpr> for DynamicFilterSnapshot {
             (inner.generation, Arc::clone(&inner.expr), inner.is_complete)
         };
         DynamicFilterSnapshot {
+            expression_id: expr.expression_id,
             children: expr.children.clone(),
             remapped_children: expr.remapped_children.clone(),
             generation,
@@ -252,6 +270,12 @@ impl Display for DynamicFilterPhysicalExpr {
 }
 
 impl DynamicFilterPhysicalExpr {
+    pub fn new_expression_id() -> u64 {
+        static NEXT_EXPRESSION_ID: LazyLock<AtomicU64> =
+            LazyLock::new(|| AtomicU64::new(random::<u64>()));
+        NEXT_EXPRESSION_ID.fetch_add(1, Ordering::Relaxed)
+    }
+
     /// Create a new [`DynamicFilterPhysicalExpr`]
     /// from an initial expression and a list of children.
     /// The list of children is provided separately because
@@ -285,6 +309,7 @@ impl DynamicFilterPhysicalExpr {
     ) -> Self {
         let (state_watch, _) = watch::channel(FilterState::InProgress { generation: 1 });
         Self {
+            expression_id: Self::new_expression_id(),
             children,
             remapped_children: None, // Initially no remapped children
             inner: Arc::new(RwLock::new(Inner::new(inner))),
@@ -476,6 +501,7 @@ impl PhysicalExpr for DynamicFilterPhysicalExpr {
         children: Vec<Arc<dyn PhysicalExpr>>,
     ) -> Result<Arc<dyn PhysicalExpr>> {
         Ok(Arc::new(Self {
+            expression_id: self.expression_id,
             children: self.children.clone(),
             remapped_children: Some(children),
             inner: Arc::clone(&self.inner),
@@ -559,13 +585,8 @@ impl PhysicalExpr for DynamicFilterPhysicalExpr {
         self.inner.read().generation
     }
 
-    fn expr_id(self: Arc<Self>, salt: &[u64]) -> Option<PhysicalExprId> {
-        Some(PhysicalExprId::new(
-            // Capture the outer arc, which contains children and the expr.
-            expr_id_from_arc(&self, salt),
-            // Capture the inner arc, which contains the expr only.
-            Some(expr_id_from_arc(&self.inner, salt)),
-        ))
+    fn expression_id(&self) -> Option<u64> {
+        Some(self.expression_id)
     }
 }
 
@@ -1033,7 +1054,7 @@ mod test {
     }
 
     #[tokio::test]
-    async fn test_expr_id() {
+    async fn test_expression_id() {
         let source_schema =
             Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
         let col_a = col("a", &source_schema).unwrap();
@@ -1056,38 +1077,24 @@ mod test {
         )
         .expect("reassign_expr_columns should succeed");
 
-        let derived_expr_id = Arc::clone(&derived)
-            .expr_id(&[])
-            .expect("combined filter should have an expr_id");
-        let source_expr_id = Arc::clone(&source)
-            .expr_id(&[])
-            .expect("source filter should have an expr_id");
-        let source_clone_expr_id = Arc::clone(&source_clone)
-            .expr_id(&[])
-            .expect("source clone should have an expr_id");
+        let derived_expression_id = derived
+            .expression_id()
+            .expect("derived filter should have an expression id");
+        let source_expression_id = source
+            .expression_id()
+            .expect("source filter should have an expression id");
+        let source_clone_expression_id = source_clone
+            .expression_id()
+            .expect("source clone should have an expression id");
 
         assert_eq!(
-            source_clone_expr_id.exact(),
-            source_expr_id.exact(),
-            "cloned filter should have the same exact id because the children are the same",
-        );
-
-        assert_eq!(
-            source_clone_expr_id.shallow(),
-            source_expr_id.shallow(),
-            "cloned filter should have the same shallow id because the exprs are the same",
-        );
-
-        assert_ne!(
-            derived_expr_id.exact(),
-            source_expr_id.exact(),
-            "filters should have different exact ids because the children are different",
+            source_clone_expression_id, source_expression_id,
+            "cloned filter should preserve its expression id",
         );
 
         assert_eq!(
-            derived_expr_id.shallow(),
-            source_expr_id.shallow(),
-            "filters should have the same shallow id because the exprs are the same",
+            derived_expression_id, source_expression_id,
+            "derived filters should carry forward the source expression id",
         );
     }
 }

datafusion/physical-plan/src/aggregates/mod.rs

@@ -1225,6 +1225,30 @@ impl AggregateExec {
         Some(total)
     }
 
+    /// The dynamic filter carried by this aggregate for cross-operator
+    /// pushdown, if one has been initialized. Returns the inner
+    /// `DynamicFilterPhysicalExpr` (the identity-bearing piece), not the
+    /// private `AggrDynFilter` wrapper.
+    pub fn dynamic_filter(&self) -> Option<Arc<DynamicFilterPhysicalExpr>> {
+        self.dynamic_filter.as_ref().map(|d| Arc::clone(&d.filter))
+    }
+
+    /// Replace the inner `DynamicFilterPhysicalExpr` while preserving the
+    /// auto-built `supported_accumulators_info`. Used by serde roundtrips to
+    /// restore the filter's `expression_id` after `try_new` has already minted
+    /// a fresh one via `init_dynamic_filter`.
+    pub fn with_dynamic_filter(mut self, filter: Arc<DynamicFilterPhysicalExpr>) -> Self {
+        if let Some(existing) = self.dynamic_filter.take() {
+            let supported_accumulators_info =
+                existing.supported_accumulators_info.clone();
+            self.dynamic_filter = Some(Arc::new(AggrDynFilter {
+                filter,
+                supported_accumulators_info,
+            }));
+        }
+        self
+    }
+
     /// Check if dynamic filter is possible for the current plan node.
     /// - If yes, init one inside `AggregateExec`'s `dynamic_filter` field.
     /// - If not supported, `self.dynamic_filter` should be kept `None`

datafusion/physical-plan/src/joins/hash_join/exec.rs

@@ -494,6 +494,21 @@ impl HashJoinExecBuilder {
         self.exec.dynamic_filter = filter;
         self
     }
+
+    /// Attach an explicit `DynamicFilterPhysicalExpr` to the join, preserving
+    /// its identity (`expression_id`). Used by serde roundtrips so a filter
+    /// doesn't get re-minted on decode. A fresh `build_accumulator` is created
+    /// — that is runtime state and not part of the wire format.
+    pub fn with_dynamic_filter_expr(
+        mut self,
+        filter: Arc<DynamicFilterPhysicalExpr>,
+    ) -> Self {
+        self.exec.dynamic_filter = Some(HashJoinExecDynamicFilter {
+            filter,
+            build_accumulator: OnceLock::new(),
+        });
+        self
+    }
 }
 
 impl From<&HashJoinExec> for HashJoinExecBuilder {
@@ -861,6 +876,14 @@ impl HashJoinExec {
         true
     }
 
+    /// The join's dynamic filter for probe-side pushdown, if one has been
+    /// attached (typically by filter-pushdown optimization). Returns the inner
+    /// `DynamicFilterPhysicalExpr` — the identity-bearing piece, not the
+    /// private wrapper.
+    pub fn dynamic_filter(&self) -> Option<Arc<DynamicFilterPhysicalExpr>> {
+        self.dynamic_filter.as_ref().map(|d| Arc::clone(&d.filter))
+    }
+
     /// left (build) side which gets hashed
     pub fn left(&self) -> &Arc<dyn ExecutionPlan> {
         &self.left

datafusion/physical-plan/src/sorts/sort.rs

@@ -920,6 +920,21 @@ impl SortExec {
         ))))
     }
 
+    /// The dynamic filter used by this sort for TopK pushdown, if any.
+    /// Returns `None` when `fetch` is `None` or the filter has not yet been
+    /// initialized (e.g. directly after `SortExec::new` before `with_fetch`).
+    pub fn dynamic_filter(&self) -> Option<Arc<DynamicFilterPhysicalExpr>> {
+        self.filter.as_ref().map(|f| f.read().expr())
+    }
+
+    /// Attach an explicit dynamic filter, replacing whatever was there.
+    /// Used by serde roundtrips to restore the filter's identity
+    /// (`expression_id`) rather than letting `with_fetch` mint a fresh one.
+    pub fn with_dynamic_filter(mut self, filter: Arc<DynamicFilterPhysicalExpr>) -> Self {
+        self.filter = Some(Arc::new(RwLock::new(TopKDynamicFilters::new(filter))));
+        self
+    }
+
     fn cloned(&self) -> Self {
         SortExec {
             input: Arc::clone(&self.input),