(datafusion) 01/01: test: cover CrossJoinExec row metrics across batched inputs (#25530)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25530-057f77f41e7dc74ce9b7a4af54a411faf2015a48
in repository https://gitbox.apache.org/repos/asf/datafusion.git

commit 816e002fe8c0932db891d1783d1ba94121c369ed
Author: Rishabh Gupta <ri...@gmail.com>
AuthorDate: Fri Oct 9 08:50:59 2026 +0000

    test: cover CrossJoinExec row metrics across batched inputs (#25530)
    
    ## Which issue does this PR close?
    
    Part of #25333 (`CrossJoinExec` test coverage only).
    
    ## Rationale for this change
    
    Join metrics can be incorrect without affecting query results. Existing
    `CrossJoinExec` tests did not assert `build_input_rows` or `input_rows`
    across different input batch layouts.
    
    ## What changes are included in this PR?
    
    Adds Rust test coverage that:
    
    * Verifies `build_input_rows`, `input_rows`, and `output_rows` for the
    existing single-batch case.
    * Runs the same 3 × 2 cross join with input batch sizes of 2 and 1.
    * Confirms the row metrics remain 3, 2, and 6 regardless of input
    batching.
    
    This is a test-only change.
    
    ## What is the testing strategy for this PR?
    
    ```bash
    cargo fmt --check
    cargo test -p datafusion-physical-plan --lib joins::cross_join
    cargo clippy -p datafusion-physical-plan --all-targets --all-features -- -D warnings
    ```
    
    All checks pass locally.
    
    ## Are there any user-facing changes?
    
    No.
---
 datafusion/physical-plan/src/joins/cross_join.rs | 71 +++++++++++++++++++++++-
 1 file changed, 70 insertions(+), 1 deletion(-)

diff --git a/datafusion/physical-plan/src/joins/cross_join.rs b/datafusion/physical-plan/src/joins/cross_join.rs
index e4dbbdf628..75b05f0b90 100644
--- a/datafusion/physical-plan/src/joins/cross_join.rs
+++ b/datafusion/physical-plan/src/joins/cross_join.rs
@@ -706,7 +706,9 @@ impl CrossJoinStream {
 mod tests {
     use super::*;
     use crate::common;
-    use crate::test::{assert_join_metrics, build_table_scan_i32};
+    use crate::test::{
+        TestMemoryExec, assert_join_metrics, build_table_i32, build_table_scan_i32,
+    };
 
     use datafusion_common::{assert_contains, test_util::batches_to_sort_string};
     use datafusion_execution::runtime_env::RuntimeEnvBuilder;
@@ -727,6 +729,29 @@ mod tests {
         Ok((columns_header, batches, metrics))
     }
 
+    /// Split a 3-column i32 table into `RecordBatch`es of at most `input_batch_size` rows.
+    fn build_table_scan_i32_chunked(
+        a: (&str, &Vec<i32>),
+        b: (&str, &Vec<i32>),
+        c: (&str, &Vec<i32>),
+        input_batch_size: usize,
+    ) -> Arc<dyn ExecutionPlan> {
+        let full = build_table_i32(a, b, c);
+        let n = full.num_rows();
+        let batches = (0..n)
+            .step_by(input_batch_size)
+            .map(|i| full.slice(i, input_batch_size.min(n - i)))
+            .collect();
+        TestMemoryExec::try_new_exec(&[batches], full.schema(), None).unwrap()
+    }
+
+    fn metric_count(metrics: &MetricsSet, name: &str) -> usize {
+        metrics
+            .sum_by_name(name)
+            .expect("missing metric")
+            .as_usize()
+    }
+
     #[tokio::test]
     async fn test_stats_cartesian_product() {
         let left_row_count = 11;
@@ -966,6 +991,50 @@ mod tests {
         ");
 
         assert_join_metrics!(metrics, 6);
+        assert_eq!(metric_count(&metrics, "build_input_rows"), 3);
+        assert_eq!(metric_count(&metrics, "input_rows"), 2);
+
+        Ok(())
+    }
+
+    /// `build_input_rows`, `input_rows`, and `output_rows` must count each row
+    /// once when the build and probe sides arrive as several batches.
+    #[tokio::test]
+    async fn join_metrics_count_rows_once_across_batches() -> Result<()> {
+        let task_ctx = Arc::new(TaskContext::default());
+        // 2: left 2+1, right 2.  1: left 3×1, right 2×1.
+        for (input_batch_size, expected_build_batches, expected_probe_batches) in
+            [(2, 2, 1), (1, 3, 2)]
+        {
+            let left = build_table_scan_i32_chunked(
+                ("a1", &vec![1, 2, 3]),
+                ("b1", &vec![4, 5, 6]),
+                ("c1", &vec![7, 8, 9]),
+                input_batch_size,
+            );
+            let right = build_table_scan_i32_chunked(
+                ("a2", &vec![10, 11]),
+                ("b2", &vec![12, 13]),
+                ("c2", &vec![14, 15]),
+                input_batch_size,
+            );
+
+            let (_, batches, metrics) =
+                join_collect(left, right, Arc::clone(&task_ctx)).await?;
+            let num_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
+            assert_eq!(num_rows, 6);
+            assert_join_metrics!(metrics, 6);
+            assert_eq!(metric_count(&metrics, "build_input_rows"), 3);
+            assert_eq!(metric_count(&metrics, "input_rows"), 2);
+            assert_eq!(
+                metric_count(&metrics, "build_input_batches"),
+                expected_build_batches
+            );
+            assert_eq!(
+                metric_count(&metrics, "input_batches"),
+                expected_probe_batches
+            );
+        }
 
         Ok(())
     }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org