Method convertCompositeInput

bolt/exec/GroupingSet.cpp:2177–2220

  const auto rowSizeFromContainer = table_->rows()->estimateRowSize();
  if (!hasExternalMemoryAccumulators_) {
    return rowSizeFromContainer;
  }

  // RowContainer::estimateRowSize() only accounts for RowContainer's row blocks
  // and HashStringAllocator (e.g. strings). For aggregates which allocate state
  // directly from MemoryPool (external memory), incorporate pool usage per
  // group to avoid under-estimation and too large output batches.
  const int64_t rowSizeFromPool = std::max<int64_t>(
      1, static_cast<int64_t>(pool_.currentBytes() / table_->numDistinct()));
  const int64_t rowSize =
      std::max<int64_t>(rowSizeFromPool, rowSizeFromContainer.value_or(0));
  return rowSize;
}

void GroupingSet::convertCompositeInput(
    const std::vector<AggregateInfo>& extractAggregates,
    const CompositeRowVectorPtr& input,
    const RowVectorPtr& result) {
  if (!table_) {
    createHashTable();
  }
  auto* container = table_->rows();
  // initialize extractAggregates to make sure Aggregate::numNulls_ not 0 if it
  // is used
  if (!extractAggregatesInitialized_) {
    // fixedRowSize should be accuracy
    auto rowSize = container->fixedRowSize() + 16;
    char* rowPtr = static_cast<char*>(pool_.allocate(rowSize));
    const std::vector<char*> tmpGroups{rowPtr};
    const std::vector<vector_size_t> groupIndex{0};
    initializeAggregates(extractAggregates, *container, false);
    for (auto i = 0; i < extractAggregates.size(); ++i) {
      extractAggregates[i].function->initializeNewGroups(
          const_cast<char**>(tmpGroups.data()), groupIndex);
    }
    pool_.free(rowPtr, rowSize);
    extractAggregatesInitialized_ = true;
  }
  auto& inputRows = input->rawRows();
  for (auto& row : inputRows) {
    ContainerRow2RowSerde::deserialize(
        const_cast<char*&>(row), rowInfo_.value(), false);
  }

  for (auto i = 0; i < keyChannels_.size(); ++i) {
    rowToColumnVector(