address feedback

Author: jayshrivastava

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

@@ -961,12 +961,12 @@ mod test {
         );
     }
 
-    /// Verifies that reassigning (which triggers `with_new_children`) followed
-    /// by rebuilding via `from_parts` preserves the filter's observable state
-    /// (original children, remapped children, expression id, and inner
-    /// generation/expr/is_complete).
+    /// Verifies that `from_parts` rebuilds a `DynamicFilterPhysicalExpr`
+    /// whose observable state (original children, remapped children,
+    /// expression id, inner generation/expr/is_complete) matches the source
+    /// filter.
     #[test]
-    fn test_current_snapshot_roundtrip() {
+    fn test_from_parts_preserves_state() {
         let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
         let col_a = col("a", &schema).unwrap();
 
@@ -981,13 +981,7 @@ mod test {
             expr as Arc<dyn PhysicalExpr>,
         );
 
-        // Update expression and mark complete
-        filter
-            .update(lit(42) as Arc<dyn PhysicalExpr>)
-            .expect("Update should succeed");
-        filter.mark_complete();
-
-        // Change the children of the expr.
+        // Add remapped children.
         let reassigned_schema = Arc::new(Schema::new(vec![
             Field::new("b", DataType::Int32, false),
             Field::new("a", DataType::Int32, false),
@@ -1001,6 +995,11 @@ mod test {
             .downcast_ref::<DynamicFilterPhysicalExpr>()
             .expect("Expected dynamic filter after reassignment");
 
+        reassigned
+            .update(lit(42) as Arc<dyn PhysicalExpr>)
+            .expect("Update should succeed");
+        reassigned.mark_complete();
+
         // Capture the parts and reconstruct.
         let reconstructed = DynamicFilterPhysicalExpr::from_parts(
             reassigned.expression_id,
@@ -1009,7 +1008,6 @@ mod test {
             reassigned.inner(),
         );
 
-        // Assert the reconstructed filter matches field-by-field.
         assert_eq!(
             reassigned.original_children(),
             reconstructed.original_children(),
@@ -1018,21 +1016,12 @@ mod test {
             reassigned.remapped_children(),
             reconstructed.remapped_children(),
         );
-        assert_eq!(
-            reassigned.expression_id().unwrap(),
-            reconstructed.expression_id().unwrap(),
-        );
-        let reassigned_inner = reassigned.inner();
-        let reconstructed_inner = reconstructed.inner();
-        assert_eq!(reassigned_inner.generation, reconstructed_inner.generation);
-        assert_eq!(
-            reassigned_inner.is_complete,
-            reconstructed_inner.is_complete
-        );
-        assert_eq!(
-            format!("{:?}", reassigned_inner.expr),
-            format!("{:?}", reconstructed_inner.expr),
-        );
+        assert_eq!(reassigned.expression_id(), reconstructed.expression_id());
+        let r = reassigned.inner();
+        let c = reconstructed.inner();
+        assert_eq!(r.generation, c.generation);
+        assert_eq!(r.is_complete, c.is_complete);
+        assert_eq!(format!("{:?}", r.expr), format!("{:?}", c.expr));
     }
 
     #[tokio::test]

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -2871,39 +2871,25 @@ fn assert_dynamic_filter_update_is_visible(
     Ok(())
 }
 
-/// Format the observable state of a `DynamicFilterPhysicalExpr` for
-/// equality comparison in roundtrip tests.
-fn fmt_dynamic_filter(filter: &DynamicFilterPhysicalExpr) -> String {
-    let inner = filter.inner();
-    format!(
-        "DynamicFilter {{ children: {:?}, remapped_children: {:?}, expression_id: {}, generation: {}, inner_expr: {:?}, is_complete: {} }}",
-        filter.original_children(),
-        filter.remapped_children(),
-        filter
-            .expression_id()
-            .expect("DynamicFilterPhysicalExpr always has an expression_id"),
-        inner.generation,
-        inner.expr,
-        inner.is_complete,
-    )
-}
-
-/// Assert that the actual dynamic filter snapshot matches the expected one
-/// after roundtrip.
-fn assert_dynamic_filter_snapshot_matches(
+/// Assert that two dynamic filters are equal both structurally (Debug output)
+/// and by identity (`expression_id`).
+///
+fn assert_dynamic_filters_equal(
     expected: &Arc<dyn PhysicalExpr>,
     actual: &Arc<dyn PhysicalExpr>,
 ) {
-    let expected_filter = expected
-        .downcast_ref::<DynamicFilterPhysicalExpr>()
-        .expect("Expected dynamic filter");
-    let actual_filter = actual
-        .downcast_ref::<DynamicFilterPhysicalExpr>()
-        .expect("Expected dynamic filter");
-
-    let actual_snapshot = fmt_dynamic_filter(actual_filter);
-    let expected_snapshot = fmt_dynamic_filter(expected_filter);
-    if expected_snapshot == actual_snapshot {
+    // TODO: Debug currently omits `expression_id` so the id has to be checked
+    // separately here. Once plan nodes like `SortExec` / `AggregateExec` /
+    // `HashJoinExec` serialize their own dynamic filter, Debug can include
+    // `expression_id`.
+    //
+    // See https://github.com/apache/datafusion/issues/20418
+    assert_eq!(expected.expression_id(), actual.expression_id());
+
+    // Structural.
+    let expected_dbg = format!("{expected:?}");
+    let actual_dbg = format!("{actual:?}");
+    if expected_dbg == actual_dbg {
         return;
     }
 
@@ -2913,12 +2899,7 @@ fn assert_dynamic_filter_snapshot_matches(
     let rewritten = Arc::clone(expected)
         .with_new_children(expected.children().iter().map(|c| Arc::clone(c)).collect())
         .expect("with_new_children on a dynamic filter should not fail");
-    let rewritten_snapshot = fmt_dynamic_filter(
-        rewritten
-            .downcast_ref::<DynamicFilterPhysicalExpr>()
-            .expect("with_new_children returns a DynamicFilterPhysicalExpr"),
-    );
-    assert_eq!(rewritten_snapshot, actual_snapshot);
+    assert_eq!(format!("{rewritten:?}"), actual_dbg);
 }
 
 // Two clones of a dynamic filter expression should be deduped to the exact same expression.
@@ -2936,15 +2917,9 @@ fn test_dynamic_filter_roundtrip_dedupe() -> Result<()> {
         )?;
 
     // Assert the filters are not modified during roundtrip.
-    assert_dynamic_filter_snapshot_matches(
-        &filter_expr_1,
-        &filter_expr_1_after_roundtrip,
-    );
-    assert_dynamic_filter_snapshot_matches(
-        &filter_expr_2,
-        &filter_expr_2_after_roundtrip,
-    );
-    assert_dynamic_filter_snapshot_matches(
+    assert_dynamic_filters_equal(&filter_expr_1, &filter_expr_1_after_roundtrip);
+    assert_dynamic_filters_equal(&filter_expr_2, &filter_expr_2_after_roundtrip);
+    assert_dynamic_filters_equal(
         &filter_expr_1_after_roundtrip,
         &filter_expr_2_after_roundtrip,
     );
@@ -2970,14 +2945,8 @@ fn test_dynamic_filter_plan_roundtrip_dedupe() -> Result<()> {
     ) = roundtrip_dynamic_filter_plan_pair()?;
 
     // Assert the filters are not modified during roundtrip.
-    assert_dynamic_filter_snapshot_matches(
-        &filter_expr_1,
-        &filter_expr_1_after_roundtrip,
-    );
-    assert_dynamic_filter_snapshot_matches(
-        &filter_expr_2,
-        &filter_expr_2_after_roundtrip,
-    );
+    assert_dynamic_filters_equal(&filter_expr_1, &filter_expr_1_after_roundtrip);
+    assert_dynamic_filters_equal(&filter_expr_2, &filter_expr_2_after_roundtrip);
 
     // Assert referential integrity.
     assert_dynamic_filter_update_is_visible(