Method spillInput

bolt/exec/RowNumber.cpp:477–519

}

void RowNumber::spillInput(
    const RowVectorPtr& input,
    memory::MemoryPool* pool) {
  const auto numInput = input->size();

  std::vector<uint32_t> spillPartitions(numInput);
  const auto singlePartition =
      spillHashFunction_->partition(*input, spillPartitions);

  const auto numPartitions = spillHashFunction_->numPartitions();

  std::vector<BufferPtr> partitionIndices(numPartitions);
  std::vector<vector_size_t*> rawPartitionIndices(numPartitions);

  for (auto i = 0; i < numPartitions; ++i) {
    partitionIndices[i] = allocateIndices(numInput, pool);
    rawPartitionIndices[i] = partitionIndices[i]->asMutable<vector_size_t>();
  }

  std::vector<vector_size_t> numSpillInputs(numPartitions, 0);

  for (auto row = 0; row < numInput; ++row) {
    const auto partition = singlePartition.has_value() ? singlePartition.value()
                                                       : spillPartitions[row];
    rawPartitionIndices[partition][numSpillInputs[partition]++] = row;
  }

  // Ensure vector are lazy loaded before spilling.
  for (auto i = 0; i < input->childrenSize(); ++i) {
    input->childAt(i)->loadedVector();
  }

  for (int32_t partition = 0; partition < numSpillInputs.size(); ++partition) {
    const auto numInputs = numSpillInputs[partition];
    if (numInputs == 0) {
      continue;
    }

    inputSpiller_->spill(
        partition,
        wrapAndCombineDict(numInputs, partitionIndices[partition], input));
  }
}

} // namespace bytedance::bolt::exec