| 212 | } |
| 213 | |
| 214 | static std::vector<RowVectorPtr> getOutputFromSourceMerger( |
| 215 | SourceMerger* sourceMerger) { |
| 216 | std::vector<ContinueFuture> sourceBlockingFutures; |
| 217 | std::vector<RowVectorPtr> results; |
| 218 | for (;;) { |
| 219 | sourceMerger->isBlocked(sourceBlockingFutures); |
| 220 | if (!sourceBlockingFutures.empty()) { |
| 221 | auto future = std::move(sourceBlockingFutures.back()); |
| 222 | sourceBlockingFutures.pop_back(); |
| 223 | future.wait(); |
| 224 | continue; |
| 225 | } |
| 226 | |
| 227 | bool atEnd = false; |
| 228 | auto output = sourceMerger->getOutput(sourceBlockingFutures, atEnd); |
| 229 | if (output != nullptr) { |
| 230 | results.emplace_back(std::move(output)); |
| 231 | } |
| 232 | if (atEnd) { |
| 233 | break; |
| 234 | } |
| 235 | } |
| 236 | return results; |
| 237 | } |
| 238 | |
| 239 | std::shared_ptr<SpillMerger> createSpillMerger( |
| 240 | std::vector<std::vector<std::unique_ptr<SpillReadFile>>> |