Method ReadAsync

cpp/src/arrow/csv/reader.cc:1050–1086

  Result<std::shared_ptr<Table>> Read() override { return ReadAsync().result(); }

  Future<std::shared_ptr<Table>> ReadAsync() override {
    // Note that this task group doesn't survive our destruction, so it's essential
    // that async callbacks own a strong ref to us.
    task_group_ = TaskGroup::MakeThreaded(cpu_executor_, io_context_.stop_token());

    auto self = shared_from_this();
    return ProcessFirstBuffer().Then([self](const std::shared_ptr<Buffer>& first_buffer) {
      auto block_generator = ThreadedBlockReader::MakeAsyncIterator(
          self->buffer_generator_, MakeChunker(self->parse_options_), first_buffer,
          self->read_options_.skip_rows_after_names);

      std::function<Status(CSVBlock)> block_visitor =
          [self](CSVBlock maybe_block) -> Status {
        // The logic in VisitAsyncGenerator ensures that we will never be
        // passed an empty block (visit does not call with the end token) so
        // we can be assured maybe_block has a value.
        DCHECK_GE(maybe_block.block_index, 0);
        DCHECK(!maybe_block.consume_bytes);

        // Launch parse task
        self->task_group_->Append(
            [self, maybe_block] { return self->ParseAndInsert(maybe_block); });
        return Status::OK();
      };

      return VisitAsyncGenerator(std::move(block_generator), block_visitor)
          .Then([self]() -> Future<> {
            // By this point we've added all top level tasks so it is safe to call
            // FinishAsync
            return self->task_group_->FinishAsync();
          })
          .Then([self]() -> Result<std::shared_ptr<Table>> {
            // Finish conversion, create schema and table
            return self->MakeTable();
          });
    });
  }

 protected:
  Future<std::shared_ptr<Buffer>> ProcessFirstBuffer() {