Class CSVBufferIterator

cpp/src/arrow/csv/reader.cc:93–142

An iterator of Buffers that makes sure there is no straddling CRLF sequence.

// An iterator of Buffers that makes sure there is no straddling CRLF sequence.
class CSVBufferIterator {
 public:
  static Iterator<std::shared_ptr<Buffer>> Make(
      Iterator<std::shared_ptr<Buffer>> buffer_iterator) {
    Transformer<std::shared_ptr<Buffer>, std::shared_ptr<Buffer>> fn =
        CSVBufferIterator();
    return MakeTransformedIterator(std::move(buffer_iterator), fn);
  }

  static AsyncGenerator<std::shared_ptr<Buffer>> MakeAsync(
      AsyncGenerator<std::shared_ptr<Buffer>> buffer_iterator) {
    Transformer<std::shared_ptr<Buffer>, std::shared_ptr<Buffer>> fn =
        CSVBufferIterator();
    return MakeTransformedGenerator(std::move(buffer_iterator), fn);
  }

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

    int64_t offset = 0;
    if (first_buffer_) {
      ARROW_ASSIGN_OR_RAISE(auto data, util::SkipUTF8BOM(buf->data(), buf->size()));
      offset += data - buf->data();
      DCHECK_GE(offset, 0);
      first_buffer_ = false;
    }

    if (trailing_cr_ && buf->data()[offset] == '\n') {
      // Skip '\r\n' line separator that started at the end of previous buffer
      ++offset;
    }

    trailing_cr_ = (buf->data()[buf->size() - 1] == '\r');
    buf = SliceBuffer(std::move(buf), offset);
    if (buf->size() == 0) {
      // EOF
      return TransformFinish();
    } else {
      return TransformYield(std::move(buf));
    }
  }

 protected:
  bool first_buffer_ = true;
  // Whether there was a trailing CR at the end of last received buffer
  bool trailing_cr_ = false;
};

struct CSVBlock {
  // (partial + completion + buffer) is an entire delimited CSV buffer.