| 1311 | } |
| 1312 | |
| 1313 | Status Sorter::InputDone() { |
| 1314 | // Sort the tuples in the last run. |
| 1315 | RETURN_IF_ERROR(SortCurrentInputRun()); |
| 1316 | |
| 1317 | if (inmem_run_max_pages_ > 0) { |
| 1318 | sorted_inmem_runs_.push_back(unsorted_run_); |
| 1319 | unsorted_run_ = nullptr; |
| 1320 | if (!HasSpilledRuns()) { |
| 1321 | if (sorted_inmem_runs_.size() == 1) { |
| 1322 | DCHECK(sorted_inmem_runs_.back()->is_pinned()); |
| 1323 | DCHECK(merge_output_run_ == nullptr); |
| 1324 | RETURN_IF_ERROR(sorted_inmem_runs_.back()->PrepareRead()); |
| 1325 | return Status::OK(); |
| 1326 | } |
| 1327 | if (enable_spilling_) { |
| 1328 | DCHECK(merge_output_run_ != nullptr); |
| 1329 | merge_output_run_->CloseAllPages(); |
| 1330 | merge_output_run_ = nullptr; |
| 1331 | } |
| 1332 | // 'merge_output_run_' is not initialized for partial sort, because the output |
| 1333 | // will be read directly from the merger. |
| 1334 | DCHECK(enable_spilling_ || merge_output_run_ == nullptr); |
| 1335 | return CreateMerger(sorted_inmem_runs_.size(), false); |
| 1336 | } |
| 1337 | DCHECK(enable_spilling_); |
| 1338 | |
| 1339 | if (sorted_inmem_runs_.size() == 1) { |
| 1340 | sorted_runs_.push_back(sorted_inmem_runs_.back()); |
| 1341 | sorted_inmem_runs_.clear(); |
| 1342 | DCHECK_GT(sorted_runs_.back()->run_size(), 0); |
| 1343 | RETURN_IF_ERROR(sorted_runs_.back()->UnpinAllPages()); |
| 1344 | } else { |
| 1345 | RETURN_IF_ERROR(MergeInMemoryRuns()); |
| 1346 | } |
| 1347 | DCHECK(sorted_inmem_runs_.empty()); |
| 1348 | // Merge intermediate runs until we have a final merge set-up. |
| 1349 | return MergeIntermediateRuns(); |
| 1350 | } else { |
| 1351 | sorted_runs_.push_back(unsorted_run_); |
| 1352 | } |
| 1353 | |
| 1354 | unsorted_run_ = nullptr; |
| 1355 | |
| 1356 | if (sorted_runs_.size() == 1) { |
| 1357 | // The entire input fit in one run. Read sorted rows in GetNext() directly from the |
| 1358 | // in-memory sorted run. |
| 1359 | DCHECK(sorted_runs_.back()->is_pinned()); |
| 1360 | RETURN_IF_ERROR(sorted_runs_.back()->PrepareRead()); |
| 1361 | return Status::OK(); |
| 1362 | } |
| 1363 | DCHECK(enable_spilling_); |
| 1364 | |
| 1365 | // Unpin the final run to free up memory for the merge. |
| 1366 | // TODO: we could keep it in memory in some circumstances as an optimisation, once |
| 1367 | // we have a buffer pool with more reliable reservations (IMPALA-3200). |
| 1368 | RETURN_IF_ERROR(sorted_runs_.back()->UnpinAllPages()); |
| 1369 | |
| 1370 | // Merge intermediate runs until we have a final merge set-up. |
nothing calls this directly
no test coverage detected