Method EvictPartitions

be/src/exec/topn-node.cc:536–576

}

Status TopNNode::EvictPartitions(RuntimeState* state, bool evict_final) {
  DCHECK(is_partitioned());
  vector<unique_ptr<Heap>> heaps_to_evict;
  if (evict_final) {
    // Move all the partitions to 'sorter_' in preparation for the final sort. Partitions
    // are evicted in the order of the partition key to reduce the amount of shuffling
    // that the final sort will do to rearrange partitions.
    for (auto& entry : partition_heaps_) {
      heaps_to_evict.push_back(move(entry.second));
    }
    partition_heaps_.clear();
  } else {
    heaps_to_evict = SelectPartitionsToEvict();
  }
  // Only count heap eviction if they are as a result of memory pressure.
  if (!evict_final) COUNTER_ADD(in_mem_heap_evicted_counter_, heaps_to_evict.size());

  RowBatch batch(row_desc(), state->batch_size(), mem_tracker());
  for (auto& heap : heaps_to_evict) {
    DCHECK(heap->DCheckConsistency());
    // Extract partition entries from the heap in sorted order to reduce amount of sorting
    // required in final sort. This sorting is not required for correctness since
    // 'sorter_' will do a full sort later.
    heap->PrepareForOutput(*this, &sorted_top_n_);
    for (int64_t i = 0; i < sorted_top_n_.size(); ++i) {
      TupleRow* row = batch.GetRow(batch.AddRow());
      row->SetTuple(0, sorted_top_n_[i]);
      batch.CommitLastRow();
      if (batch.AtCapacity() || i == sorted_top_n_.size() - 1) {
        RETURN_IF_ERROR(sorter_->AddBatch(&batch));
        batch.Reset();
      }
    }
    sorted_top_n_.clear();
  }
  heaps_to_evict.clear();

  // ReclaimTuplePool() can now reclaim memory that is not used by in-memory partitions.
  RETURN_IF_ERROR(ReclaimTuplePool(state));
  return Status::OK();
}

vector<unique_ptr<TopNNode::Heap>> TopNNode::SelectPartitionsToEvict() {
  // Evict a subset of heaps to free enough memory to continue.