| 1552 | } |
| 1553 | |
| 1554 | Status Sorter::CreateMerger(int num_runs, bool external) { |
| 1555 | std::deque<impala::Sorter::Run *>* runs_to_merge; |
| 1556 | |
| 1557 | if (external) { |
| 1558 | DCHECK_GE(sorted_runs_.size(), 2); |
| 1559 | runs_to_merge = &sorted_runs_; |
| 1560 | } else { |
| 1561 | runs_to_merge = &sorted_inmem_runs_; |
| 1562 | } |
| 1563 | DCHECK_GE(num_runs, 2); |
| 1564 | // Clean up the runs from the previous merge. |
| 1565 | Run::CleanupRuns(&merging_runs_); |
| 1566 | |
| 1567 | // TODO: 'deep_copy_input' is set to true, which forces the merger to copy all rows |
| 1568 | // from the runs being merged. This is unnecessary overhead that is not required if we |
| 1569 | // correctly transfer resources. |
| 1570 | merger_.reset( |
| 1571 | new SortedRunMerger(*compare_less_than_, output_row_desc_, profile_, external, |
| 1572 | codegend_heapify_helper_fn_)); |
| 1573 | |
| 1574 | vector<function<Status (RowBatch**)>> merge_runs; |
| 1575 | merge_runs.reserve(num_runs); |
| 1576 | for (int i = 0; i < num_runs; ++i) { |
| 1577 | Run* run = runs_to_merge->front(); |
| 1578 | RETURN_IF_ERROR(run->PrepareRead()); |
| 1579 | |
| 1580 | // Run::GetNextBatch() is used by the merger to retrieve a batch of rows to merge |
| 1581 | // from this run. |
| 1582 | merge_runs.emplace_back(bind<Status>(mem_fn(&Run::GetNextBatch), run, _1)); |
| 1583 | runs_to_merge->pop_front(); |
| 1584 | merging_runs_.push_back(run); |
| 1585 | } |
| 1586 | RETURN_IF_ERROR(merger_->Prepare(merge_runs)); |
| 1587 | if (external) { |
| 1588 | num_merges_counter_->Add(1); |
| 1589 | } |
| 1590 | return Status::OK(); |
| 1591 | } |
| 1592 | |
| 1593 | Status Sorter::ExecuteIntermediateMerge(Sorter::Run* merged_run) { |
| 1594 | RowBatch intermediate_merge_batch( |