From 4d147cbedd483e91f6f103c727f2fd9ec6ef2a4c Mon Sep 17 00:00:00 2001 From: Ilias Aberkane Date: Sat, 19 Sep 2026 17:41:46 +0200 Subject: [PATCH] GH-51368: [C++] Fix CSV reader dropping '\n' after a straddling '\r' inside a quoted field With newlines_in_values, a quoted CRLF split across two input buffers was corrupted: CSVBufferIterator unconditionally skipped a leading '\n' whenever the previous buffer ended with '\r', so "xxx|yyy" inside a quoted field silently lost the '\n'. The iterator has no quoting context, so it cannot tell a CRLF line separator from quoted field contents. Move the straddling-CRLF decision into the block readers, where partial_ shows whether the previous '\r' was consumed as a line separator (partial empty -> skip the leading '\n') or is unfinished quoted content (keep it). The initial flag is plumbed through for the case where a header line ends exactly at a buffer boundary with '\r'. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- cpp/src/arrow/csv/reader.cc | 110 +++++++++++++++++++++++-------- cpp/src/arrow/csv/reader_test.cc | 82 +++++++++++++++++++++-- 2 files changed, 158 insertions(+), 34 deletions(-) diff --git a/cpp/src/arrow/csv/reader.cc b/cpp/src/arrow/csv/reader.cc index 1ae9d27e6bfd..58637f4d5c8b 100644 --- a/cpp/src/arrow/csv/reader.cc +++ b/cpp/src/arrow/csv/reader.cc @@ -89,7 +89,10 @@ struct ConversionSchema { std::vector columns; }; -// An iterator of Buffers that makes sure there is no straddling CRLF sequence. +// An iterator of Buffers that strips a leading UTF-8 BOM (if present). +// Straddling CRLF sequences cannot be resolved at this stage: a '\r' ending a +// buffer may either be a line separator or the contents of an unfinished +// quoted field, and only the block readers further downstream know which. class CSVBufferIterator { public: static Iterator> Make( @@ -120,12 +123,6 @@ class CSVBufferIterator { 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 @@ -137,8 +134,6 @@ class CSVBufferIterator { protected: bool first_buffer_ = true; - // Whether there was a trailing CR at the end of last received buffer - bool trailing_cr_ = false; }; struct CSVBlock { @@ -170,19 +165,40 @@ namespace { // iterator APIs (e.g. Visit)) even though an empty optional is never used in this code. class BlockReader { public: + // `prev_ended_cr` should be true if the raw bytes delivered just before + // `first_buffer` ended with a '\r' (for instance the first block ends right + // after a header line separated by CRLF). It is only meaningful when + // `first_buffer` is empty, since otherwise the pending '\r' is carried in + // `first_buffer` itself. BlockReader(std::unique_ptr chunker, std::shared_ptr first_buffer, - int64_t skip_rows) + int64_t skip_rows, bool prev_ended_cr = false) : chunker_(std::move(chunker)), partial_(std::make_shared("")), buffer_(std::move(first_buffer)), - skip_rows_(skip_rows) {} + skip_rows_(skip_rows), + trailing_cr_(prev_ended_cr && buffer_->size() == 0) {} + + // If the previous buffer ended with a '\r' that was consumed as a line + // separator, a '\n' at the start of `buffer` merely completes that separator + // and must be skipped so that it doesn't produce a spurious row. When the + // '\r' sits inside an unfinished quoted field instead, it is carried over in + // `partial_` (which is then non-empty) and the '\n' is genuine field contents. + static std::shared_ptr SkipStraddlingCRLF(std::shared_ptr buffer, + bool trailing_cr) { + if (trailing_cr && buffer->size() != 0 && buffer->data()[0] == '\n') { + // Skip '\r\n' line separator that started at the end of previous buffer + return SliceBuffer(std::move(buffer), 1); + } + return buffer; + } protected: std::unique_ptr chunker_; std::shared_ptr partial_, buffer_; int64_t skip_rows_; int64_t block_index_ = 0; - // Whether there was a trailing CR at the end of last received buffer + // Whether the last buffer ended with a '\r' that was fully consumed + // (i.e. as a line separator rather than unfinished field contents) bool trailing_cr_ = false; }; @@ -195,9 +211,10 @@ class SerialBlockReader : public BlockReader { static Iterator MakeIterator( Iterator> buffer_iterator, std::unique_ptr chunker, - std::shared_ptr first_buffer, int64_t skip_rows) { - auto block_reader = - std::make_shared(std::move(chunker), first_buffer, skip_rows); + std::shared_ptr first_buffer, int64_t skip_rows, + bool prev_ended_cr = false) { + auto block_reader = std::make_shared( + std::move(chunker), first_buffer, skip_rows, prev_ended_cr); // Wrap shared pointer in callable Transformer, CSVBlock> block_reader_fn = [block_reader](std::shared_ptr buf) { @@ -209,9 +226,9 @@ class SerialBlockReader : public BlockReader { static AsyncGenerator MakeAsyncIterator( AsyncGenerator> buffer_generator, std::unique_ptr chunker, std::shared_ptr first_buffer, - int64_t skip_rows) { - auto block_reader = - std::make_shared(std::move(chunker), first_buffer, skip_rows); + int64_t skip_rows, bool prev_ended_cr = false) { + auto block_reader = std::make_shared( + std::move(chunker), first_buffer, skip_rows, prev_ended_cr); // Wrap shared pointer in callable Transformer, CSVBlock> block_reader_fn = [block_reader](std::shared_ptr next) { @@ -228,6 +245,13 @@ class SerialBlockReader : public BlockReader { bool is_final = (next_buffer == nullptr); int64_t bytes_skipped = 0; + // An empty buffer can't resolve a straddling CRLF, so don't let it + // overwrite the current trailing_cr_ state. + const bool has_bytes = buffer_->size() != 0; + buffer_ = SkipStraddlingCRLF(std::move(buffer_), trailing_cr_); + const bool ends_with_cr = + buffer_->size() != 0 && buffer_->data()[buffer_->size() - 1] == '\r'; + if (skip_rows_) { bytes_skipped += partial_->size(); auto orig_size = buffer_->size(); @@ -237,6 +261,9 @@ class SerialBlockReader : public BlockReader { auto empty = std::make_shared(nullptr, 0); if (skip_rows_) { // Still have rows beyond this buffer to skip return empty block + if (has_bytes) { + trailing_cr_ = ends_with_cr && buffer_->size() == 0; + } partial_ = std::move(buffer_); buffer_ = next_buffer; return TransformYield(CSVBlock{empty, empty, empty, block_index_++, @@ -258,14 +285,17 @@ class SerialBlockReader : public BlockReader { } int64_t bytes_before_buffer = partial_->size() + completion->size(); - auto consume_bytes = [this, bytes_before_buffer, - next_buffer](int64_t nbytes) -> Status { + auto consume_bytes = [this, bytes_before_buffer, next_buffer, has_bytes, + ends_with_cr](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); + if (has_bytes) { + trailing_cr_ = ends_with_cr && partial_->size() == 0; + } buffer_ = next_buffer; return Status::OK(); }; @@ -284,9 +314,9 @@ class ThreadedBlockReader : public BlockReader { static AsyncGenerator MakeAsyncIterator( AsyncGenerator> buffer_generator, std::unique_ptr chunker, std::shared_ptr first_buffer, - int64_t skip_rows) { - auto block_reader = std::make_shared(std::move(chunker), - first_buffer, skip_rows); + int64_t skip_rows, bool prev_ended_cr = false) { + auto block_reader = std::make_shared( + std::move(chunker), first_buffer, skip_rows, prev_ended_cr); // Wrap shared pointer in callable Transformer, CSVBlock> block_reader_fn = [block_reader](std::shared_ptr next) { return (*block_reader)(next); }; @@ -305,6 +335,13 @@ class ThreadedBlockReader : public BlockReader { auto current_buffer = std::move(buffer_); int64_t bytes_skipped = 0; + // An empty buffer can't resolve a straddling CRLF, so don't let it + // overwrite the current trailing_cr_ state. + const bool has_bytes = current_buffer->size() != 0; + current_buffer = SkipStraddlingCRLF(std::move(current_buffer), trailing_cr_); + const bool ends_with_cr = current_buffer->size() != 0 && + current_buffer->data()[current_buffer->size() - 1] == '\r'; + if (skip_rows_) { auto orig_size = current_buffer->size(); bytes_skipped = current_partial->size(); @@ -313,6 +350,9 @@ class ThreadedBlockReader : public BlockReader { bytes_skipped += orig_size - current_buffer->size(); current_partial = std::make_shared(nullptr, 0); if (skip_rows_) { + if (has_bytes) { + trailing_cr_ = ends_with_cr && current_buffer->size() == 0; + } partial_ = std::move(current_buffer); buffer_ = std::move(next_buffer); return TransformYield(CSVBlock{current_partial, @@ -341,6 +381,9 @@ class ThreadedBlockReader : public BlockReader { // 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)); + if (has_bytes) { + trailing_cr_ = ends_with_cr && next_partial->size() == 0; + } } partial_ = std::move(next_partial); @@ -890,6 +933,9 @@ class StreamingReaderImpl : public ReaderMixin, return Status::Invalid("Empty CSV file"); } + const bool first_ended_with_cr = + first_buffer->size() != 0 && + first_buffer->data()[first_buffer->size() - 1] == '\r'; std::shared_ptr after_header; ARROW_ASSIGN_OR_RAISE(auto header_bytes_consumed, ProcessHeader(first_buffer, &after_header)); @@ -901,7 +947,7 @@ class StreamingReaderImpl : public ReaderMixin, auto block_gen = SerialBlockReader::MakeAsyncIterator( std::move(buffer_generator), MakeChunker(parse_options_), std::move(after_header), - read_options_.skip_rows_after_names); + read_options_.skip_rows_after_names, first_ended_with_cr); auto parsed_block_gen = MakeMappedGenerator(std::move(block_gen), *parsing_operator_); auto rb_gen = MakeMappedGenerator(std::move(parsed_block_gen), std::move(decoder_op)); @@ -992,12 +1038,15 @@ class SerialTableReader : public BaseTableReader { if (first_buffer == nullptr) { return Status::Invalid("Empty CSV file"); } + const bool first_ended_with_cr = + first_buffer->size() != 0 && + first_buffer->data()[first_buffer->size() - 1] == '\r'; RETURN_NOT_OK(ProcessHeader(first_buffer, &first_buffer)); RETURN_NOT_OK(MakeColumnBuilders()); auto block_iterator = SerialBlockReader::MakeIterator( std::move(buffer_iterator_), MakeChunker(parse_options_), std::move(first_buffer), - read_options_.skip_rows_after_names); + read_options_.skip_rows_after_names, first_ended_with_cr); while (true) { RETURN_NOT_OK(io_context_.stop_token().Poll()); @@ -1064,7 +1113,7 @@ class AsyncThreadedTableReader return ProcessFirstBuffer().Then([self](const std::shared_ptr& first_buffer) { auto block_generator = ThreadedBlockReader::MakeAsyncIterator( self->buffer_generator_, MakeChunker(self->parse_options_), first_buffer, - self->read_options_.skip_rows_after_names); + self->read_options_.skip_rows_after_names, self->first_buffer_ended_cr_); std::function block_visitor = [self](CSVBlock maybe_block) -> Status { @@ -1103,6 +1152,9 @@ class AsyncThreadedTableReader if (first_buffer == nullptr) { return Status::Invalid("Empty CSV file"); } + self->first_buffer_ended_cr_ = + first_buffer->size() != 0 && + first_buffer->data()[first_buffer->size() - 1] == '\r'; std::shared_ptr first_buffer_processed; RETURN_NOT_OK(self->ProcessHeader(first_buffer, &first_buffer_processed)); RETURN_NOT_OK(self->MakeColumnBuilders()); @@ -1112,6 +1164,7 @@ class AsyncThreadedTableReader Executor* cpu_executor_; AsyncGenerator> buffer_generator_; + bool first_buffer_ended_cr_ = false; }; Result> MakeTableReader( @@ -1185,10 +1238,13 @@ class CSVRowCounter : public ReaderMixin, if (!first_buffer) { return Status::Invalid("Empty CSV file"); } + const bool first_ended_with_cr = + first_buffer->size() != 0 && + first_buffer->data()[first_buffer->size() - 1] == '\r'; RETURN_NOT_OK(self->ProcessHeader(first_buffer, &first_buffer)); self->block_generator_ = SerialBlockReader::MakeAsyncIterator( buffer_generator, MakeChunker(self->parse_options_), - std::move(first_buffer), 0); + std::move(first_buffer), 0, first_ended_with_cr); return Status::OK(); }); } diff --git a/cpp/src/arrow/csv/reader_test.cc b/cpp/src/arrow/csv/reader_test.cc index 009bbd5fc25a..e2773581d025 100644 --- a/cpp/src/arrow/csv/reader_test.cc +++ b/cpp/src/arrow/csv/reader_test.cc @@ -29,6 +29,7 @@ #include #include +#include "arrow/array/array_binary.h" #include "arrow/array/array_dict.h" #include "arrow/csv/options.h" #include "arrow/csv/test_common.h" @@ -135,6 +136,65 @@ void TestStraddling(TableReaderFactory reader_factory) { internal::GetCpuThreadPool()->WaitForIdle(); } +void TestStraddlingCRLF(TableReaderFactory reader_factory) { + // GH-51368: when a CRLF sequence straddles a block boundary, the '\n' must be + // skipped only if the '\r' was a line separator. If the '\r' is the contents + // of an unfinished quoted field, the '\n' is genuine field contents. + ParseOptions options = ParseOptions::Defaults(); + options.newlines_in_values = true; + + constexpr int32_t kBlockSize = 1024; + { + // '\r' inside a quoted field as the last byte of the first block + std::string csv = "x,y\n\""; + csv += std::string(kBlockSize - csv.size() - 2, 'z'); + csv += "A\r\nB\",c\n1,plain\n"; + ASSERT_EQ('\r', csv[static_cast(kBlockSize) - 1]); + ASSERT_EQ('\n', csv[kBlockSize]); + + auto input = std::make_shared(std::make_shared(csv)); + ASSERT_OK_AND_ASSIGN(auto reader, + reader_factory(input, options, /*block_size=*/kBlockSize)); + ASSERT_FINISHES_OK_AND_ASSIGN(auto table, reader->ReadAsync()); + + ASSERT_EQ(2, table->num_rows()); + const auto& col0 = + internal::checked_cast(*table->column(0)->chunk(0)); + ASSERT_EQ(std::string(kBlockSize - 7, 'z') + "A\r\nB", col0.GetString(0)); + ASSERT_EQ("1", col0.GetString(1)); + } + { + // '\r' as a line separator as the last byte of the first block + std::string csv = "x,y\na," + std::string(kBlockSize - 7, 'q') + "\r\n3,4\n"; + ASSERT_EQ('\r', csv[static_cast(kBlockSize) - 1]); + ASSERT_EQ('\n', csv[kBlockSize]); + + auto input = std::make_shared(std::make_shared(csv)); + ASSERT_OK_AND_ASSIGN(auto reader, + reader_factory(input, options, /*block_size=*/kBlockSize)); + ASSERT_FINISHES_OK_AND_ASSIGN(auto table, reader->ReadAsync()); + + ASSERT_EQ(2, table->num_rows()); + const auto& col0 = + internal::checked_cast(*table->column(0)->chunk(0)); + ASSERT_EQ("a", col0.GetString(0)); + ASSERT_EQ("3", col0.GetString(1)); + } + { + // Header row ending with '\r' as the last byte of the first block + std::string csv = "x," + std::string(kBlockSize - 3, 'h') + "\r\na,b\n"; + ASSERT_EQ('\r', csv[static_cast(kBlockSize) - 1]); + ASSERT_EQ('\n', csv[kBlockSize]); + + auto input = std::make_shared(std::make_shared(csv)); + ASSERT_OK_AND_ASSIGN(auto reader, + reader_factory(input, options, /*block_size=*/kBlockSize)); + ASSERT_FINISHES_OK_AND_ASSIGN(auto table, reader->ReadAsync()); + + ASSERT_EQ(1, table->num_rows()); + } +} + void StressTableReader(TableReaderFactory reader_factory) { #ifdef ARROW_VALGRIND const int NTASKS = 10; @@ -239,7 +299,7 @@ void TestInvalidRowsSkipped(TableReaderFactory reader_factory, bool async) { auto input = std::make_shared(table_buffer); ASSERT_OK_AND_ASSIGN(auto reader, reader_factory(input, std::move(opts), /*block_size=*/{})); - ASSERT_OK_AND_ASSIGN(auto table, reader->Read()); + ASSERT_FINISHES_OK_AND_ASSIGN(auto table, reader->ReadAsync()); ASSERT_EQ(NROWS - NINVALID, table->num_rows()); ASSERT_EQ(NINVALID, num_invalid_rows); } @@ -258,6 +318,7 @@ TableReaderFactory MakeSerialFactory() { TEST(SerialReaderTests, Empty) { TestEmptyTable(MakeSerialFactory()); } TEST(SerialReaderTests, HeaderOnly) { TestHeaderOnly(MakeSerialFactory()); } TEST(SerialReaderTests, Straddling) { TestStraddling(MakeSerialFactory()); } +TEST(SerialReaderTests, StraddlingCRLF) { TestStraddlingCRLF(MakeSerialFactory()); } TEST(SerialReaderTests, Stress) { StressTableReader(MakeSerialFactory()); } TEST(SerialReaderTests, StressInvalid) { StressInvalidTableReader(MakeSerialFactory()); } TEST(SerialReaderTests, NestedParallelism) { @@ -298,6 +359,10 @@ TEST(AsyncReaderTests, Straddling) { ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory()); TestStraddling(table_factory); } +TEST(AsyncReaderTests, StraddlingCRLF) { + ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory()); + TestStraddlingCRLF(table_factory); +} TEST(AsyncReaderTests, Stress) { ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory()); StressTableReader(table_factory); @@ -348,6 +413,9 @@ TEST(StreamingReaderTests, HeaderOnly) { TestHeaderOnlyStreaming(table_factory); } TEST(StreamingReaderTests, Straddling) { TestStraddling(MakeStreamingFactory()); } +TEST(StreamingReaderTests, StraddlingCRLF) { + TestStraddlingCRLF(MakeStreamingFactory()); +} TEST(StreamingReaderTests, Stress) { StressTableReader(MakeStreamingFactory()); } TEST(StreamingReaderTests, StressInvalid) { StressInvalidTableReader(MakeStreamingFactory()); @@ -550,7 +618,7 @@ TEST(ReaderTests, DefaultColumnTypePartialDefault) { ASSERT_OK_AND_ASSIGN(auto reader, TableReader::Make(io::default_io_context(), input, read_options, parse_options, convert_options)); - ASSERT_OK_AND_ASSIGN(auto table, reader->Read()); + ASSERT_FINISHES_OK_AND_ASSIGN(auto table, reader->ReadAsync()); auto expected_schema = schema({field("id", int64()), field("name", utf8()), field("value", utf8()), field("date", utf8())}); @@ -579,7 +647,7 @@ TEST(ReaderTests, DefaultColumnTypeForcesTypedColumns) { ASSERT_OK_AND_ASSIGN(auto reader, TableReader::Make(io::default_io_context(), input, read_options, parse_options, convert_options)); - ASSERT_OK_AND_ASSIGN(auto table, reader->Read()); + ASSERT_FINISHES_OK_AND_ASSIGN(auto table, reader->ReadAsync()); auto expected_schema = schema({field("id", utf8()), field("amount", utf8()), field("code", utf8())}); @@ -606,7 +674,7 @@ TEST(ReaderTests, DefaultColumnTypeAllStringsNoHeader) { ASSERT_OK_AND_ASSIGN(auto reader, TableReader::Make(io::default_io_context(), input, read_options, parse_options, convert_options)); - ASSERT_OK_AND_ASSIGN(auto table, reader->Read()); + ASSERT_FINISHES_OK_AND_ASSIGN(auto table, reader->ReadAsync()); auto expected_schema = schema({field("f0", utf8()), field("f1", utf8()), field("f2", utf8())}); @@ -635,7 +703,7 @@ TEST(ReaderTests, ShortRows) { ASSERT_OK_AND_ASSIGN(auto reader, TableReader::Make(io::default_io_context(), input, read_options, parse_options, convert_options)); - ASSERT_OK_AND_ASSIGN(auto table, reader->Read()); + ASSERT_FINISHES_OK_AND_ASSIGN(auto table, reader->ReadAsync()); auto expected_schema = schema({field("a", utf8()), field("b", utf8()), field("c", utf8())}); @@ -655,7 +723,7 @@ TEST(ReaderTests, IgnoreExtraColumns) { ASSERT_OK_AND_ASSIGN(auto reader, TableReader::Make(io::default_io_context(), input, ReadOptions::Defaults(), parse_options, convert_options)); - ASSERT_OK_AND_ASSIGN(auto table, reader->Read()); + ASSERT_FINISHES_OK_AND_ASSIGN(auto table, reader->ReadAsync()); auto expected_schema = schema({field("a", int64()), field("b", int64())}); auto expected_table = TableFromJSON(expected_schema, {R"([ @@ -677,7 +745,7 @@ TEST(ReaderTests, ShortRowsTypedConverters) { ASSERT_OK_AND_ASSIGN(auto reader, TableReader::Make(io::default_io_context(), input, read_options, parse_options, convert_options)); - ASSERT_OK_AND_ASSIGN(auto table, reader->Read()); + ASSERT_FINISHES_OK_AND_ASSIGN(auto table, reader->ReadAsync()); ASSERT_TRUE(table->column(0)->chunk(0)->Equals(*ArrayFromJSON(int64(), "[1, 2]"))); const auto& dict_array = internal::checked_cast(*table->column(1)->chunk(0));