Method Process

cpp/src/arrow/acero/asof_join_node.cc:1065–1104

  }

  bool Process() {
    std::lock_guard<std::mutex> guard(gate_);
    if (!CheckEnded()) {
      return false;
    }

    // Process batches while we have data
    for (;;) {
      Result<std::shared_ptr<RecordBatch>> result = ProcessInner();

      if (result.ok()) {
        auto out_rb = *result;
        if (!out_rb) break;
        ExecBatch out_b(*out_rb);
        out_b.index = batches_produced_++;
        DEBUG_SYNC(this, "produce batch ", out_b.index, ":", DEBUG_MANIP(std::endl),
                   out_rb->ToString(), DEBUG_MANIP(std::endl));
        Status st = output_->InputReceived(this, std::move(out_b));
        if (!st.ok()) {
          EndFromProcessThread(std::move(st));
        }
      } else {
        EndFromProcessThread(result.status());
        return false;
      }
    }

    // Report to the output the total batch count, if we've already finished everything
    // (there are two places where this can happen: here and InputFinished)
    //
    // It may happen here in cases where InputFinished was called before we were finished
    // producing results (so we didn't know the output size at that time)
    if (!CheckEnded()) {
      return false;
    }

    // There is no more we can do now but there is still work remaining for later when
    // more data arrives.
    return true;
  }

  void ProcessThread() {
    for (;;) {