add check to verify predicate is always present when initiating probe side scan

Author: rkrishn7

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

@@ -1065,7 +1065,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() {
     ];
     let probe_repartition = Arc::new(
         RepartitionExec::try_new(
-            probe_scan,
+            Arc::clone(&probe_scan),
             Partitioning::Hash(probe_hash_exprs, partition_count),
         )
         .unwrap(),
@@ -1199,6 +1199,13 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() {
 
     let result = format!("{}", pretty_format_batches(&batches).unwrap());
 
+    let probe_scan_metrics = probe_scan.metrics().unwrap();
+
+    // The probe side had 4 rows, but after applying the dynamic filter only 2 rows should remain.
+    // The number of output rows from the probe side scan should stay consistent across executions.
+    // Issue: https://github.com/apache/datafusion/issues/17451
+    assert_eq!(probe_scan_metrics.output_rows().unwrap(), 2);
+
     insta::assert_snapshot!(
         result,
         @r"

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

@@ -27,6 +27,7 @@ use datafusion_datasource::{
 };
 use datafusion_physical_expr_common::physical_expr::fmt_sql;
 use datafusion_physical_optimizer::PhysicalOptimizerRule;
+use datafusion_physical_plan::filter::batch_filter;
 use datafusion_physical_plan::filter_pushdown::{FilterPushdownPhase, PushedDown};
 use datafusion_physical_plan::{
     displayable,
@@ -62,12 +63,6 @@ impl FileOpener for TestOpener {
         _file_meta: FileMeta,
         _file: PartitionedFile,
     ) -> Result<FileOpenFuture> {
-        if let Some(predicate) = &self.predicate {
-            println!(
-                "Predicate when calling open: {}",
-                fmt_sql(predicate.as_ref())
-            );
-        }
         let mut batches = self.batches.clone();
         if let Some(batch_size) = self.batch_size {
             let batch = concat_batches(&batches[0].schema(), &batches)?;
@@ -84,6 +79,12 @@ impl FileOpener for TestOpener {
             let (mapper, projection) = factory.map_schema(&batches[0].schema()).unwrap();
             let mut new_batches = Vec::new();
             for batch in batches {
+                let batch = if let Some(predicate) = &self.predicate {
+                    batch_filter(&batch, predicate)?
+                } else {
+                    batch
+                };
+
                 let batch = batch.project(&projection).unwrap();
                 let batch = mapper.map_batch(batch).unwrap();
                 new_batches.push(batch);