Method addInput

bolt/exec/HashBuild.cpp:534–717

}

void HashBuild::addInput(RowVectorPtr input) {
  checkRunning();

  if (!ensureInputFits(input, {nullptr, 0})) {
    BOLT_CHECK_NOT_NULL(input_);
    BOLT_CHECK(future_.valid());
    return;
  }

  BOLT_TEST_ADJUST("bytedance::bolt::exec::HashBuild::addInput", this);

  activeRows_.resize(input->size());
  activeRows_.setAll();

  auto& hashers = table_->hashers();

  for (auto i = 0; i < hashers.size(); ++i) {
    auto key = input->childAt(hashers[i]->channel())->loadedVector();
    hashers[i]->decode(*key, activeRows_);
  }

  // Update statistics for null keys in join operator.
  // We use activeRows_ to store which rows have some null keys,
  // and reset it after using it.
  // Only process when input is not spilled, to avoid overcounting.
  if (!isInputFromSpill()) {
    auto lockedStats = stats_.wlock();
    deselectRowsWithNulls(hashers, activeRows_);
    lockedStats->numNullKeys +=
        activeRows_.size() - activeRows_.countSelected();
    activeRows_.setAll();
  }

  if (!isRightJoin(joinType_) && !isFullJoin(joinType_) &&
      !isRightSemiProjectJoin(joinType_) &&
      !isLeftNullAwareJoinWithFilter(joinNode_)) {
    deselectRowsWithNulls(hashers, activeRows_);
    if (nullAware_ && !joinHasNullKeys_ &&
        activeRows_.countSelected() < input->size()) {
      joinHasNullKeys_ = true;
    }
  } else if (nullAware_ && !joinHasNullKeys_) {
    for (auto& hasher : hashers) {
      auto& decoded = hasher->decodedVector();
      if (decoded.mayHaveNulls()) {
        auto* nulls = decoded.nulls(&activeRows_);
        if (nulls && bits::countNulls(nulls, 0, activeRows_.end()) > 0) {
          joinHasNullKeys_ = true;
          break;
        }
      }
    }
  }

  for (auto i = 0; i < dependentChannels_.size(); ++i) {
    decoders_[i]->decode(
        *input->childAt(dependentChannels_[i])->loadedVector(), activeRows_);
  }