perf: Emit LeftSemi hash join rows while probing

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

@@ -888,13 +888,28 @@ impl HashJoinStream {
             (left_indices, right_indices)
         };
 
-        // mark joined left-side indices as visited, if required by join type
+        // Mark joined build-side indices as visited, if the join type tracks
+        // them. A semi join emits a build row the first time it matches, so
+        // collect the rows whose bit flips here and emit those.
+        let mut first_matched = None;
         if need_produce_result_in_final(self.join_type) {
             let mut bitmap = build_side.left_data.visited_indices_bitmap().lock();
-            left_indices.iter().flatten().for_each(|x| {
-                bitmap.set_bit(x as usize, true);
-            });
+            if self.join_type == JoinType::LeftSemi {
+                let mut matched = Vec::new();
+                for index in left_indices.iter().flatten() {
+                    if !bitmap.get_bit(index as usize) {
+                        bitmap.set_bit(index as usize, true);
+                        matched.push(index);
+                    }
+                }
+                first_matched = Some(UInt64Array::from(matched));
+            } else {
+                left_indices.iter().flatten().for_each(|x| {
+                    bitmap.set_bit(x as usize, true);
+                });
+            }
         }
+        let semi_matched = first_matched;
 
         // The goals of index alignment for different join types are:
         //
@@ -927,13 +942,18 @@ impl HashJoinStream {
             last_joined_right_idx.map_or(0, |v| v + 1)
         };
 
-        let (left_indices, mut right_indices) = adjust_indices_by_join_type(
-            left_indices,
-            right_indices,
-            index_alignment_range_start..index_alignment_range_end,
-            self.join_type,
-            self.right_side_ordered,
-        )?;
+        let (left_indices, mut right_indices) = match semi_matched {
+            // A semi join emits each build row once, the first time it matches,
+            // so there is nothing left to align or to produce at the end.
+            Some(matched) => (matched, UInt32Array::from_iter_values(vec![])),
+            None => adjust_indices_by_join_type(
+                left_indices,
+                right_indices,
+                index_alignment_range_start..index_alignment_range_end,
+                self.join_type,
+                self.right_side_ordered,
+            )?,
+        };
 
         // If null-aware RightAnti join, we don't want to emit NULL probe keys
         if self.join_type == JoinType::RightAnti && self.null_aware {
@@ -1006,7 +1026,10 @@ impl HashJoinStream {
     ) -> Result<StatefulStreamResult<Option<RecordBatch>>> {
         let timer = self.join_metrics.join_time.timer();
 
-        if !need_produce_result_in_final(self.join_type) {
+        // `LeftSemi` has already emitted its matched rows while probing.
+        if !need_produce_result_in_final(self.join_type)
+            || self.join_type == JoinType::LeftSemi
+        {
             self.state = HashJoinStreamState::Completed;
             return Ok(StatefulStreamResult::Continue);
         }

datafusion/sqllogictest/test_files/push_down_filter_parquet.slt

@@ -888,7 +888,7 @@ WHERE EXISTS (
 );
 ----
 Plan with Metrics
-01)HashJoinExec: mode=CollectLeft, join_type=LeftSemi, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=2, input_rows=4, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
+01)HashJoinExec: mode=CollectLeft, join_type=LeftSemi, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, predicate_cache_inner_records=0, predicate_cache_records=0, scan_efficiency_ratio=19.58% (196/1.00 K)]
 03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_probe.parquet]]}, projection=[a, b], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_count@2 AND b_max@4 >= ba AND b_null_count@5 != row_count@2 AND b_min@6 <= bb, required_guarantees=[], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=2, pushdown_rows_pruned=2, predicate_cache_inner_records=8, predicate_cache_records=4, scan_efficiency_ratio=14.89% (154/1.03 K)]