fix tests

Author: kosiew

datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs

@@ -934,9 +934,6 @@ async fn test_hashjoin_dynamic_filter_pushdown() {
     stream.next().await.unwrap().unwrap();
 
     // Now check what our filter looks like
-    // `probe_keys=2` indicates the dynamic filter was derived from two probe-side
-    // join key columns (here `a` and `b`). It verifies the optimizer generated a
-    // probe-side predicate that constrains both join keys from the small build side.
     insta::assert_snapshot!(
         format!("{}", format_plan_for_test(&plan)),
         @r"
@@ -1245,8 +1242,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_high_cardinality() {
     .expect("optimizer should finish in time")
     .unwrap();
     let formatted = format_plan_for_test(&plan);
-    // Programmatic check: ensure some child received a dynamic filter for Inner join
-    assert_dynamic_filter_location_plan(&plan, &JoinType::Inner);
+    assert_contains!(&formatted, "DynamicFilterPhysicalExpr");
 
     assert_contains!(&formatted, "probe_keys=0");
 
@@ -1348,63 +1344,6 @@ fn assert_dynamic_filter_location(formatted: &str, join_type: &JoinType) {
     }
 }
 
-/// Programmatic check: inspect the `HashJoinExec` children and verify which
-/// side received a pushed predicate in the underlying `TestSource`.
-fn assert_dynamic_filter_location_plan(
-    plan: &Arc<dyn ExecutionPlan>,
-    join_type: &JoinType,
-) {
-    // Find top-level HashJoinExec
-    let join = plan
-        .as_any()
-        .downcast_ref::<HashJoinExec>()
-        .expect("expected HashJoinExec");
-
-    // Helper to check whether a child has a pushed predicate
-    let child_has_predicate = |child: &Arc<dyn ExecutionPlan>| -> bool {
-        if let Some(data_src) = child
-            .as_any()
-            .downcast_ref::<datafusion::datasource::source::DataSourceExec>(
-        ) {
-            if let Some((_, test_source)) =
-                data_src.downcast_to_file_source::<util::TestSource>()
-            {
-                return test_source.predicate().is_some();
-            }
-        }
-        false
-    };
-
-    match join_type {
-        JoinType::Left | JoinType::LeftSemi | JoinType::LeftAnti | JoinType::LeftMark => {
-            // For left-type joins the right side (probe) should receive the dynamic filter
-            assert!(
-                child_has_predicate(&join.right),
-                "expected predicate on right child"
-            );
-            assert!(
-                !child_has_predicate(&join.left),
-                "expected no predicate on left child"
-            );
-        }
-        JoinType::Right
-        | JoinType::RightSemi
-        | JoinType::RightAnti
-        | JoinType::RightMark => {
-            // For right-type joins the left side should receive the dynamic filter
-            assert!(
-                child_has_predicate(&join.left),
-                "expected predicate on left child"
-            );
-            assert!(
-                !child_has_predicate(&join.right),
-                "expected no predicate on right child"
-            );
-        }
-        _ => unreachable!(),
-    }
-}
-
 #[rstest(
     join_type,
     case::inner(JoinType::Inner),
@@ -1469,7 +1408,7 @@ async fn test_hashjoin_outer_dynamic_filter_pushdown(join_type: JoinType) {
         .optimize(plan, &config)
         .unwrap();
     let formatted = format_plan_for_test(&plan);
-    assert_dynamic_filter_location_plan(&plan, &join_type);
+    assert_dynamic_filter_location(&formatted, &join_type);
     assert_contains!(&formatted, "probe_keys=0");
 }
 
@@ -1503,8 +1442,14 @@ async fn test_hashjoin_left_dynamic_filter_pushdown_collect_left() {
         .optimize(plan, &config)
         .unwrap();
     let formatted = format_plan_for_test(&plan);
-    // Programmatic check: verify the dynamic filter was pushed to the expected side
-    assert_dynamic_filter_location_plan(&plan, &JoinType::Left);
+    assert_contains!(
+        &formatted,
+        "DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilterPhysicalExpr"
+    );
+    assert_contains!(
+        &formatted,
+        "DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=<none>"
+    );
     assert_contains!(&formatted, "probe_keys=0");
 }
 
@@ -1524,7 +1469,7 @@ async fn test_hashjoin_semi_dynamic_filter_pushdown(join_type: JoinType) {
         .optimize(plan, &config)
         .unwrap();
     let formatted = format_plan_for_test(&plan);
-    assert_dynamic_filter_location_plan(&plan, &join_type);
+    assert_dynamic_filter_location(&formatted, &join_type);
     assert_contains!(&formatted, "probe_keys=0");
 }
 
@@ -1544,7 +1489,7 @@ async fn test_hashjoin_anti_dynamic_filter_pushdown(join_type: JoinType) {
         .optimize(plan, &config)
         .unwrap();
     let formatted = format_plan_for_test(&plan);
-    assert_dynamic_filter_location_plan(&plan, &join_type);
+    assert_dynamic_filter_location(&formatted, &join_type);
     assert_contains!(&formatted, "probe_keys=0");
 }
 
@@ -1564,7 +1509,7 @@ async fn test_hashjoin_mark_dynamic_filter_pushdown(join_type: JoinType) {
         .optimize(plan, &config)
         .unwrap();
     let formatted = format_plan_for_test(&plan);
-    assert_dynamic_filter_location_plan(&plan, &join_type);
+    assert_dynamic_filter_location(&formatted, &join_type);
     assert_contains!(&formatted, "probe_keys=0");
 }
 

datafusion/core/tests/physical_optimizer/filter_pushdown/util.rs

@@ -152,7 +152,8 @@ impl FileSource for TestSource {
     }
 
     fn as_any(&self) -> &dyn Any {
-        todo!("should not be called")
+        // Allow downcasting a FileSource to the concrete TestSource in tests.
+        self
     }
 
     fn with_batch_size(&self, batch_size: usize) -> Arc<dyn FileSource> {