feat: support RIGHT/FULL JOIN with bloom filter skip

- Skip bloom filter for RIGHT/FULL joins to prevent false negatives
- Add Phase 3-5 orchestration for FULL JOIN unmatched LEFT rows
- Add NULL-padding for unmatched right rows with left-side NULLs
- Add integration tests for bloom filter skip and Phase 3-5
- Allow LEFT joins in distributed shuffle join

The bloom filter can cause false negatives (rows filtered when they shouldn't be),
which violates outer join semantics. For RIGHT/FULL joins, all right rows must be
sent to ensure correct results.

NOTE: Phase 3-5 is disabled for RIGHT JOIN because the local executor on each
data node already correctly handles unmatched right rows via the hash table scan.
Phase 3-5 is only needed for FULL JOIN to collect unmatched LEFT rows.

Author: poyrazK

include/common/cluster_manager.hpp

@@ -337,6 +337,73 @@ class ClusterManager {
         local_num_hashes_map_.erase(context_id);
     }
 
+    /**
+     * @brief Store local right table rows for outer join processing
+     * Called during Phase 2 shuffle when sending right table rows to other nodes
+     */
+    void set_local_right_rows(const std::string& context_id, const std::string& table_name,
+                              std::vector<executor::Tuple> rows) {
+        const std::scoped_lock<std::mutex> lock(mutex_);
+        local_right_table_rows_[context_id][table_name] = std::move(rows);
+    }
+
+    /**
+     * @brief Get stored local right table rows
+     */
+    [[nodiscard]] std::vector<executor::Tuple> get_local_right_rows(
+        const std::string& context_id, const std::string& table_name) const {
+        const std::scoped_lock<std::mutex> lock(mutex_);
+        auto ctx_it = local_right_table_rows_.find(context_id);
+        if (ctx_it != local_right_table_rows_.end()) {
+            auto table_it = ctx_it->second.find(table_name);
+            if (table_it != ctx_it->second.end()) {
+                return table_it->second;
+            }
+        }
+        return {};
+    }
+
+    /**
+     * @brief Clear local right table rows for a context
+     */
+    void clear_local_right_rows(const std::string& context_id) {
+        const std::scoped_lock<std::mutex> lock(mutex_);
+        local_right_table_rows_.erase(context_id);
+    }
+
+    /**
+     * @brief Store unmatched rows for a context (used by outer join processing)
+     */
+    void set_unmatched_rows(const std::string& context_id, const std::string& table_name,
+                           std::vector<executor::Tuple> rows) {
+        const std::scoped_lock<std::mutex> lock(mutex_);
+        unmatched_rows_[context_id][table_name] = std::move(rows);
+    }
+
+    /**
+     * @brief Get stored unmatched rows for a context
+     */
+    [[nodiscard]] std::vector<executor::Tuple> get_unmatched_rows(
+        const std::string& context_id, const std::string& table_name) const {
+        const std::scoped_lock<std::mutex> lock(mutex_);
+        auto ctx_it = unmatched_rows_.find(context_id);
+        if (ctx_it != unmatched_rows_.end()) {
+            auto table_it = ctx_it->second.find(table_name);
+            if (table_it != ctx_it->second.end()) {
+                return table_it->second;
+            }
+        }
+        return {};
+    }
+
+    /**
+     * @brief Clear unmatched rows for a context
+     */
+    void clear_unmatched_rows(const std::string& context_id) {
+        const std::scoped_lock<std::mutex> lock(mutex_);
+        unmatched_rows_.erase(context_id);
+    }
+
    private:
     /**
      * @brief Stored bloom filter data for a context
@@ -365,6 +432,12 @@ class ClusterManager {
     std::unordered_map<std::string, std::vector<uint8_t>> local_bloom_bits_;
     std::unordered_map<std::string, size_t> local_expected_elements_map_;
     std::unordered_map<std::string, size_t> local_num_hashes_map_;
+    /* context_id -> table_name -> local right table rows for outer join tracking */
+    std::unordered_map<std::string, std::unordered_map<std::string, std::vector<executor::Tuple>>>
+        local_right_table_rows_;
+    /* context_id -> table_name -> unmatched rows for outer join NULL-padding */
+    std::unordered_map<std::string, std::unordered_map<std::string, std::vector<executor::Tuple>>>
+        unmatched_rows_;
     mutable std::mutex mutex_;
 };
 

include/executor/operator.hpp

@@ -358,6 +358,18 @@ class HashJoinOperator : public Operator {
 
     void set_memory_resource(std::pmr::memory_resource* mr) override;
     void set_params(const std::vector<common::Value>* params) override;
+
+    /**
+     * @brief Get unmatched right rows after join execution
+     * @return Vector of tuples - the right-side rows that had no match
+     */
+    [[nodiscard]] std::vector<Tuple> get_unmatched_right_rows() const;
+
+    /**
+     * @brief Get join key values of unmatched right rows
+     * @return Vector of strings - the join key values for unmatched right rows
+     */
+    [[nodiscard]] std::vector<std::string> get_unmatched_right_keys() const;
 };
 
 /**

include/network/rpc_message.hpp

@@ -35,6 +35,9 @@ enum class RpcType : uint8_t {
     ShuffleFragment = 10,
     BloomFilterPush = 11,
     BloomFilterBits = 12,
+    UnmatchedRowsReport = 13,  // Data node reports unmatched right rows for outer join
+    UnmatchedRowsPush = 14,   // Coordinator sends unmatched rows for NULL-padding
+    FetchUnmatchedRows = 15,  // Coordinator fetches stored unmatched rows from data node
     Error = 255
 };
 
@@ -566,6 +569,136 @@ struct BloomFilterBitsArgs {
     }
 };
 
+/**
+ * @brief Arguments for UnmatchedRowsReport RPC
+ * @note Data node reports unmatched right row keys to coordinator after local join
+ */
+struct UnmatchedRowsReportArgs {
+    std::string context_id;
+    std::string right_table;
+    std::string join_key_col;  // Which column was the join key
+    std::vector<std::string> unmatched_keys;  // Key values that had no match
+    uint32_t left_column_count = 0;  // Number of left table columns for NULL-padding
+
+    [[nodiscard]] std::vector<uint8_t> serialize() const {
+        std::vector<uint8_t> out;
+        Serializer::serialize_string(context_id, out);
+        Serializer::serialize_string(right_table, out);
+        Serializer::serialize_string(join_key_col, out);
+
+        // Serialize left column count
+        const uint32_t left_count = left_column_count;
+        const size_t lc_off = out.size();
+        out.resize(lc_off + Serializer::VAL_SIZE_32);
+        std::memcpy(out.data() + lc_off, &left_count, Serializer::VAL_SIZE_32);
+
+        // Serialize unmatched keys count
+        const uint32_t count = static_cast<uint32_t>(unmatched_keys.size());
+        const size_t off = out.size();
+        out.resize(off + Serializer::VAL_SIZE_32);
+        std::memcpy(out.data() + off, &count, Serializer::VAL_SIZE_32);
+
+        // Serialize each key
+        for (const auto& key : unmatched_keys) {
+            Serializer::serialize_string(key, out);
+        }
+        return out;
+    }
+
+    static UnmatchedRowsReportArgs deserialize(const std::vector<uint8_t>& in) {
+        UnmatchedRowsReportArgs args;
+        size_t offset = 0;
+        args.context_id = Serializer::deserialize_string(in.data(), offset, in.size());
+        args.right_table = Serializer::deserialize_string(in.data(), offset, in.size());
+        args.join_key_col = Serializer::deserialize_string(in.data(), offset, in.size());
+
+        // Deserialize left column count
+        if (offset + Serializer::VAL_SIZE_32 <= in.size()) {
+            std::memcpy(&args.left_column_count, in.data() + offset, Serializer::VAL_SIZE_32);
+            offset += Serializer::VAL_SIZE_32;
+        }
+
+        uint32_t count = 0;
+        if (offset + Serializer::VAL_SIZE_32 <= in.size()) {
+            std::memcpy(&count, in.data() + offset, Serializer::VAL_SIZE_32);
+            offset += Serializer::VAL_SIZE_32;
+        }
+
+        for (uint32_t i = 0; i < count; ++i) {
+            args.unmatched_keys.push_back(
+                Serializer::deserialize_string(in.data(), offset, in.size()));
+        }
+        return args;
+    }
+};
+
+/**
+ * @brief Arguments for UnmatchedRowsPush RPC
+ * @note Coordinator sends unmatched rows to data nodes for NULL-padding
+ */
+struct UnmatchedRowsPushArgs {
+    std::string context_id;
+    std::vector<executor::Tuple> unmatched_rows;  // Right rows needing NULL padding
+
+    [[nodiscard]] std::vector<uint8_t> serialize() const {
+        std::vector<uint8_t> out;
+        Serializer::serialize_string(context_id, out);
+
+        // Serialize row count
+        const uint32_t count = static_cast<uint32_t>(unmatched_rows.size());
+        const size_t off = out.size();
+        out.resize(off + Serializer::VAL_SIZE_32);
+        std::memcpy(out.data() + off, &count, Serializer::VAL_SIZE_32);
+
+        // Serialize each row
+        for (const auto& row : unmatched_rows) {
+            Serializer::serialize_tuple(row, out);
+        }
+        return out;
+    }
+
+    static UnmatchedRowsPushArgs deserialize(const std::vector<uint8_t>& in) {
+        UnmatchedRowsPushArgs args;
+        size_t offset = 0;
+        args.context_id = Serializer::deserialize_string(in.data(), offset, in.size());
+
+        uint32_t count = 0;
+        if (offset + Serializer::VAL_SIZE_32 <= in.size()) {
+            std::memcpy(&count, in.data() + offset, Serializer::VAL_SIZE_32);
+            offset += Serializer::VAL_SIZE_32;
+        }
+
+        for (uint32_t i = 0; i < count; ++i) {
+            args.unmatched_rows.push_back(Serializer::deserialize_tuple(in.data(), offset, in.size()));
+        }
+        return args;
+    }
+};
+
+/**
+ * @brief Arguments for FetchUnmatchedRows RPC
+ * @note Coordinator fetches stored unmatched rows from a data node
+ */
+struct FetchUnmatchedRowsArgs {
+    std::string context_id;
+    std::string table_name;
+
+    [[nodiscard]] std::vector<uint8_t> serialize() const {
+        std::vector<uint8_t> out;
+        Serializer::serialize_string(context_id, out);
+        Serializer::serialize_string(table_name, out);
+        return out;
+    }
+
+    static FetchUnmatchedRowsArgs deserialize(const std::vector<uint8_t>& in) {
+        FetchUnmatchedRowsArgs args;
+        size_t offset = 0;
+        args.context_id = Serializer::deserialize_string(in.data(), offset, in.size());
+        args.table_name = Serializer::deserialize_string(in.data(), offset, in.size());
+        return args;
+    }
+};
+
 /**
  * @brief Arguments for TxnPrepare/Commit/Abort RPC
  */