| 133 | } |
| 134 | |
| 135 | bool HashJoinBridge::probeFinished() { |
| 136 | std::vector<ContinuePromise> promises; |
| 137 | bool hasSpillInput = false; |
| 138 | { |
| 139 | std::lock_guard<std::mutex> l(mutex_); |
| 140 | BOLT_CHECK(started_); |
| 141 | BOLT_CHECK(buildResult_.has_value()); |
| 142 | BOLT_CHECK( |
| 143 | !restoringSpillPartitionId_.has_value() && |
| 144 | restoringSpillShards_.empty()); |
| 145 | BOLT_CHECK_GT(numBuilders_, 0); |
| 146 | |
| 147 | // NOTE: we are clearing the hash table as it has been fully processed and |
| 148 | // not needed anymore. We'll wait for the HashBuild operator to build a new |
| 149 | // table from the next spill partition now. |
| 150 | buildResult_.reset(); |
| 151 | |
| 152 | if (!spillPartitionSets_.empty()) { |
| 153 | hasSpillInput = true; |
| 154 | restoringSpillPartitionId_ = spillPartitionSets_.begin()->first; |
| 155 | restoringSpillShards_ = |
| 156 | spillPartitionSets_.begin()->second->split(numBuilders_); |
| 157 | BOLT_CHECK_EQ(restoringSpillShards_.size(), numBuilders_); |
| 158 | spillPartitionSets_.erase(spillPartitionSets_.begin()); |
| 159 | promises = std::move(promises_); |
| 160 | } else { |
| 161 | BOLT_CHECK(promises_.empty()); |
| 162 | } |
| 163 | } |
| 164 | notify(std::move(promises)); |
| 165 | return hasSpillInput; |
| 166 | } |
| 167 | |
| 168 | std::optional<HashJoinBridge::SpillInput> HashJoinBridge::spillInputOrFuture( |
| 169 | ContinueFuture* future) { |