Method AddIntermediateTuple

be/src/exec/grouping-aggregator-ir.cc:139–173

template <bool AGGREGATED_ROWS>
Status GroupingAggregator::AddIntermediateTuple(Partition* __restrict__ partition,
    TupleRow* __restrict__ row, uint32_t hash, HashTable::Iterator insert_it,
    bool has_more_rows) {
  while (true) {
    DCHECK(partition->aggregated_row_stream->is_pinned());
    Tuple* intermediate_tuple = ConstructIntermediateTuple(partition->agg_fn_evals,
        partition->aggregated_row_stream.get(), &add_batch_status_);

    if (LIKELY(intermediate_tuple != nullptr)) {
      UpdateTuple(
          partition->agg_fn_evals.data(), intermediate_tuple, row, AGGREGATED_ROWS);
      // After copying and initializing the tuple, insert it into the hash table.
      insert_it.SetTuple(intermediate_tuple, hash);
      return Status::OK();
    } else if (!add_batch_status_.ok()) {
      return std::move(add_batch_status_);
    }

    // If we don't need to reserve extra space for the serialize stream, restore them
    // before spilling any partitions. One case is we don't need the serialize stream at
    // all. The other case is this is the last row to add, and 'partition' will be read
    // out and closed after adding this row, so no partitions need to be spilled.
    if ((!needs_serialize_ || !has_more_rows)
        && large_write_page_reservation_.GetReservation() > 0) {
      RestoreLargeWritePageReservation();
      continue;
    }

    // We did not have enough memory to add intermediate_tuple to the stream.
    RETURN_IF_ERROR(SpillPartition(AGGREGATED_ROWS));
    if (partition->is_spilled()) {
      return AppendSpilledRow<AGGREGATED_ROWS>(partition, row);
    }
  }
}

Status GroupingAggregator::AddBatchStreamingImpl(int agg_idx, bool needs_serialize,
    TPrefetchMode::type prefetch_mode, RowBatch* in_batch, RowBatch* out_batch,