style: automated clang-format fixes

Author: poyrazK

include/common/cluster_manager.hpp

@@ -375,7 +375,7 @@ class ClusterManager {
      * @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) {
+                            std::vector<executor::Tuple> rows) {
         const std::scoped_lock<std::mutex> lock(mutex_);
         unmatched_rows_[context_id][table_name] = std::move(rows);
     }

include/network/rpc_message.hpp

@@ -36,8 +36,8 @@ enum class RpcType : uint8_t {
     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
+    UnmatchedRowsPush = 14,    // Coordinator sends unmatched rows for NULL-padding
+    FetchUnmatchedRows = 15,   // Coordinator fetches stored unmatched rows from data node
     Error = 255
 };
 
@@ -576,9 +576,9 @@ struct BloomFilterBitsArgs {
 struct UnmatchedRowsReportArgs {
     std::string context_id;
     std::string right_table;
-    std::string join_key_col;  // Which column was the join key
+    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
+    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;
@@ -669,7 +669,8 @@ struct UnmatchedRowsPushArgs {
         }
 
         for (uint32_t i = 0; i < count; ++i) {
-            args.unmatched_rows.push_back(Serializer::deserialize_tuple(in.data(), offset, in.size()));
+            args.unmatched_rows.push_back(
+                Serializer::deserialize_tuple(in.data(), offset, in.size()));
         }
         return args;
     }

src/distributed/distributed_executor.cpp

@@ -202,7 +202,9 @@ QueryResult DistributedExecutor::execute(const parser::Statement& stmt,
                     join.type != parser::SelectStatement::JoinType::Right &&
                     join.type != parser::SelectStatement::JoinType::Full) {
                     QueryResult res;
-                    res.set_error("Distributed Shuffle Join only supports INNER, LEFT, RIGHT, and FULL joins");
+                    res.set_error(
+                        "Distributed Shuffle Join only supports INNER, LEFT, RIGHT, and FULL "
+                        "joins");
                     return res;
                 }
 
@@ -653,33 +655,31 @@ QueryResult DistributedExecutor::execute(const parser::Statement& stmt,
         std::vector<std::future<std::pair<bool, network::UnmatchedRowsReportArgs>>> report_futures;
 
         for (const auto& node : data_nodes) {
-            report_futures.push_back(std::async(std::launch::async, [node, context_id,
-                                                                    outer_join_right_table,
-                                                                    outer_join_right_key,
-                                                                    matched_keys,
-                                                                    left_column_count]() {
-                network::RpcClient client(node.address, node.cluster_port);
-                network::UnmatchedRowsReportArgs reply;
-                reply.context_id = context_id;
-                if (client.connect()) {
-                    network::UnmatchedRowsReportArgs report_args;
-                    report_args.context_id = context_id;
-                    report_args.right_table = outer_join_right_table;
-                    report_args.join_key_col = outer_join_right_key;
-                    // Attach matched keys so node knows what was matched
-                    report_args.unmatched_keys = matched_keys;
-                    // Attach left column count for NULL-padding
-                    report_args.left_column_count = left_column_count;
+            report_futures.push_back(std::async(
+                std::launch::async, [node, context_id, outer_join_right_table, outer_join_right_key,
+                                     matched_keys, left_column_count]() {
+                    network::RpcClient client(node.address, node.cluster_port);
+                    network::UnmatchedRowsReportArgs reply;
+                    reply.context_id = context_id;
+                    if (client.connect()) {
+                        network::UnmatchedRowsReportArgs report_args;
+                        report_args.context_id = context_id;
+                        report_args.right_table = outer_join_right_table;
+                        report_args.join_key_col = outer_join_right_key;
+                        // Attach matched keys so node knows what was matched
+                        report_args.unmatched_keys = matched_keys;
+                        // Attach left column count for NULL-padding
+                        report_args.left_column_count = left_column_count;
 
-                    std::vector<uint8_t> resp;
-                    if (client.call(network::RpcType::UnmatchedRowsReport,
-                                    report_args.serialize(), resp)) {
-                        reply = network::UnmatchedRowsReportArgs::deserialize(resp);
-                        return std::make_pair(true, reply);
+                        std::vector<uint8_t> resp;
+                        if (client.call(network::RpcType::UnmatchedRowsReport,
+                                        report_args.serialize(), resp)) {
+                            reply = network::UnmatchedRowsReportArgs::deserialize(resp);
+                            return std::make_pair(true, reply);
+                        }
                     }
-                }
-                return std::make_pair(false, reply);
-            }));
+                    return std::make_pair(false, reply);
+                }));
         }
 
         // Wait for all report futures to complete
@@ -691,25 +691,25 @@ QueryResult DistributedExecutor::execute(const parser::Statement& stmt,
         std::vector<std::future<std::pair<bool, std::vector<executor::Tuple>>>> fetch_futures;
 
         for (const auto& node : data_nodes) {
-            fetch_futures.push_back(std::async(std::launch::async, [node, context_id,
-                                                                    outer_join_right_table]() {
-                network::RpcClient client(node.address, node.cluster_port);
-                std::vector<executor::Tuple> rows;
-                if (client.connect()) {
-                    network::FetchUnmatchedRowsArgs fetch_args;
-                    fetch_args.context_id = context_id;
-                    fetch_args.table_name = outer_join_right_table;
+            fetch_futures.push_back(
+                std::async(std::launch::async, [node, context_id, outer_join_right_table]() {
+                    network::RpcClient client(node.address, node.cluster_port);
+                    std::vector<executor::Tuple> rows;
+                    if (client.connect()) {
+                        network::FetchUnmatchedRowsArgs fetch_args;
+                        fetch_args.context_id = context_id;
+                        fetch_args.table_name = outer_join_right_table;
 
-                    std::vector<uint8_t> resp;
-                    if (client.call(network::RpcType::FetchUnmatchedRows,
-                                    fetch_args.serialize(), resp)) {
-                        auto reply = network::UnmatchedRowsPushArgs::deserialize(resp);
-                        rows = std::move(reply.unmatched_rows);
-                        return std::make_pair(true, std::move(rows));
+                        std::vector<uint8_t> resp;
+                        if (client.call(network::RpcType::FetchUnmatchedRows,
+                                        fetch_args.serialize(), resp)) {
+                            auto reply = network::UnmatchedRowsPushArgs::deserialize(resp);
+                            rows = std::move(reply.unmatched_rows);
+                            return std::make_pair(true, std::move(rows));
+                        }
                     }
-                }
-                return std::make_pair(false, std::move(rows));
-            }));
+                    return std::make_pair(false, std::move(rows));
+                }));
         }
 
         // Aggregate all unmatched rows from all nodes

src/main.cpp

@@ -745,28 +745,35 @@ int main(int argc, char* argv[]) {
                                             const auto& key_val = t_meta.tuple.get(key_idx);
                                             std::string key_str = key_val.to_string();
                                             // Only include if NOT in matched keys
-                                            if (matched_keys_set.find(key_str) == matched_keys_set.end()) {
+                                            if (matched_keys_set.find(key_str) ==
+                                                matched_keys_set.end()) {
                                                 reply.unmatched_keys.push_back(key_str);
-                                                // Pad with NULLs for left columns and append right row
+                                                // Pad with NULLs for left columns and append right
+                                                // row
                                                 std::vector<cloudsql::common::Value> padded_values;
-                                                padded_values.reserve(args.left_column_count + t_meta.tuple.size());
+                                                padded_values.reserve(args.left_column_count +
+                                                                      t_meta.tuple.size());
                                                 // Prepend NULLs for left table columns
-                                                for (uint32_t i = 0; i < args.left_column_count; ++i) {
-                                                    padded_values.push_back(cloudsql::common::Value::make_null());
+                                                for (uint32_t i = 0; i < args.left_column_count;
+                                                     ++i) {
+                                                    padded_values.push_back(
+                                                        cloudsql::common::Value::make_null());
                                                 }
                                                 // Append right table column values
                                                 for (size_t j = 0; j < t_meta.tuple.size(); ++j) {
                                                     padded_values.push_back(t_meta.tuple.get(j));
                                                 }
-                                                unmatched_tuples.emplace_back(std::move(padded_values));
+                                                unmatched_tuples.emplace_back(
+                                                    std::move(padded_values));
                                             }
                                         }
                                     }
-                                    // Store properly padded tuples in ClusterManager for coordinator to collect
+                                    // Store properly padded tuples in ClusterManager for
+                                    // coordinator to collect
                                     if (cluster_manager != nullptr && !unmatched_tuples.empty()) {
-                                        cluster_manager->set_unmatched_rows(args.context_id,
-                                                                          args.right_table,
-                                                                          std::move(unmatched_tuples));
+                                        cluster_manager->set_unmatched_rows(
+                                            args.context_id, args.right_table,
+                                            std::move(unmatched_tuples));
                                     }
                                 }
                             }
@@ -786,7 +793,8 @@ int main(int argc, char* argv[]) {
                     });
 
                 // Handler for receiving unmatched rows from coordinator for NULL-padding emission
-                // Coordinator broadcasts unmatched right rows to all nodes for final result assembly
+                // Coordinator broadcasts unmatched right rows to all nodes for final result
+                // assembly
                 rpc_server->set_handler(
                     cloudsql::network::RpcType::UnmatchedRowsPush,
                     [&](const cloudsql::network::RpcHeader& h, const std::vector<uint8_t>& p,
@@ -796,8 +804,8 @@ int main(int argc, char* argv[]) {
 
                         if (cluster_manager != nullptr && !args.unmatched_rows.empty()) {
                             cluster_manager->buffer_shuffle_data(args.context_id,
-                                                                    "_unmatched_right_rows",
-                                                                    std::move(args.unmatched_rows));
+                                                                 "_unmatched_right_rows",
+                                                                 std::move(args.unmatched_rows));
                         }
 
                         cloudsql::network::QueryResultsReply reply;

tests/distributed_tests.cpp

@@ -577,11 +577,12 @@ TEST(DistributedExecutorTests, BloomFilterSkipForOuterJoin) {
     DistributedExecutor exec(*catalog, cm);
 
     // Execute RIGHT join - bloom filter should NOT be sent
-    auto lexer = std::make_unique<Lexer>(
-        "SELECT * FROM table1 RIGHT JOIN table2 ON table1.id = table2.id");
+    auto lexer =
+        std::make_unique<Lexer>("SELECT * FROM table1 RIGHT JOIN table2 ON table1.id = table2.id");
     Parser parser(std::move(lexer));
     auto stmt = parser.parse_statement();
-    auto res = exec.execute(*stmt, "SELECT * FROM table1 RIGHT JOIN table2 ON table1.id = table2.id");
+    auto res =
+        exec.execute(*stmt, "SELECT * FROM table1 RIGHT JOIN table2 ON table1.id = table2.id");
 
     // For RIGHT join, bloom filter is skipped (should be 0)
     // Even though we returned valid bloom bits, the coordinator should not push for outer joins
@@ -615,15 +616,18 @@ TEST(DistributedExecutorTests, Phase3SkippedForRightJoin) {
         Schema schema;
         schema.add_column("id", common::ValueType::TYPE_INT64);
         schema.add_column("val", common::ValueType::TYPE_INT64);
-        schema.add_column("id", common::ValueType::TYPE_INT64);  // table2.id (ambiguous but matches)
+        schema.add_column("id",
+                          common::ValueType::TYPE_INT64);  // table2.id (ambiguous but matches)
         schema.add_column("val", common::ValueType::TYPE_INT64);  // table2.val
         reply.schema = schema;
 
         // Return some matched rows (e.g., rows where table2.id = 1 and 2 matched)
         // Format: {table1.id, table1.val, table2.id, table2.val}
         reply.rows = {
-            Tuple{Value::make_int64(100), Value::make_int64(10), Value::make_int64(1), Value::make_int64(100)},
-            Tuple{Value::make_int64(200), Value::make_int64(20), Value::make_int64(2), Value::make_int64(200)},
+            Tuple{Value::make_int64(100), Value::make_int64(10), Value::make_int64(1),
+                  Value::make_int64(100)},
+            Tuple{Value::make_int64(200), Value::make_int64(20), Value::make_int64(2),
+                  Value::make_int64(200)},
         };
 
         auto resp_p = reply.serialize();
@@ -665,8 +669,8 @@ TEST(DistributedExecutorTests, Phase3SkippedForRightJoin) {
 
         UnmatchedRowsPushArgs reply;
         reply.context_id = args.context_id;
-        reply.unmatched_rows = {
-            Tuple{Value::make_null(), Value::make_null(), Value::make_int64(3), Value::make_int64(30)}};
+        reply.unmatched_rows = {Tuple{Value::make_null(), Value::make_null(), Value::make_int64(3),
+                                      Value::make_int64(30)}};
 
         auto resp_p = reply.serialize();
         RpcHeader resp_h;
@@ -743,16 +747,17 @@ TEST(DistributedExecutorTests, Phase3SkippedForRightJoin) {
     DistributedExecutor exec(*catalog, cm);
 
     // Execute RIGHT join
-    auto lexer = std::make_unique<Lexer>(
-        "SELECT * FROM table1 RIGHT JOIN table2 ON table1.id = table2.id");
+    auto lexer =
+        std::make_unique<Lexer>("SELECT * FROM table1 RIGHT JOIN table2 ON table1.id = table2.id");
     Parser parser(std::move(lexer));
     auto stmt = parser.parse_statement();
-    auto res = exec.execute(*stmt, "SELECT * FROM table1 RIGHT JOIN table2 ON table1.id = table2.id");
+    auto res =
+        exec.execute(*stmt, "SELECT * FROM table1 RIGHT JOIN table2 ON table1.id = table2.id");
 
     // Verify Phase 3-4 RPCs were NOT called for RIGHT JOIN
     // (local executor handles unmatched right rows correctly)
     EXPECT_EQ(unmatched_report_calls.load(), 0);  // NOT called for RIGHT JOIN
-    EXPECT_EQ(fetch_unmatched_calls.load(), 0);    // NOT called for RIGHT JOIN
+    EXPECT_EQ(fetch_unmatched_calls.load(), 0);   // NOT called for RIGHT JOIN
     EXPECT_TRUE(res.success());
 
     node1.stop();