Method StartProducing

cpp/src/arrow/acero/source_node.cc:166–301

  }

  Status StartProducing() override {
    NoteStartProducing(ToStringExtra());
    {
      // If another exec node encountered an error during its StartProducing call
      // it might have already called StopProducing on all of its inputs (including this
      // node).
      //
      std::unique_lock<std::mutex> lock(mutex_);
      if (stop_requested_) {
        return Status::OK();
      }
      started_ = true;
    }

    CallbackOptions options;
    // These options will transfer execution to the desired Executor if necessary.
    // This can happen for in-memory scans where batches don't require
    // any CPU work to decode. Otherwise, parsing etc should have already
    // been placed us on the desired Executor and no queues will be pushed to.
    options.executor = plan()->query_context()->executor();
    options.should_schedule = ShouldSchedule::IfDifferentExecutor;
    ARROW_ASSIGN_OR_RAISE(Future<> scan_task, plan_->query_context()->BeginExternalTask(
                                                  "SourceNode::DatasetScan"));
    if (!scan_task.is_valid()) {
      // Plan has already been aborted, no need to start scanning
      return Status::OK();
    }
    auto fut = Loop([this, options] {
      std::unique_lock<std::mutex> lock(mutex_);
      if (stop_requested_) {
        return Future<ControlFlow<int>>::MakeFinished(Break(batch_count_));
      }
      lock.unlock();

      arrow::util::tracing::Span fetch_batch_span;
      auto fetch_batch_scope =
          START_SCOPED_SPAN(fetch_batch_span, "SourceNode::ReadBatch");
      return generator_().Then(
          [this](
              const std::optional<ExecBatch>& morsel_or_end) -> Future<ControlFlow<int>> {
            std::unique_lock<std::mutex> lock(mutex_);
            if (IsIterationEnd(morsel_or_end) || stop_requested_) {
              return Break(batch_count_);
            }
            lock.unlock();
            SliceAndDeliverMorsel(*morsel_or_end);
            lock.lock();
            if (!backpressure_future_.is_finished()) {
              EVENT_ON_CURRENT_SPAN("SourceNode::BackpressureApplied");
              return backpressure_future_.Then(
                  []() -> ControlFlow<int> { return Continue(); });
            }
            return Future<ControlFlow<int>>::MakeFinished(Continue());
          },
          [](const Status& err) -> Future<ControlFlow<int>> { return err; }, options);
    });
    fut.AddCallback(
        [this, scan_task](Result<int> maybe_total_batches) mutable {