Method ReadNext

r/src/compute-exec.cpp:106–149

  std::shared_ptr<arrow::Schema> schema() const override { return schema_; }

  arrow::Status ReadNext(std::shared_ptr<arrow::RecordBatch>* batch_out) override {
    // If this is the first batch getting pulled, tell the exec plan to
    // start producing
    if (plan_status_ == PLAN_NOT_STARTED) {
      StartProducing();
    }

    // If we've closed the reader, keep sending nullptr
    // (consistent with what most RecordBatchReader subclasses do)
    if (plan_status_ == PLAN_FINISHED) {
      batch_out->reset();
      return arrow::Status::OK();
    }

    // Check for cancellation and stop the plan if we have a request. When
    // the ExecPlan supports passing a StopToken and handling this itself,
    // this will be redundant.
    if (stop_token_.IsStopRequested()) {
      StopProducing();
      return stop_token_.Poll();
    }

    auto out = sink_gen_().result();
    if (!out.ok()) {
      StopProducing();
      return out.status();
    }

    if (out.ValueUnsafe()) {
      auto batch_result = out.ValueUnsafe()->ToRecordBatch(schema_, gc_memory_pool());
      if (!batch_result.ok()) {
        StopProducing();
        return batch_result.status();
      }

      *batch_out = batch_result.ValueUnsafe();
    } else {
      batch_out->reset();
      plan_status_ = PLAN_FINISHED;
      return plan_->finished().status();
    }

    return arrow::Status::OK();
  }

  arrow::Status Close() override {
    StopProducing();