Method ARROW_ASSIGN_OR_RAISE

cpp/src/arrow/acero/source_node.cc:187–237

    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 {
          if (maybe_total_batches.ok()) {
            plan_->query_context()->ScheduleTask(
                [this, total_batches = *maybe_total_batches] {
                  return output_->InputFinished(this, total_batches);
                },
                "SourceNode::InputFinished");
          }
          scan_task.MarkFinished(maybe_total_batches.status());
        },
        options);
    return Status::OK();
  }

  const Ordering& ordering() const override { return ordering_; }

  void PauseProducing(ExecNode* output, int32_t counter) override {
    std::lock_guard<std::mutex> lg(mutex_);