wip

Author: jayshrivastava

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

@@ -1021,85 +1021,53 @@ mod test {
     }
 
     #[tokio::test]
-    async fn test_with_new_children_preserves_shared_inner_state() {
+    async fn test_expr_id() {
         // Create a source filter
         let source = Arc::new(DynamicFilterPhysicalExpr::new(
             vec![],
             lit(42) as Arc<dyn PhysicalExpr>,
         ));
+        let source_clone = Arc::clone(&source);
 
-        // Create a target filter with different children
+        // Create a derived filter with different children
         let schema = Arc::new(Schema::new(vec![Field::new("x", DataType::Int32, false)]));
         let col_x = col("x", &schema).unwrap();
-        // Rebuild a wrapper from the source filter using the target's children.
-        let combined = Arc::clone(&source)
+        let derived = Arc::clone(&source)
             .with_new_children(vec![Arc::clone(&col_x)])
             .unwrap();
 
-        let combined_expr_id = Arc::clone(&combined)
+        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");
-        assert_ne!(
-            combined_expr_id.exact(),
-            source_expr_id.exact(),
-            "with_new_children should produce a distinct outer Arc identity",
-        );
-        assert_eq!(
-            combined_expr_id.shallow(),
-            source_expr_id.shallow(),
-            "dynamic filters with shared inner state should have the same shallow identity",
-        );
-
-        let combined_dyn_filter = combined
-            .as_any()
-            .downcast_ref::<DynamicFilterPhysicalExpr>()
-            .unwrap();
+        let source_clone_expr_id = Arc::clone(&source_clone)
+            .expr_id(&[])
+            .expect("source clone should have an expr_id");
 
-        // Verify the children are unchanged.
         assert_eq!(
-            format!("{:?}", combined.children()),
-            format!("{:?}", vec![col_x]),
-            "Combined filter's children should be unchanged"
+            source_clone_expr_id.exact(),
+            source_expr_id.exact(),
+            "cloned filter should have the same exact id because the outer Arc is the same",
         );
 
-        // Verify inner expression changed to the one from source.
         assert_eq!(
-            format!("{:?}", combined_dyn_filter.current().unwrap()),
-            format!("{:?}", lit(42)),
-            "Combined filter should have inner expression from source filter"
+            source_clone_expr_id.shallow(),
+            source_expr_id.shallow(),
+            "cloned filter should have the same shallow id because the inner state is the same",
         );
 
-        // Verify that completing one filter also completes the other.
-        let combined_binding = Arc::clone(&combined) as Arc<dyn PhysicalExpr>;
-        #[expect(clippy::disallowed_methods)]
-        let wait_handle = tokio::task::spawn({
-            async move {
-                let df = combined_binding
-                    .as_any()
-                    .downcast_ref::<DynamicFilterPhysicalExpr>()
-                    .unwrap();
-                df.wait_complete().await;
-                format!("{:?}", df.current().unwrap())
-            }
-        });
-        source.update(lit(999) as Arc<dyn PhysicalExpr>).unwrap();
-        source.mark_complete();
-
-        // The linked filter should be notified via the shared watch
-        let result = tokio::time::timeout(std::time::Duration::from_secs(1), wait_handle)
-            .await
-            .expect(
-                "linked filter should have been notified of completion within timeout",
-            )
-            .expect("task should not panic");
+        assert_ne!(
+            derived_expr_id.exact(),
+            source_expr_id.exact(),
+            "filters should have different exact ids because the children are different",
+        );
 
         assert_eq!(
-            result,
-            format!("{:?}", lit(999)),
-            "linked filter should see source's updated inner expression"
+            derived_expr_id.shallow(),
+            source_expr_id.shallow(),
+            "filters should have the same shallow id because they are the same expression",
         );
     }
 }

datafusion/proto/src/physical_plan/to_proto.rs

@@ -33,7 +33,6 @@ use datafusion_datasource_parquet::file_format::ParquetSink;
 use datafusion_expr::WindowFrame;
 use datafusion_physical_expr::ScalarFunctionExpr;
 use datafusion_physical_expr::window::{SlidingAggregateWindowExpr, StandardWindowExpr};
-use datafusion_physical_expr_common::physical_expr::PhysicalExprId;
 use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
 use datafusion_physical_plan::expressions::{
     BinaryExpr, CaseExpr, CastExpr, Column, DynamicFilterPhysicalExpr,
@@ -54,13 +53,6 @@ use crate::protobuf::{
     physical_aggregate_expr_node, physical_window_expr_node,
 };
 
-fn to_proto_expr_id(expr_id: PhysicalExprId) -> protobuf::PhysicalExprId {
-    protobuf::PhysicalExprId {
-        exact: expr_id.exact(),
-        shallow: expr_id.shallow(),
-    }
-}
-
 #[expect(clippy::needless_pass_by_value)]
 pub fn serialize_physical_aggr_expr(
     aggr_expr: Arc<AggregateFunctionExpr>,