fix: Track metrics in hash joins with empty build sides (#20810)

## Which issue does this PR close?

- Closes #20809.

## Rationale for this change

Show the correct metrics when executing hash joins.

## What changes are included in this PR?

- The fast path in `process_probe_batch` now stores the output batch in
`output_buffer` instead of returning it, since the metrics are only
updated with batches stored there.
- Added unit test.

## Are these changes tested?

Yes.

## Are there any user-facing changes?

No.

---------

Co-authored-by: Yongting You <2010youy01@gmail.com>

Author: nuno-faria

datafusion/physical-plan/src/joins/hash_join/stream.rs

@@ -655,10 +655,10 @@ impl HashJoinStream {
                 self.join_type,
             )?;
             timer.done();
-
+            self.output_buffer.push_batch(result)?;
             self.state = HashJoinStreamState::FetchProbeBatch;
 
-            return Ok(StatefulStreamResult::Ready(Some(result)));
+            return Ok(StatefulStreamResult::Continue);
         }
 
         // get the matched by join keys indices

datafusion/sqllogictest/test_files/joins.slt

@@ -5056,6 +5056,49 @@ LEFT ANTI JOIN t2 ON k1 = k2
 WHERE k1 < 0
 ----
 
+# Also check that the reported number of output rows/batches are correct in the "empty build side"
+# optimization.
+# Issue: https://github.com/apache/datafusion/issues/20809
+query TT
+EXPLAIN ANALYZE
+WITH t1 (k) AS (
+    VALUES (1), (2)
+), t2 (k) AS (
+    VALUES (1)
+)
+SELECT *
+FROM t1
+LEFT ANTI JOIN (
+    SELECT *
+    FROM t2
+    WHERE k <> 1
+) t2 ON t1.k = t2.k;
+----
+Plan with Metrics
+01)HashJoinExec: mode=CollectLeft, join_type=RightAnti, on=[(k@0, k@0)], metrics=[output_rows=2, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>, output_batches=1, array_map_created_count=0, build_input_batches=0, build_input_rows=0, input_batches=1, input_rows=2, build_mem_used=<slt:ignore>, build_time=<slt:ignore>, join_time=<slt:ignore>, avg_fanout=N/A (0/0), probe_hit_rate=0% (0/2)]     
+02)--ProjectionExec: expr=[column1@0 as k], metrics=[output_rows=0, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>, output_batches=0, expr_0_eval_time=<slt:ignore>]
+03)----FilterExec: column1@0 != 1, metrics=[output_rows=0, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>, output_batches=0, selectivity=0% (0/1)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1], metrics=[]
+05)--ProjectionExec: expr=[column1@0 as k], metrics=[output_rows=2, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>, output_batches=1, expr_0_eval_time=<slt:ignore>]
+06)----DataSourceExec: partitions=1, partition_sizes=[1], metrics=[]
+
+query I
+WITH t1 (k) AS (
+    VALUES (1), (2)
+), t2 (k) AS (
+    VALUES (1)
+)
+SELECT *
+FROM t1
+LEFT ANTI JOIN (
+    SELECT *
+    FROM t2
+    WHERE k <> 1
+) t2 ON t1.k = t2.k;
+----
+1
+2
+
 # Mark testing
 statement ok
 CREATE OR REPLACE TABLE t1(b INT, c INT, d INT);