Method AddBatchStreaming

be/src/exec/grouping-aggregator.cc:475–525

}

Status GroupingAggregator::AddBatchStreaming(
    RuntimeState* state, RowBatch* out_batch, RowBatch* child_batch, bool* eos) {
  SCOPED_TIMER(streaming_timer_);
  RETURN_IF_ERROR(QueryMaintenance(state));
  num_input_rows_ += child_batch->num_rows();

  int remaining_capacity[PARTITION_FANOUT];
  bool ht_needs_expansion = false;
  for (int i = 0; i < PARTITION_FANOUT; ++i) {
    HashTable* hash_tbl = GetHashTable(i);
    remaining_capacity[i] = hash_tbl->NumInsertsBeforeResize();
    ht_needs_expansion |= remaining_capacity[i] < child_batch->num_rows();
  }

  // Stop expanding hash tables if we're not reducing the input sufficiently. As our
  // hash tables expand out of each level of cache hierarchy, every hash table lookup
  // will take longer. We also may not be able to expand hash tables because of memory
  // pressure. In this case HashTable::CheckAndResize() will fail. In either case we
  // should always use the remaining space in the hash table to avoid wasting memory.
  if (ht_needs_expansion && ShouldExpandPreaggHashTables()) {
    for (int i = 0; i < PARTITION_FANOUT; ++i) {
      HashTable* ht = GetHashTable(i);
      if (remaining_capacity[i] < child_batch->num_rows()) {
        SCOPED_TIMER(ht_resize_timer_);
        bool resized;
        RETURN_IF_ERROR(
            ht->CheckAndResize(child_batch->num_rows(), ht_ctx_.get(), &resized));
        if (resized) {
          remaining_capacity[i] = ht->NumInsertsBeforeResize();
        }
      }
    }
  }

  TPrefetchMode::type prefetch_mode = state->query_options().prefetch_mode;
  int64_t num_row_out_batch_old = out_batch->num_rows();
  GroupingAggregatorConfig::AddBatchStreamingImplFn fn
      = add_batch_streaming_impl_fn_.load();
  if (fn != nullptr) {
    RETURN_IF_ERROR(fn(this, agg_idx_, needs_serialize_,
        prefetch_mode, child_batch, out_batch, ht_ctx_.get(), remaining_capacity));
  } else {
    RETURN_IF_ERROR(AddBatchStreamingImpl(agg_idx_, needs_serialize_, prefetch_mode,
        child_batch, out_batch, ht_ctx_.get(), remaining_capacity));
  }
  *eos = (streaming_idx_ == 0);
  DCHECK_GE(out_batch->num_rows(), num_row_out_batch_old);
  num_rows_returned_ += out_batch->num_rows() - num_row_out_batch_old;
  COUNTER_SET(num_passthrough_rows_, num_rows_returned_);
  return Status::OK();
}

Status GroupingAggregator::InputDone() {
  return MoveHashPartitions(num_input_rows_);