Method Open

be/src/exec/grouping-aggregator.cc:212–265

}

Status GroupingAggregator::Open(RuntimeState* state) {
  RETURN_IF_ERROR(Aggregator::Open(state));

  // Claim reservation after the child has been opened to reduce the peak reservation
  // requirement.
  if (!buffer_pool_client()->is_registered()) {
    DCHECK_GE(resource_profile_.min_reservation, MinReservation());
    RETURN_IF_ERROR(reservation_manager_.ClaimBufferReservation(state));
  }

  // Init the SubReservations for non-streaming instances. Open() may be called many times
  // if this is in a subplan. Only init them once.
  if (!is_streaming_preagg_ && large_write_page_reservation_.is_closed()) {
    DCHECK(large_read_page_reservation_.is_closed());
    large_write_page_reservation_.Init(buffer_pool_client());
    large_read_page_reservation_.Init(buffer_pool_client());
    // The min reservation only guarantees reading one large page and writing one large
    // page at the same time. Save the extra reservation so we can restore them when we
    // actually need them, to avoid accidentally occupy them for other purposes.
    int64_t extra_reservation = resource_profile_.max_row_buffer_size
        - resource_profile_.spillable_buffer_size;
    if (extra_reservation > 0) {
      DCHECK_GT(buffer_pool_client()->GetUnusedReservation(), extra_reservation * 2)
        << buffer_pool_client()->DebugString() << "\n" << resource_profile_;
      SaveLargeReadPageReservation();
      SaveLargeWritePageReservation();
    }
  }

  DCHECK(ht_ctx_.get() != nullptr);
  RETURN_IF_ERROR(ht_ctx_->Open(state));

  if (ht_allocator_ == nullptr) {
    // Allocate 'serialize_stream_' and 'ht_allocator_' on the first Open() call.
    ht_allocator_.reset(new Suballocator(ExecEnv::GetInstance()->buffer_pool(),
        buffer_pool_client(), resource_profile_.spillable_buffer_size));

    if (!is_streaming_preagg_ && needs_serialize_) {
      serialize_stream_.reset(new BufferedTupleStream(state, &intermediate_row_desc_,
          buffer_pool_client(), resource_profile_.spillable_buffer_size,
          resource_profile_.max_row_buffer_size));
      RETURN_IF_ERROR(serialize_stream_->Init(exec_node_->label(), false));
      bool got_buffer;
      // Reserve the memory for 'serialize_stream_' so we don't need to scrounge up
      // another buffer during spilling.
      RETURN_IF_ERROR(serialize_stream_->PrepareForWrite(&got_buffer));
      DCHECK(got_buffer) << "Accounted in min reservation"
                         << buffer_pool_client()->DebugString();
      DCHECK(serialize_stream_->has_write_iterator());
    }
  }
  RETURN_IF_ERROR(CreateHashPartitions(0));
  return Status::OK();
}

Status GroupingAggregator::GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos) {
  RETURN_IF_ERROR(QueryMaintenance(state));