| 1524 | } |
| 1525 | |
| 1526 | Status Sorter::MergeIntermediateRuns() { |
| 1527 | DCHECK_GE(sorted_runs_.size(), 2); |
| 1528 | DCHECK_GT(sorted_runs_.back()->fixed_len_size(), 0); |
| 1529 | |
| 1530 | // Attempt to allocate more memory before doing intermediate merges. This may |
| 1531 | // be possible if other operators have relinquished memory after the sort has built |
| 1532 | // its runs. |
| 1533 | TryToIncreaseMemAllocationForMerge(); |
| 1534 | |
| 1535 | while (true) { |
| 1536 | int num_of_runs_to_merge = GetNumOfRunsForMerge(); |
| 1537 | |
| 1538 | DCHECK(merge_output_run_ == nullptr) << "Should have finished previous merge."; |
| 1539 | RETURN_IF_ERROR(CreateMerger(num_of_runs_to_merge, true)); |
| 1540 | |
| 1541 | // If CreateMerger() consumed all the sorted runs, we have set up the final merge. |
| 1542 | if (sorted_runs_.empty()) return Status::OK(); |
| 1543 | |
| 1544 | merge_output_run_ = run_pool_.Add( |
| 1545 | new Run(this, output_row_desc_->tuple_descriptors()[0], false)); |
| 1546 | RETURN_IF_ERROR(merge_output_run_->Init()); |
| 1547 | RETURN_IF_ERROR(ExecuteIntermediateMerge(merge_output_run_)); |
| 1548 | sorted_runs_.push_back(merge_output_run_); |
| 1549 | merge_output_run_ = nullptr; |
| 1550 | } |
| 1551 | return Status::OK(); |
| 1552 | } |
| 1553 | |
| 1554 | Status Sorter::CreateMerger(int num_runs, bool external) { |
| 1555 | std::deque<impala::Sorter::Run *>* runs_to_merge; |