| 521 | |
| 522 | protected: |
| 523 | Status DoFinish() { |
| 524 | auto scope = TraceFinish(); |
| 525 | ARROW_ASSIGN_OR_RAISE(Datum sorted, impl_->DoFinish()); |
| 526 | TableBatchReader reader(*sorted.table()); |
| 527 | while (true) { |
| 528 | std::shared_ptr<RecordBatch> batch; |
| 529 | RETURN_NOT_OK(reader.ReadNext(&batch)); |
| 530 | if (!batch) break; |
| 531 | bool did_push = producer_.Push(ExecBatch(*batch)); |
| 532 | if (!did_push) break; // producer_ was Closed already |
| 533 | } |
| 534 | return Status::OK(); |
| 535 | } |
| 536 | |
| 537 | Status Finish() override { |
| 538 | arrow::util::tracing::Span span; |