MCPcopy Create free account
hub / github.com/apache/arrow / ReadAsync

Method ReadAsync

cpp/src/arrow/csv/reader.cc:1056–1092  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

1054 Result<std::shared_ptr<Table>> Read() override { return ReadAsync().result(); }
1055
1056 Future<std::shared_ptr<Table>> ReadAsync() override {
1057 // Note that this task group doesn't survive our destruction, so it's essential
1058 // that async callbacks own a strong ref to us.
1059 task_group_ = TaskGroup::MakeThreaded(cpu_executor_, io_context_.stop_token());
1060
1061 auto self = shared_from_this();
1062 return ProcessFirstBuffer().Then([self](const std::shared_ptr<Buffer>& first_buffer) {
1063 auto block_generator = ThreadedBlockReader::MakeAsyncIterator(
1064 self->buffer_generator_, MakeChunker(self->parse_options_), first_buffer,
1065 self->read_options_.skip_rows_after_names);
1066
1067 std::function<Status(CSVBlock)> block_visitor =
1068 [self](CSVBlock maybe_block) -> Status {
1069 // The logic in VisitAsyncGenerator ensures that we will never be
1070 // passed an empty block (visit does not call with the end token) so
1071 // we can be assured maybe_block has a value.
1072 DCHECK_GE(maybe_block.block_index, 0);
1073 DCHECK(!maybe_block.consume_bytes);
1074
1075 // Launch parse task
1076 self->task_group_->Append(
1077 [self, maybe_block] { return self->ParseAndInsert(maybe_block); });
1078 return Status::OK();
1079 };
1080
1081 return VisitAsyncGenerator(std::move(block_generator), block_visitor)
1082 .Then([self]() -> Future<> {
1083 // By this point we've added all top level tasks so it is safe to call
1084 // FinishAsync
1085 return self->task_group_->FinishAsync();
1086 })
1087 .Then([self]() -> Result<std::shared_ptr<Table>> {
1088 // Finish conversion, create schema and table
1089 return self->MakeTable();
1090 });
1091 });
1092 }
1093
1094 protected:
1095 Future<std::shared_ptr<Buffer>> ProcessFirstBuffer() {

Callers

nothing calls this directly

Calls 8

VisitAsyncGeneratorFunction · 0.85
ThenMethod · 0.80
MakeTableMethod · 0.80
MakeChunkerFunction · 0.70
OKFunction · 0.50
AppendMethod · 0.45
ParseAndInsertMethod · 0.45
FinishAsyncMethod · 0.45

Tested by

no test coverage detected