| 1085 | } |
| 1086 | |
| 1087 | bool HashBuild::finishHashBuild() { |
| 1088 | checkRunning(); |
| 1089 | |
| 1090 | // Release the unused memory reservation before building the merged join |
| 1091 | // table. |
| 1092 | pool()->release(); |
| 1093 | |
| 1094 | std::vector<ContinuePromise> promises; |
| 1095 | std::vector<std::shared_ptr<Driver>> peers; |
| 1096 | // The last Driver to hit HashBuild::finish gathers the data from |
| 1097 | // all build Drivers and hands it over to the probe side. At this |
| 1098 | // point all build Drivers are continued and will free their |
| 1099 | // state. allPeersFinished is true only for the last Driver of the |
| 1100 | // build pipeline. |
| 1101 | if (!operatorCtx_->task()->allPeersFinished( |
| 1102 | planNodeId(), operatorCtx_->driver(), &future_, promises, peers)) { |
| 1103 | setState(State::kWaitForBuild); |
| 1104 | return false; |
| 1105 | } |
| 1106 | |
| 1107 | BOLT_TEST_ADJUST("bytedance::bolt::exec::HashBuild::finishHashBuild", this); |
| 1108 | |
| 1109 | auto promisesGuard = folly::makeGuard([&]() { |
| 1110 | // Realize the promises so that the other Drivers (which were not |
| 1111 | // the last to finish) can continue from the barrier and finish. |
| 1112 | peers.clear(); |
| 1113 | for (auto& promise : promises) { |
| 1114 | promise.setValue(); |
| 1115 | } |
| 1116 | }); |
| 1117 | |
| 1118 | if (joinHasNullKeys_ && isAntiJoin(joinType_) && nullAware_ && |
| 1119 | !joinNode_->filter()) { |
| 1120 | joinBridge_->setAntiJoinHasNullKeys(); |
| 1121 | return true; |
| 1122 | } |
| 1123 | |
| 1124 | std::vector<HashBuild*> otherBuilds; |
| 1125 | otherBuilds.reserve(peers.size()); |
| 1126 | uint64_t numRows = table_->rows()->numRows(); |
| 1127 | for (auto& peer : peers) { |
| 1128 | auto op = peer->findOperator(planNodeId()); |
| 1129 | HashBuild* build = dynamic_cast<HashBuild*>(op); |
| 1130 | BOLT_CHECK_NOT_NULL(build); |
| 1131 | if (build->joinHasNullKeys_) { |
| 1132 | joinHasNullKeys_ = true; |
| 1133 | if (isAntiJoin(joinType_) && nullAware_ && !joinNode_->filter()) { |
| 1134 | joinBridge_->setAntiJoinHasNullKeys(); |
| 1135 | return true; |
| 1136 | } |
| 1137 | } |
| 1138 | { |
| 1139 | std::lock_guard<std::mutex> l(build->intermediateStateMutex_); |
| 1140 | BOLT_CHECK( |
| 1141 | !build->intermediateStateCleared_, |
| 1142 | "Intermediate state for a peer is empty. It might have been " |
| 1143 | "already closed."); |
| 1144 | numRows += build->table_->rows()->numRows(); |
nothing calls this directly
no test coverage detected