Method addInputForActiveRows

bolt/exec/GroupingSet.cpp:244–341

}

void GroupingSet::addInputForActiveRows(
    const RowVectorPtr& input,
    bool mayPushdown) {
  BOLT_CHECK(!isGlobal_);
  if (!table_) {
    createHashTable();
  }
  ensureInputFits(input);

  BOLT_TEST_ADJUST(
      "bytedance::bolt::exec::GroupingSet::addInputForActiveRows", this);

  if (!bypassProbeHT_) { // probing time
    NanosecondTimer probeTimer(&stats_.aggProbeTimeNs);
    table_->prepareForGroupProbe(
        *lookup_,
        input,
        activeRows_,
        ignoreNullKeys_,
        BaseHashTable::kNoSpillInputStartPartitionBit);
    if (lookup_->rows.empty()) {
      // No rows to probe. Can happen when ignoreNullKeys_ is true and all rows
      // have null keys.
      return;
    }
    table_->groupProbe(*lookup_);
    masks_.addInput(input, activeRows_);
  } else {
    // if there are too many distinct group-by keys and they should be spilled,
    // we can bypass probing the hash table, directly add rows to the table. In
    // the merge phase of output, dong the final agg can get correct results.

    NanosecondTimer probeTimer(&stats_.aggProbeBypassTimeNs);
    stats_.aggProbeBypassCount++;
    table_->directAddRows(*lookup_, input, activeRows_, ignoreNullKeys_);
    if (lookup_->rows.empty()) {
      // No rows to probe. Can happen when ignoreNullKeys_ is true and all rows
      // have null keys.
      return;
    }
    masks_.addInput(input, activeRows_);
  }

  // agg func time
  {
    NanosecondTimer funcTimer(&stats_.aggFunctionTimeNs);
    auto* groups = lookup_->hits.data();
    auto& newGroups = lookup_->newGroups;
    for (auto i = 0; i < aggregates_.size(); ++i) {
      if (!aggregates_[i].sortingKeys.empty()) {
        continue;
      }

      const auto& rows = getSelectivityVector(i);

      if (aggregates_[i].distinct) {
        if (!newGroups.empty()) {
          distinctAggregations_[i]->initializeNewGroups(groups, newGroups);