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,18 @@ 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`].
+    ///
+    /// 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: once <https://github.com/apache/datafusion/pull/20416> lands and every
+// plan node serializes its dynamic filter through proto, drop this manual
+// impl and go back to `#[derive(Debug)]`. The id is excluded here so that
+// plan nodes which still auto-create a dynamic filter on decode
+// (`AggregateExec`, `HashJoinExec`) don't fail plan-equivalence assertions
+// purely on the fresh `expression_id`.
+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,
@@ -271,9 +285,9 @@ 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)
+        // Collision probability at ~10⁹ filters is 2.7%; realistic plans
+        // carry a handful. 2⁻⁶⁴ per pair is not worth an atomic counter.
+        random::<u64>()
     }
 
     /// Create a new [`DynamicFilterPhysicalExpr`]
@@ -309,9 +323,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)),
@@ -501,9 +515,9 @@ 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),
+            expression_id: self.expression_id,
             inner: Arc::clone(&self.inner),
             state_watch: self.state_watch.clone(),
             data_type: Arc::clone(&self.data_type),