Method ProcessRow

be/src/exec/grouping-aggregator-ir.cc:93–136

template <bool AGGREGATED_ROWS>
Status GroupingAggregator::ProcessRow(
    TupleRow* __restrict__ row, HashTableCtx* __restrict__ ht_ctx, bool has_more_rows) {
  HashTableCtx::ExprValuesCache* expr_vals_cache = ht_ctx->expr_values_cache();
  // Hoist lookups out of non-null branch to speed up non-null case.
  const uint32_t hash = expr_vals_cache->CurExprValuesHash();
  const uint32_t partition_idx = hash >> (32 - NUM_PARTITIONING_BITS);
  if (expr_vals_cache->IsRowNull()) return Status::OK();
  // To process this row, we first see if it can be aggregated or inserted into this
  // partition's hash table. If we need to insert it and that fails, due to OOM, we
  // spill the partition. The partition to spill is not necessarily dst_partition,
  // so we can try again to insert the row.
  HashTable* hash_tbl = GetHashTable(partition_idx);
  Partition* dst_partition = hash_partitions_[partition_idx];
  DCHECK(dst_partition != nullptr);
  DCHECK_EQ(dst_partition->is_spilled(), hash_tbl == nullptr);
  if (hash_tbl == nullptr) {
    // This partition is already spilled, just append the row.
    return AppendSpilledRow<AGGREGATED_ROWS>(dst_partition, row);
  }

  DCHECK(dst_partition->aggregated_row_stream->is_pinned());
  bool found;
  // Find the appropriate bucket in the hash table. There will always be a free
  // bucket because we checked the size above.
  HashTable::Iterator it =
      hash_tbl->FindBuildRowBucket<BucketType::MATCH_UNSET>(ht_ctx, &found);
  DCHECK(!it.AtEnd()) << "Hash table had no free buckets";
  if (AGGREGATED_ROWS) {
    // If the row is already an aggregate row, it cannot match anything in the
    // hash table since we process the aggregate rows first. These rows should
    // have been aggregated in the initial pass.
    DCHECK(!found);
  } else if (found) {
    // Row is already in hash table. Do the aggregation and we're done.
    UpdateTuple(
        dst_partition->agg_fn_evals.data(), it.GetTuple<BucketType::MATCH_UNSET>(), row);
    return Status::OK();
  }

  // If we are seeing this result row for the first time, we need to construct the
  // result row and initialize it.
  return AddIntermediateTuple<AGGREGATED_ROWS>(dst_partition, row, hash, it,
      has_more_rows);
}

template <bool AGGREGATED_ROWS>
Status GroupingAggregator::AddIntermediateTuple(Partition* __restrict__ partition,