wip

Author: jayshrivastava

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

@@ -163,6 +163,9 @@ pub trait PhysicalExpr: Any + Send + Sync + Display + Debug + DynEq + DynHash {
     fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>>;
 
     /// Returns a new PhysicalExpr where all children were replaced by new exprs.
+    ///
+    /// If the implementation returns a [`PhysicalExpr::expression_id`], then
+    /// the identifier should be preserved by the new expression.
     fn with_new_children(
         self: Arc<Self>,
         children: Vec<Arc<dyn PhysicalExpr>>,
@@ -439,9 +442,19 @@ pub trait PhysicalExpr: Any + Send + Sync + Display + Debug + DynEq + DynHash {
         ExpressionPlacement::KeepInPlace
     }
 
-    /// 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.
+    /// Return a stable, globally-unique identifier for this [`PhysicalExpr`], if it
+    /// has one.
+    ///
+    /// This identifier tracks which expressions which are connected (e.g. `DynamicFilterPhysicalExpr`
+    /// where two expressions may be different but store the same mutable inner state). Tracking
+    /// connected expressions helps preserve referential integrity within plan nodes
+    /// during serialization and deserialization.
+    ///
+    /// This id must be preserved across [`PhysicalExpr::with_new_children`] or any other
+    /// methods which may want to preserve identity.
+    ///
+    /// Default is `None`: the expression has no identity worth preserving across a
+    /// serialization boundary.
     fn expression_id(&self) -> Option<u64> {
         None
     }

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

@@ -16,14 +16,7 @@
 // under the License.
 
 use parking_lot::RwLock;
-use std::{
-    fmt::Display,
-    hash::Hash,
-    sync::{
-        Arc, LazyLock,
-        atomic::{AtomicU64, Ordering},
-    },
-};
+use std::{fmt::Display, hash::Hash, sync::Arc};
 use tokio::sync::watch;
 
 use crate::PhysicalExpr;
@@ -63,9 +56,7 @@ impl FilterState {
 /// For more background, please also see the [Dynamic Filters: Passing Information Between Operators During Execution for 25x Faster Queries blog]
 ///
 /// [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.
@@ -74,6 +65,10 @@ pub struct DynamicFilterPhysicalExpr {
     /// If any of the children were remapped / modified (e.g. to adjust for projections) we need to keep track of the new children
     /// so that when we update `current()` in subsequent iterations we can re-apply the replacements.
     remapped_children: Option<Vec<Arc<dyn PhysicalExpr>>>,
+    /// Unique identifier for this dynamic filter.
+    ///
+    /// Derived filters (ex. via `with_new_children`) should inherit the expression id of the source filter.
+    expression_id: u64,
     /// The source of dynamic filters.
     inner: Arc<RwLock<Inner>>,
     /// Broadcasts filter state (updates and completion) to all waiters.
@@ -100,9 +95,9 @@ 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>>>,
+    expression_id: u64,
     // Inner state.
     generation: u64,
     inner_expr: Arc<dyn PhysicalExpr>,
@@ -119,9 +114,9 @@ impl DynamicFilterSnapshot {
         is_complete: bool,
     ) -> Self {
         Self {
-            expression_id,
             children,
             remapped_children,
+            expression_id,
             generation,
             inner_expr,
             is_complete,
@@ -153,14 +148,33 @@ impl DynamicFilterSnapshot {
     }
 }
 
+// TODO: Include expression_id in debug output. 
+//
+// 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.
+impl std::fmt::Debug for DynamicFilterPhysicalExpr {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("DynamicFilterPhysicalExpr")
+            .field("children", &self.children)
+            .field("remapped_children", &self.remapped_children)
+            .field("inner", &self.inner)
+            .field("state_watch", &self.state_watch)
+            .field("data_type", &self.data_type)
+            .field("nullable", &self.nullable)
+            .finish()
+    }
+}
+
 impl Display for DynamicFilterSnapshot {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "DynamicFilterSnapshot {{ expression_id: {}, children: {:?}, remapped_children: {:?}, generation: {}, inner_expr: {:?}, is_complete: {} }}",
-            self.expression_id,
+            "DynamicFilterSnapshot {{ children: {:?}, remapped_children: {:?}, expression_id: {}, generation: {}, inner_expr: {:?}, is_complete: {} }}",
             self.children,
             self.remapped_children,
+            self.expression_id,
             self.generation,
             self.inner_expr,
             self.is_complete
@@ -171,9 +185,9 @@ impl Display for DynamicFilterSnapshot {
 impl From<DynamicFilterSnapshot> for DynamicFilterPhysicalExpr {
     fn from(snapshot: DynamicFilterSnapshot) -> Self {
         let DynamicFilterSnapshot {
-            expression_id,
             children,
             remapped_children,
+            expression_id,
             generation,
             inner_expr,
             is_complete,
@@ -187,9 +201,9 @@ impl From<DynamicFilterSnapshot> for DynamicFilterPhysicalExpr {
         let (state_watch, _) = watch::channel(state);
 
         Self {
-            expression_id,
             children,
             remapped_children,
+            expression_id,
             inner: Arc::new(RwLock::new(Inner {
                 generation,
                 expr: inner_expr,
@@ -210,9 +224,9 @@ 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(),
+            expression_id: expr.expression_id,
             generation,
             inner_expr,
             is_complete,
@@ -270,12 +284,6 @@ 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
@@ -309,9 +317,9 @@ 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
+            expression_id: Self::new_expression_id(),
             inner: Arc::new(RwLock::new(Inner::new(inner))),
             state_watch,
             data_type: Arc::new(RwLock::new(None)),
@@ -485,6 +493,11 @@ impl DynamicFilterPhysicalExpr {
 
         write!(f, " ]")
     }
+
+    /// Generate a new expression id for this filter.
+    pub fn new_expression_id() -> u64 {
+       random::<u64>()
+    }
 }
 
 impl PhysicalExpr for DynamicFilterPhysicalExpr {
@@ -501,9 +514,11 @@ 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),
+            // Note that we ensure the derived expression linked to `self` 
+            // via the unique identifier.
+            expression_id: self.expression_id,
             inner: Arc::clone(&self.inner),
             state_watch: self.state_watch.clone(),
             data_type: Arc::clone(&self.data_type),

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

@@ -920,21 +920,6 @@ 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),

datafusion/proto/proto/datafusion.proto

@@ -870,12 +870,12 @@ message PhysicalExtensionNode {
   repeated PhysicalPlanNode inputs = 2;
 }
 
+// physical expressions
 message PhysicalExprNode {
   // Was date_time_interval_expr
   reserved 17;
 
-  // Was expr_id (generic per-expression dedup identifier). Identity is now
-  // carried only by DynamicFilterPhysicalExpr via PhysicalDynamicFilterNode.expression_id.
+  // Was expr_id
   reserved 30;
 
   oneof ExprType {
@@ -926,8 +926,6 @@ message PhysicalDynamicFilterNode {
   uint64 generation = 3;
   PhysicalExprNode inner_expr = 4;
   bool is_complete = 5;
-  // Process-local identifier preserved across serde roundtrip so multiple
-  // occurrences of the same dynamic filter in a plan share inner state.
   uint64 expression_id = 6;
 }
 
@@ -1172,10 +1170,6 @@ message HashJoinExecNode {
   JoinFilter filter = 8;
   repeated uint32 projection = 9;
   bool null_aware = 10;
-  // Optional dynamic filter pushed down to the probe side.
-  // Carries an `expression_id` so shared identity is preserved across a
-  // serde roundtrip.
-  PhysicalExprNode dynamic_filter = 11;
 }
 
 enum StreamPartitionMode {
@@ -1307,10 +1301,6 @@ message AggregateExecNode {
   repeated MaybeFilter filter_expr = 10;
   AggLimit limit = 11;
   bool has_grouping_set = 12;
-  // Optional dynamic filter attached to this aggregate (min/max pushdown).
-  // Carries an `expression_id` so shared identity is preserved across a
-  // serde roundtrip rather than being re-minted by `init_dynamic_filter`.
-  PhysicalExprNode dynamic_filter = 13;
 }
 
 message GlobalLimitExecNode {
@@ -1332,10 +1322,6 @@ message SortExecNode {
   // Maximum number of highest/lowest rows to fetch; negative means no limit
   int64 fetch = 3;
   bool preserve_partitioning = 4;
-  // Optional TopK dynamic filter attached to this sort. Present when `fetch`
-  // is set and the filter has been initialized. Carries an `expression_id`
-  // so shared identity is preserved across a serde roundtrip.
-  PhysicalExprNode dynamic_filter = 5;
 }
 
 message SortPreservingMergeExecNode {