| 226 | } |
| 227 | |
| 228 | Status Finish() { |
| 229 | writer_state_->staged_rows_count -= rows_currently_staged_; |
| 230 | while (!staged_batches_.empty()) { |
| 231 | RETURN_NOT_OK(PopAndDeliverStagedBatch().status().OrElse( |
| 232 | [&](auto&&) { file_tasks_.reset(); })); |
| 233 | } |
| 234 | // At this point all write tasks have been added. Because the scheduler |
| 235 | // is a 1-task FIFO we know this task will run at the very end and can |
| 236 | // add it now. |
| 237 | file_tasks_->AddSimpleTask([self = shared_from_this()] { return self->DoFinish(); }, |
| 238 | "DatasetWriter::FinishFile"sv); |
| 239 | file_tasks_.reset(); |
| 240 | return Status::OK(); |
| 241 | } |
| 242 | |
| 243 | private: |
| 244 | Future<> WriteNext(std::shared_ptr<RecordBatch> next) { |