Method operator()

cpp/src/arrow/csv/reader.cc:223–276

  }

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

    bool is_final = (next_buffer == nullptr);
    int64_t bytes_skipped = 0;

    if (skip_rows_) {
      bytes_skipped += partial_->size();
      auto orig_size = buffer_->size();
      RETURN_NOT_OK(
          chunker_->ProcessSkip(partial_, buffer_, is_final, &skip_rows_, &buffer_));
      bytes_skipped += orig_size - buffer_->size();
      auto empty = std::make_shared<Buffer>(nullptr, 0);
      if (skip_rows_) {
        // Still have rows beyond this buffer to skip return empty block
        partial_ = std::move(buffer_);
        buffer_ = next_buffer;
        return TransformYield<CSVBlock>(CSVBlock{empty, empty, empty, block_index_++,
                                                 is_final, bytes_skipped,
                                                 [](int64_t) { return Status::OK(); }});
      }
      partial_ = std::move(empty);
    }

    std::shared_ptr<Buffer> completion;

    if (is_final) {
      // End of file reached => compute completion from penultimate block
      RETURN_NOT_OK(chunker_->ProcessFinal(partial_, buffer_, &completion, &buffer_));
    } else {
      // Get completion of partial from previous block.
      RETURN_NOT_OK(
          chunker_->ProcessWithPartial(partial_, buffer_, &completion, &buffer_));
    }
    int64_t bytes_before_buffer = partial_->size() + completion->size();

    auto consume_bytes = [this, bytes_before_buffer,
                          next_buffer](int64_t nbytes) -> Status {
      DCHECK_GE(nbytes, 0);
      int64_t offset = nbytes - bytes_before_buffer;
      // All data before the buffer should have been consumed.
      // This is checked in Parse() and BlockParsingOperator::operator().
      DCHECK_GE(offset, 0);
      partial_ = SliceBuffer(buffer_, offset);
      buffer_ = next_buffer;
      return Status::OK();
    };

    return TransformYield<CSVBlock>(CSVBlock{partial_, completion, buffer_,
                                             block_index_++, is_final, bytes_skipped,
                                             std::move(consume_bytes)});
  }
};

// An object that reads delimited CSV blocks for threaded use.