Merge pull request #64 from poyrazK/fix/vectorized-hash-join-first-match-bug

fix(vectorized): remove break to allow multiple matches per left row in hash join

Author: poyrazK

include/executor/vectorized_operator.hpp

@@ -577,6 +577,12 @@ class VectorizedHashJoinOperator : public VectorizedOperator {
     std::vector<bool> left_matched_in_batch_;
     std::vector<size_t> unmatched_indices_;
 
+    // Probe state for resumable bucket scanning (prevents batch overflow)
+    bool resuming_bucket_scan_ = false;  // True if we're resuming a mid-bucket scan
+    size_t resumed_bucket_idx_ = 0;      // Bucket index when resuming
+    size_t resumed_entry_idx_ = 0;       // Entry index within bucket when resuming
+    common::Value resumed_key_val_;      // Key value being probed when resuming
+
     // Join type
     JoinType join_type_;
 
@@ -711,10 +717,52 @@ class VectorizedHashJoinOperator : public VectorizedOperator {
                 left_row_idx_ = 0;
                 // Reset matched tracking for new batch
                 std::fill(left_matched_in_batch_.begin(), left_matched_in_batch_.end(), false);
+                // Clear resume state when advancing to new batch
+                resuming_bucket_scan_ = false;
             }
 
             // Process rows in current batch
             while (left_row_idx_ < left_batch_->row_count() && out_batch.row_count() < BATCH_SIZE) {
+                // Check if we need to resume an interrupted bucket scan
+                if (resuming_bucket_scan_) {
+                    // We were in the middle of scanning a bucket - resume from saved position
+                    const auto& key_val = resumed_key_val_;
+                    auto& bucket = buckets_[resumed_bucket_idx_];
+                    bool found_match = left_matched_in_batch_[left_row_idx_];
+
+                    // Resume scanning bucket from resumed_entry_idx_
+                    for (size_t i = resumed_entry_idx_; i < bucket.key_values.size(); ++i) {
+                        if (out_batch.row_count() >= BATCH_SIZE) {
+                            // Batch full - save state and return
+                            resuming_bucket_scan_ = true;
+                            resumed_bucket_idx_ = resumed_bucket_idx_;
+                            resumed_entry_idx_ = i;
+                            resumed_key_val_ = key_val;
+                            return true;  // Caller must consume batch before continuing
+                        }
+
+                        const auto& bucket_key = bucket.key_values[i][right_key_col_idx_];
+                        if (bucket_key == key_val) {
+                            emit_joined_row(out_batch, left_row_idx_, bucket.payload_rows[i]);
+                            found_match = true;
+                            if (join_type_ == JoinType::Left) {
+                                left_matched_in_batch_[left_row_idx_] = true;
+                            }
+                        }
+                    }
+
+                    // Finished scanning this bucket
+                    resuming_bucket_scan_ = false;
+
+                    // Track unmatched for LEFT join
+                    if (join_type_ == JoinType::Left && !found_match) {
+                        unmatched_indices_.push_back(left_row_idx_);
+                    }
+
+                    left_row_idx_++;
+                    continue;
+                }
+
                 const auto& key_val = left_batch_->get_column(left_key_col_idx_).get(left_row_idx_);
 
                 if (key_val.is_null()) {
@@ -732,6 +780,15 @@ class VectorizedHashJoinOperator : public VectorizedOperator {
                 // Search for match in this bucket
                 bool found_match = false;
                 for (size_t i = 0; i < bucket.key_values.size(); ++i) {
+                    if (out_batch.row_count() >= BATCH_SIZE) {
+                        // Batch full - save state and return
+                        resuming_bucket_scan_ = true;
+                        resumed_bucket_idx_ = bucket_idx;
+                        resumed_entry_idx_ = i;
+                        resumed_key_val_ = key_val;
+                        return true;  // Caller must consume batch before continuing
+                    }
+
                     const auto& bucket_key = bucket.key_values[i][right_key_col_idx_];
                     if (bucket_key == key_val) {
                         // Match found - emit row
@@ -740,7 +797,7 @@ class VectorizedHashJoinOperator : public VectorizedOperator {
                         if (join_type_ == JoinType::Left) {
                             left_matched_in_batch_[left_row_idx_] = true;
                         }
-                        break;  // Each left row matches at most one right row
+                        // Continue scanning bucket for all matching right rows
                     }
                 }
 

tests/vectorized_operator_tests.cpp

@@ -1119,8 +1119,7 @@ TEST_F(VectorizedGroupByTests, VectorizedHashJoinEmptyLeft) {
 
 TEST_F(VectorizedGroupByTests, VectorizedHashJoinMultipleMatches) {
     // Test when right has duplicate keys: id=1 appears twice
-    // Current implementation limitation: each left row matches at most ONE right row
-    // This is a known issue - proper hash join should match ALL right rows with same key
+    // Each left row should match ALL right rows with the same key
     Schema left_schema;
     left_schema.add_column("id", common::ValueType::TYPE_INT64);
 
@@ -1164,13 +1163,17 @@ TEST_F(VectorizedGroupByTests, VectorizedHashJoinMultipleMatches) {
         result->clear();
     }
 
-    // NOTE: Current implementation uses "break" after first match per left row
-    // So each left row only matches ONE right row, even if duplicates exist
-    // This test documents the current behavior; proper hash join would return 3
-    EXPECT_EQ(left_ids.size(), 2);  // left_id=1 matches once, left_id=2 matches once
-    EXPECT_EQ(right_ids.size(), 2);
+    // INNER: left_id=1 matches 2 right rows, left_id=2 matches 1 right row = 3 total
+    EXPECT_EQ(left_ids.size(), 3);
+    EXPECT_EQ(right_ids.size(), 3);
     EXPECT_EQ(left_ids[0], 1);
-    EXPECT_EQ(left_ids[1], 2);
+    EXPECT_EQ(left_ids[1], 1);  // Second match for left id=1
+    EXPECT_EQ(left_ids[2], 2);
+    // Right-side: two rows with id=1 (matches for left_id=1), then one row with id=2 (match for
+    // left_id=2)
+    EXPECT_EQ(right_ids[0], 1);
+    EXPECT_EQ(right_ids[1], 1);  // Second right row with id=1
+    EXPECT_EQ(right_ids[2], 2);
 }
 
 TEST_F(VectorizedGroupByTests, VectorizedHashJoinLeftNullKeys) {