Class KeyHasher

cpp/src/arrow/acero/asof_join_node.cc:389–460

a specialized higher-performance variation of Hashing64 logic from hash_join_node the code here avoids recreating objects that are independent of each batch processed

// a specialized higher-performance variation of Hashing64 logic from hash_join_node
// the code here avoids recreating objects that are independent of each batch processed
class KeyHasher {
  friend class AsofJoinNode;

  static constexpr int kMiniBatchLength = arrow::util::MiniBatch::kMiniBatchLength;

 public:
  // the key hasher is not thread-safe and is only used in sequential batch processing
  // of the input it is associated with
  KeyHasher(size_t index, const std::vector<col_index_t>& indices)
      : index_(index),
        indices_(indices),
        metadata_(indices.size()),
        batch_(NULLPTR),
        hashes_(),
        ctx_(),
        column_arrays_(),
        stack_() {
    ctx_.stack = &stack_;
    column_arrays_.resize(indices.size());
  }

  Status Init(ExecContext* exec_context, const std::shared_ptr<arrow::Schema>& schema) {
    ctx_.hardware_flags = exec_context->cpu_info()->hardware_flags();
    const auto& fields = schema->fields();
    for (size_t k = 0; k < metadata_.size(); k++) {
      ARROW_ASSIGN_OR_RAISE(metadata_[k],
                            ColumnMetadataFromDataType(fields[indices_[k]]->type()));
    }
    return stack_.Init(exec_context->memory_pool(),
                       4 * kMiniBatchLength * sizeof(uint32_t));
  }

  // invalidate cached hashes for batch - required when it changes
  // only this method can be called concurrently with HashesFor
  void Invalidate() { batch_ = NULLPTR; }

  // compute and cache a hash for each row of the given batch
  const std::vector<HashType>& HashesFor(const RecordBatch* batch) {
    if (batch_ == batch) {
      return hashes_;  // cache hit - return cached hashes
    }
    Invalidate();
    size_t batch_length = batch->num_rows();
    hashes_.resize(batch_length);
    for (int64_t i = 0; i < static_cast<int64_t>(batch_length); i += kMiniBatchLength) {
      int64_t length = std::min(static_cast<int64_t>(batch_length - i),
                                static_cast<int64_t>(kMiniBatchLength));
      for (size_t k = 0; k < indices_.size(); k++) {
        auto array_data = batch->column_data(indices_[k]);
        column_arrays_[k] =
            ColumnArrayFromArrayDataAndMetadata(array_data, metadata_[k], i, length);
      }
      // write directly to the cache
      Hashing64::HashMultiColumn(column_arrays_, &ctx_, hashes_.data() + i);
    }
    DEBUG_SYNC(node_, "key hasher ", index_, " got hashes ",
               compute::internal::GenericToString(hashes_), DEBUG_MANIP(std::endl));
    batch_ = batch;  // associate cache with current batch