Method operator()

cpp/src/arrow/csv/reader.cc:403–453

TODO: this is almost entirely the same as ReaderMixin::Parse(). Refactor?

  // TODO: this is almost entirely the same as ReaderMixin::Parse(). Refactor?
  Result<ParsedBlock> operator()(const CSVBlock& block) {
    constexpr int32_t max_num_rows = std::numeric_limits<int32_t>::max();
    auto parser = std::make_shared<BlockParser>(
        io_context_.pool(), parse_options_, num_csv_cols_, num_rows_seen_, max_num_rows);

    std::shared_ptr<Buffer> straddling;
    std::vector<std::string_view> views;
    if (block.partial->size() != 0 || block.completion->size() != 0) {
      if (block.partial->size() == 0) {
        straddling = block.completion;
      } else if (block.completion->size() == 0) {
        straddling = block.partial;
      } else {
        ARROW_ASSIGN_OR_RAISE(
            straddling,
            ConcatenateBuffers({block.partial, block.completion}, io_context_.pool()));
      }
      views = {std::string_view(*straddling), std::string_view(*block.buffer)};
    } else {
      views = {std::string_view(*block.buffer)};
    }
    uint32_t parsed_size;
    if (block.is_final) {
      RETURN_NOT_OK(parser->ParseFinal(views, &parsed_size));
    } else {
      RETURN_NOT_OK(parser->Parse(views, &parsed_size));
    }

    // `partial + completion` should have been entirely consumed.
    const int64_t bytes_before_buffer = block.partial->size() + block.completion->size();
    if (static_cast<int64_t>(parsed_size) < bytes_before_buffer) {
      // This can happen if `newlines_in_values` is not enabled and
      // `partial + completion` ends with a newline inside a quoted string.
      // In this case, the BlockParser stops at the truncated data in the first
      // block (see gh-39857).
      return Status::Invalid(
          "CSV parser got out of sync with chunker. This can mean the data file "
          "contains cell values spanning multiple lines; please consider enabling "
          "the option 'newlines_in_values'.");
    }

    if (count_rows_) {
      num_rows_seen_ += parser->total_num_rows();
    }

    if (block.consume_bytes) {
      RETURN_NOT_OK(block.consume_bytes(parsed_size));
    }
    return ParsedBlock{std::move(parser), block.block_index,
                       static_cast<int64_t>(parsed_size) + block.bytes_skipped};
  }

  int num_csv_cols() const { return num_csv_cols_; }