Method ReclaimTuplePool

be/src/exec/topn-node.cc:685–719

}

Status TopNNode::ReclaimTuplePool(RuntimeState* state) {
  COUNTER_ADD(tuple_pool_reclaim_counter_, 1);
  unique_ptr<MemPool> temp_pool(new MemPool(mem_tracker()));

  if (is_partitioned()) {
    vector<unique_ptr<Heap>> rematerialized_heaps;
    for (auto& entry : partition_heaps_) {
      RETURN_IF_ERROR(entry.second->RematerializeTuples(this, state, temp_pool.get()));
      DCHECK(entry.second->DCheckConsistency());
    }
    // The second loop is needed for IMPALA-11631. We only move heaps from partition_heap_
    // to rematerialized_heaps once all have been rematerialized. Otherwise, in case of
    // an error, we may call Close() on a nullptr or leak the memory by not explicitly
    // calling Close() on the heap pointer. Maybe better to add Close() in the Heap
    // destructor later.
    for (auto& entry : partition_heaps_) {
      // The key references memory in 'tuple_pool_'. Replace it with a rematerialized
      // tuple.
      rematerialized_heaps.push_back(move(entry.second));
    }
    partition_heaps_.clear();
    for (auto& heap_ptr : rematerialized_heaps) {
      const Tuple* key_tuple = heap_ptr->top();
      partition_heaps_.emplace(key_tuple, move(heap_ptr));
    }
  } else {
    RETURN_IF_ERROR(heap_->RematerializeTuples(this, state, temp_pool.get()));
    DCHECK(heap_->DCheckConsistency());
  }
  rows_to_reclaim_ = 0;
  RETURN_IF_ERROR(InitTmpTuple(state, temp_pool.get()));
  tuple_pool_->FreeAll();
  tuple_pool_ = move(temp_pool);
  return Status::OK();
}

Status TopNNode::InitTmpTuple(RuntimeState* state, MemPool* pool) {
  tmp_tuple_ = reinterpret_cast<Tuple*>(pool->TryAllocate(