proto: serialize and dedupe dynamic filters

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

@@ -327,6 +327,14 @@ impl DynamicFilterPhysicalExpr {
         Arc::strong_count(self) > 1 || Arc::strong_count(&self.inner) > 1
     }
 
+    /// Returns a unique identifier for the inner shared state.
+    ///
+    /// Useful for checking if two [Arc<PhysicalExpr>] with the same
+    /// underlying [DynamicFilterPhysicalExpr] are the same.
+    pub fn inner_id(&self) -> u64 {
+        Arc::as_ptr(&self.inner) as *const () as u64
+    }
+
     fn render(
         &self,
         f: &mut std::fmt::Formatter<'_>,

datafusion/proto/proto/datafusion.proto

@@ -860,6 +860,12 @@ message PhysicalExprNode {
   // across serde roundtrips.
   optional uint64 expr_id = 30;
 
+  // For DynamicFilterPhysicalExpr, this identifies the shared inner state.
+  // Multiple expressions may have different expr_id values (different outer Arc wrappers)
+  // but the same dynamic_filter_inner_id (shared inner state).
+  // Used to reconstruct shared inner state during deserialization.
+  optional uint64 dynamic_filter_inner_id = 31;
+
   oneof ExprType {
     // column references
     PhysicalColumn column = 1;
@@ -897,9 +903,16 @@ message PhysicalExprNode {
     UnknownColumn unknown_column = 20;
 
     PhysicalHashExprNode hash_expr = 21;
+
+    PhysicalDynamicFilterNode dynamic_filter = 22;
   }
 }
 
+message PhysicalDynamicFilterNode {
+  repeated PhysicalExprNode children = 1;
+  PhysicalExprNode initial_expr = 2;
+}
+
 message PhysicalScalarUdfNode {
   string name = 1;
   repeated PhysicalExprNode args = 2;

datafusion/proto/src/physical_plan/from_proto.rs

@@ -58,6 +58,7 @@ use super::{
 use crate::logical_plan::{self};
 use crate::protobuf::physical_expr_node::ExprType;
 use crate::{convert_required, protobuf};
+use datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr;
 
 impl From<&protobuf::PhysicalColumn> for Column {
     fn from(c: &protobuf::PhysicalColumn) -> Column {
@@ -495,6 +496,27 @@ pub fn parse_physical_expr_with_converter(
                 hash_expr.description.clone(),
             ))
         }
+        ExprType::DynamicFilter(dynamic_filter) => {
+            let children = parse_physical_exprs(
+                &dynamic_filter.children,
+                ctx,
+                input_schema,
+                codec,
+                proto_converter,
+            )?;
+
+            let initial_expr = parse_required_physical_expr(
+                dynamic_filter.initial_expr.as_deref(),
+                ctx,
+                "initial_expr",
+                input_schema,
+                codec,
+                proto_converter,
+            )?;
+
+            // Constructor signature is: new(children, inner)
+            Arc::new(DynamicFilterPhysicalExpr::new(children, initial_expr))
+        }
         ExprType::Extension(extension) => {
             let inputs: Vec<Arc<dyn PhysicalExpr>> = extension
                 .inputs