Method Execute

cpp/src/arrow/compute/exec.cc:1024–1077

class VectorExecutor : public KernelExecutorImpl<VectorKernel> {
 public:
  Status Execute(const ExecBatch& batch, ExecListener* listener) override {
    // Some vector kernels have a separate code path for handling
    // chunked arrays (VectorKernel::exec_chunked) so we check if we
    // have any chunked arrays. If we do and an exec_chunked function
    // is defined then we call that.
    bool have_chunked_arrays = false;
    for (const Datum& arg : batch.values) {
      if (arg.is_chunked_array()) have_chunked_arrays = true;
    }

    output_num_buffers_ = static_cast<int>(output_type_.type->layout().buffers.size());

    // Decide if we need to preallocate memory for this kernel
    validity_preallocated_ =
        (kernel_->null_handling != NullHandling::COMPUTED_NO_PREALLOCATE &&
         kernel_->null_handling != NullHandling::OUTPUT_NOT_NULL);
    if (kernel_->mem_allocation == MemAllocation::PREALLOCATE) {
      data_preallocated_.clear();
      ComputeDataPreallocate(*output_type_.type, &data_preallocated_);
    }

    if (kernel_->can_execute_chunkwise) {
      RETURN_NOT_OK(span_iterator_.Init(batch, exec_context()->exec_chunksize()));
      ExecSpan span;
      while (span_iterator_.Next(&span)) {
        RETURN_NOT_OK(Exec(span, listener));
      }
    } else {
      // Kernel cannot execute chunkwise. If we have any chunked
      // arrays, then VectorKernel::exec_chunked must be defined
      // otherwise we raise an error
      if (have_chunked_arrays) {
        RETURN_NOT_OK(ExecChunked(batch, listener));
      } else {
        // No chunked arrays. We pack the args into an ExecSpan and
        // call the regular exec code path
        ExecSpan span(batch);
        if (CheckIfAllScalar(batch)) {
          PromoteExecSpanScalars(&span);
        }
        RETURN_NOT_OK(Exec(span, listener));
      }
    }

    if (kernel_->finalize) {
      // Intermediate results require post-processing after the execution is
      // completed (possibly involving some accumulated state)
      RETURN_NOT_OK(kernel_->finalize(kernel_ctx_, &results_));
      for (const auto& result : results_) {
        RETURN_NOT_OK(listener->OnResult(result));
      }
    }
    return Status::OK();
  }

  Datum WrapResults(const std::vector<Datum>& inputs,
                    const std::vector<Datum>& outputs) override {