better assertions

Author: jayshrivastava

datafusion/proto/src/physical_plan/mod.rs

@@ -3885,13 +3885,7 @@ impl PhysicalProtoConverterExtension for DeduplicatingDeserializer {
         let mut cache = self.cache.borrow_mut();
         if let Some(cached) = cache.get(&id) {
             let children: Vec<_> = parsed.children().into_iter().cloned().collect();
-            let cached_children: Vec<_> =
-                cached.children().into_iter().cloned().collect();
-            return if children == cached_children {
-                Ok(Arc::clone(cached))
-            } else {
-                Arc::clone(cached).with_new_children(children)
-            };
+            return Arc::clone(cached).with_new_children(children);
         }
 
         cache.insert(id, Arc::clone(&parsed));

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -2877,21 +2877,38 @@ fn assert_dynamic_filter_update_is_visible(
     Ok(())
 }
 
+/// Assert that the actual dynamic filter snapshot matches the expected one
+/// after roundtrip.
 fn assert_dynamic_filter_snapshot_matches(
     expected: &Arc<dyn PhysicalExpr>,
     actual: &Arc<dyn PhysicalExpr>,
 ) {
-    let expected = expected
+    let expected_filter = expected
         .downcast_ref::<DynamicFilterPhysicalExpr>()
         .expect("Expected dynamic filter");
-    let actual = actual
+    let actual_filter = actual
         .downcast_ref::<DynamicFilterPhysicalExpr>()
         .expect("Expected dynamic filter");
 
-    assert_eq!(
-        DynamicFilterSnapshot::from(expected).to_string(),
-        DynamicFilterSnapshot::from(actual).to_string(),
-    );
+    let actual_snapshot = DynamicFilterSnapshot::from(actual_filter).to_string();
+    let expected_snapshot = DynamicFilterSnapshot::from(expected_filter).to_string();
+    if expected_snapshot == actual_snapshot {
+        return;
+    }
+
+    // Note that the `DeduplicatingDeserializer` routes every cache hit through
+    // `with_new_children`. This produces an equivalent expression, but with
+    // remapped children that are equal to the original. Handle that case here.
+    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 = DynamicFilterSnapshot::from(
+        rewritten
+            .downcast_ref::<DynamicFilterPhysicalExpr>()
+            .expect("with_new_children returns a DynamicFilterPhysicalExpr"),
+    )
+    .to_string();
+    assert_eq!(rewritten_snapshot, actual_snapshot);
 }
 
 // Two clones of a dynamic filter expression should be deduped to the exact same expression.
@@ -2943,19 +2960,6 @@ fn test_dynamic_filter_plan_roundtrip_dedupe() -> Result<()> {
     ) = roundtrip_dynamic_filter_plan_pair()?;
 
     // Assert the filters are not modified during roundtrip.
-    //
-    // There's a small technicality that `filter_expr_1` is rewritten to an equivalent expression
-    // during deserialization, so we capture that here by calling
-    // `filter.with_new_children(filter.children)`.
-    let filter_expr_1_children = Arc::clone(&filter_expr_1)
-        .children()
-        .iter()
-        .map(|child| Arc::clone(child))
-        .collect::<Vec<_>>();
-    let filter_expr_1 = filter_expr_1
-        .clone()
-        .with_new_children(filter_expr_1_children)
-        .unwrap();
     assert_dynamic_filter_snapshot_matches(
         &filter_expr_1,
         &filter_expr_1_after_roundtrip,