Method operator()

cpp/src/arrow/csv/reader.cc:296–351

  }

  Result<TransformFlow<CSVBlock>> operator()(std::shared_ptr<Buffer> next_buffer) {
    if (buffer_ == nullptr) {
      // EOF
      return TransformFinish();
    }

    bool is_final = (next_buffer == nullptr);

    auto current_partial = std::move(partial_);
    auto current_buffer = std::move(buffer_);
    int64_t bytes_skipped = 0;

    if (skip_rows_) {
      auto orig_size = current_buffer->size();
      bytes_skipped = current_partial->size();
      RETURN_NOT_OK(chunker_->ProcessSkip(current_partial, current_buffer, is_final,
                                          &skip_rows_, &current_buffer));
      bytes_skipped += orig_size - current_buffer->size();
      current_partial = std::make_shared<Buffer>(nullptr, 0);
      if (skip_rows_) {
        partial_ = std::move(current_buffer);
        buffer_ = std::move(next_buffer);
        return TransformYield<CSVBlock>(CSVBlock{current_partial,
                                                 current_partial,
                                                 current_partial,
                                                 block_index_++,
                                                 is_final,
                                                 bytes_skipped,
                                                 {}});
      }
    }

    std::shared_ptr<Buffer> whole, completion, next_partial;

    if (is_final) {
      // End of file reached => compute completion from penultimate block
      RETURN_NOT_OK(
          chunker_->ProcessFinal(current_partial, current_buffer, &completion, &whole));
    } else {
      // Get completion of partial from previous block.
      std::shared_ptr<Buffer> starts_with_whole;
      // Get completion of partial from previous block.
      RETURN_NOT_OK(chunker_->ProcessWithPartial(current_partial, current_buffer,
                                                 &completion, &starts_with_whole));

      // Get a complete CSV block inside `partial + block`, and keep
      // the rest for the next iteration.
      RETURN_NOT_OK(chunker_->Process(starts_with_whole, &whole, &next_partial));
    }

    partial_ = std::move(next_partial);
    buffer_ = std::move(next_buffer);

    return TransformYield<CSVBlock>(CSVBlock{
        current_partial, completion, whole, block_index_++, is_final, bytes_skipped, {}});
  }
};