Method extractValues

bolt/exec/SortedAggregations.cpp:396–463

}

void SortedAggregations::extractValues(
    folly::Range<char**> groups,
    const RowVectorPtr& result) {
  raw_vector<int32_t> temp;
  SelectivityVector rows;
  std::vector<char*> groupRows;
  for (const auto& [sortingSpec, aggregates] : aggregates_) {
    std::vector<VectorPtr> inputVectors;
    size_t numInputColumns = 0;
    for (const auto& aggregate : aggregates) {
      numInputColumns += aggregate->inputs.size();
    }
    inputVectors.resize(numInputColumns);

    // For each group, sort inputs, add them to aggregate.
    for (auto* group : groups) {
      auto* accumulator = reinterpret_cast<RowPointers*>(group + offset_);
      groupRows.resize(accumulator->size);
      accumulator->read(folly::Range(groupRows.data(), groupRows.size()));

      sortSingleGroup(groupRows, sortingSpec);

      size_t firstInputColumn = 0;
      for (const auto& aggregate : aggregates) {
        std::vector<VectorPtr> aggregateInputs;
        aggregateInputs.reserve(aggregate->inputs.size());
        for (auto i = 0; i < aggregate->inputs.size(); ++i) {
          aggregateInputs.push_back(
              std::move(inputVectors[firstInputColumn + i]));
        }

        // TODO Process group rows in batches to avoid creating very large
        // input vectors.
        const auto numRows =
            extractSingleGroup(groupRows, *aggregate, aggregateInputs);
        if (numRows == 0) {
          // Mask must be false for all 'groupRows'.
          continue;
        }

        rows.resize(numRows);
        aggregate->function->addSingleGroupRawInput(
            group, rows, aggregateInputs, false);

        for (auto i = 0; i < aggregate->inputs.size(); ++i) {
          inputVectors[firstInputColumn + i] = std::move(aggregateInputs[i]);
        }

        firstInputColumn += aggregateInputs.size();
      }
    }

    for (const auto& aggregate : aggregates) {
      aggregate->function->extractValues(
          groups.data(), groups.size(), &result->childAt(aggregate->output));

      // Release memory back to HashStringAllocator to allow next aggregate to
      // reuse it.