@@ -202,7 +202,9 @@ QueryResult DistributedExecutor::execute(const parser::Statement& stmt,
202202 join.type != parser::SelectStatement::JoinType::Right &&
203203 join.type != parser::SelectStatement::JoinType::Full) {
204204 QueryResult res;
205- res.set_error (" Distributed Shuffle Join only supports INNER, LEFT, RIGHT, and FULL joins" );
205+ res.set_error (
206+ " Distributed Shuffle Join only supports INNER, LEFT, RIGHT, and FULL "
207+ " joins" );
206208 return res;
207209 }
208210
@@ -654,33 +656,31 @@ QueryResult DistributedExecutor::execute(const parser::Statement& stmt,
654656 std::vector<std::future<std::pair<bool , network::UnmatchedRowsReportArgs>>> report_futures;
655657
656658 for (const auto & node : data_nodes) {
657- report_futures.push_back (std::async (std::launch::async, [node, context_id,
658- outer_join_right_table,
659- outer_join_right_key,
660- matched_keys,
661- left_column_count]() {
662- network::RpcClient client (node.address , node.cluster_port );
663- network::UnmatchedRowsReportArgs reply;
664- reply.context_id = context_id;
665- if (client.connect ()) {
666- network::UnmatchedRowsReportArgs report_args;
667- report_args.context_id = context_id;
668- report_args.right_table = outer_join_right_table;
669- report_args.join_key_col = outer_join_right_key;
670- // Attach matched keys so node knows what was matched
671- report_args.unmatched_keys = matched_keys;
672- // Attach left column count for NULL-padding
673- report_args.left_column_count = left_column_count;
659+ report_futures.push_back (std::async (
660+ std::launch::async, [node, context_id, outer_join_right_table, outer_join_right_key,
661+ matched_keys, left_column_count]() {
662+ network::RpcClient client (node.address , node.cluster_port );
663+ network::UnmatchedRowsReportArgs reply;
664+ reply.context_id = context_id;
665+ if (client.connect ()) {
666+ network::UnmatchedRowsReportArgs report_args;
667+ report_args.context_id = context_id;
668+ report_args.right_table = outer_join_right_table;
669+ report_args.join_key_col = outer_join_right_key;
670+ // Attach matched keys so node knows what was matched
671+ report_args.unmatched_keys = matched_keys;
672+ // Attach left column count for NULL-padding
673+ report_args.left_column_count = left_column_count;
674674
675- std::vector<uint8_t > resp;
676- if (client.call (network::RpcType::UnmatchedRowsReport,
677- report_args.serialize (), resp)) {
678- reply = network::UnmatchedRowsReportArgs::deserialize (resp);
679- return std::make_pair (true , reply);
675+ std::vector<uint8_t > resp;
676+ if (client.call (network::RpcType::UnmatchedRowsReport,
677+ report_args.serialize (), resp)) {
678+ reply = network::UnmatchedRowsReportArgs::deserialize (resp);
679+ return std::make_pair (true , reply);
680+ }
680681 }
681- }
682- return std::make_pair (false , reply);
683- }));
682+ return std::make_pair (false , reply);
683+ }));
684684 }
685685
686686 // Wait for all report futures to complete
@@ -692,25 +692,25 @@ QueryResult DistributedExecutor::execute(const parser::Statement& stmt,
692692 std::vector<std::future<std::pair<bool , std::vector<executor::Tuple>>>> fetch_futures;
693693
694694 for (const auto & node : data_nodes) {
695- fetch_futures.push_back (std::async (std::launch::async, [node, context_id,
696- outer_join_right_table]() {
697- network::RpcClient client (node.address , node.cluster_port );
698- std::vector<executor::Tuple> rows;
699- if (client.connect ()) {
700- network::FetchUnmatchedRowsArgs fetch_args;
701- fetch_args.context_id = context_id;
702- fetch_args.table_name = outer_join_right_table;
695+ fetch_futures.push_back (
696+ std::async (std::launch::async, [node, context_id, outer_join_right_table]() {
697+ network::RpcClient client (node.address , node.cluster_port );
698+ std::vector<executor::Tuple> rows;
699+ if (client.connect ()) {
700+ network::FetchUnmatchedRowsArgs fetch_args;
701+ fetch_args.context_id = context_id;
702+ fetch_args.table_name = outer_join_right_table;
703703
704- std::vector<uint8_t > resp;
705- if (client.call (network::RpcType::FetchUnmatchedRows,
706- fetch_args.serialize (), resp)) {
707- auto reply = network::UnmatchedRowsPushArgs::deserialize (resp);
708- rows = std::move (reply.unmatched_rows );
709- return std::make_pair (true , std::move (rows));
704+ std::vector<uint8_t > resp;
705+ if (client.call (network::RpcType::FetchUnmatchedRows,
706+ fetch_args.serialize (), resp)) {
707+ auto reply = network::UnmatchedRowsPushArgs::deserialize (resp);
708+ rows = std::move (reply.unmatched_rows );
709+ return std::make_pair (true , std::move (rows));
710+ }
710711 }
711- }
712- return std::make_pair (false , std::move (rows));
713- }));
712+ return std::make_pair (false , std::move (rows));
713+ }));
714714 }
715715
716716 // Aggregate all unmatched rows from all nodes
0 commit comments