feat: support RIGHT/FULL JOIN with bloom filter skip and Phase 3-5

- Skip bloom filter for RIGHT/FULL joins to prevent false negatives
- Add Phase 3-5 orchestration to collect and redistribute unmatched rows
- Add NULL-padding for unmatched right rows with left-side NULLs
- Add integration tests for bloom filter skip and Phase 3-5 RPC calls
- Allow LEFT joins in distributed shuffle join (regression from previous commit)

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. Phase 3-5 collects unmatched rows after the join
and adds them back with proper NULL-padding.

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
  */