Method InitFromBlock

cpp/src/arrow/csv/reader.cc:906–952

  }

  Future<> InitFromBlock(const DecodedBlock& block,
                         AsyncGenerator<DecodedBlock> batch_gen, int max_readahead,
                         int64_t prev_bytes_processed) {
    if (!block.record_batch) {
      // End of file just return null batches
      record_batch_gen_ = MakeEmptyGenerator<std::shared_ptr<RecordBatch>>();
      return Status::OK();
    }

    schema_ = block.record_batch->schema();

    if (block.record_batch->num_rows() == 0) {
      // Keep consuming blocks until the first non empty block is found
      auto self = shared_from_this();
      prev_bytes_processed += block.bytes_processed;
      return batch_gen().Then([self, batch_gen, max_readahead,
                               prev_bytes_processed](const DecodedBlock& next_block) {
        return self->InitFromBlock(next_block, std::move(batch_gen), max_readahead,
                                   prev_bytes_processed);
      });
    }

    AsyncGenerator<DecodedBlock> readahead_gen;
    if (read_options_.use_threads) {
      readahead_gen = MakeReadaheadGenerator(std::move(batch_gen), max_readahead);
    } else {
      readahead_gen = std::move(batch_gen);
    }

    AsyncGenerator<DecodedBlock> restarted_gen =
        MakeGeneratorStartsWith({block}, std::move(readahead_gen));

    auto bytes_decoded = bytes_decoded_;
    auto unwrap_and_record_bytes =
        [bytes_decoded, prev_bytes_processed](
            const DecodedBlock& block) mutable -> Result<std::shared_ptr<RecordBatch>> {
      bytes_decoded->fetch_add(block.bytes_processed + prev_bytes_processed);
      prev_bytes_processed = 0;
      return block.record_batch;
    };

    auto unwrapped =
        MakeMappedGenerator(std::move(restarted_gen), std::move(unwrap_and_record_bytes));

    record_batch_gen_ = MakeCancellable(std::move(unwrapped), io_context_.stop_token());
    return Status::OK();
  }

  std::shared_ptr<Schema> schema_;
  AsyncGenerator<std::shared_ptr<RecordBatch>> record_batch_gen_;