Method addInput

bolt/exec/SortWindowBuild.cpp:103–142

Handle top N

// Handle top N
void SortWindowBuild::addInput(RowVectorPtr input) {
  if (followedTopNum_ <= 0) {
    return;
  }

  ensureInputFits(input);

  const auto numInput = input->size();
  vector_size_t rowCnt = 0;
  if (table_) {
    // keep rows in topN heap
    SelectivityVector rows(numInput);
    table_->prepareForGroupProbe(
        *lookup_,
        input,
        rows,
        false,
        BaseHashTable::kNoSpillInputStartPartitionBit);
    table_->groupProbe(*lookup_);

    // Initialize new partitions.
    initializeNewPartitions();

    for (auto row = 0; row < numInput; ++row) {
      // get partitionKeys and calc hashid
      auto& partition = partitionAt(lookup_->hits[row]);
      if (processInputRow(row, partition)) {
        rowCnt++;
      }
    }
  } else {
    // no partition key
    for (auto row = 0; row < numInput; ++row) {
      if (processInputRow(row, *singlePartition_)) {
        rowCnt++;
      }
    }
  }
  numRows_ += rowCnt;
}

void SortWindowBuild::ensureInputFits(const RowVectorPtr& input) {
  if (spillConfig_ == nullptr) {