Method evaluateAggregates

bolt/exec/StreamingAggregation.cpp:251–292

}

void StreamingAggregation::evaluateAggregates() {
  NanosecondTimer funcTimer(&stats_.aggFunctionTimeNs);
  for (auto i = 0; i < aggregates_.size(); ++i) {
    const auto& aggregate = aggregates_.at(i);
    if (!aggregate.sortingKeys.empty()) {
      continue;
    }

    const auto& rows = getSelectivityVector(i);
    if (!rows.hasSelections()) {
      continue;
    }

    if (aggregate.distinct) {
      distinctAggregations_.at(i)->addInput(inputGroups_.data(), input_, rows);
      continue;
    }

    const auto& function = aggregate.function;
    const auto& inputs = aggregate.inputs;
    const auto& constantInputs = aggregate.constantInputs;

    std::vector<VectorPtr> args;
    for (auto j = 0; j < inputs.size(); ++j) {
      if (inputs[j] == kConstantChannel) {
        args.push_back(constantInputs[j]);
      } else {
        args.push_back(input_->childAt(inputs[j]));
      }
    }

    if (isRawInput(step_)) {
      function->addRawInput(inputGroups_.data(), rows, args, false);
    } else {
      function->addIntermediateResults(inputGroups_.data(), rows, args, false);
    }
  }

  if (sortedAggregations_) {
    sortedAggregations_->addInput(inputGroups_.data(), input_);
  }
}

bool StreamingAggregation::isFinished() {
  return noMoreInput_ && input_ == nullptr && numGroups_ == 0;