From abeddb857dadcd130644bf4073b55f94287d9016 Mon Sep 17 00:00:00 2001 From: Aditya Raj Date: Sat, 17 Oct 2026 19:25:33 +0530 Subject: [PATCH 1/3] GH-51368: [C++] Fix silent LF drop when quoted CRLF straddles CSV block boundary --- cpp/src/arrow/csv/reader.cc | 56 +++++++++++++++++++++++++++----- cpp/src/arrow/csv/reader_test.cc | 49 ++++++++++++++++++++++++++++ 2 files changed, 96 insertions(+), 9 deletions(-) diff --git a/cpp/src/arrow/csv/reader.cc b/cpp/src/arrow/csv/reader.cc index 1ae9d27e6bfd..a40b60c15ce3 100644 --- a/cpp/src/arrow/csv/reader.cc +++ b/cpp/src/arrow/csv/reader.cc @@ -109,6 +109,16 @@ class CSVBufferIterator { Result>> operator()(std::shared_ptr buf) { if (buf == nullptr) { // EOF + if (pending_cr_) { + // A '\r' was withheld from the end of the previous buffer. Emit it + // now so that no data is lost, and ask to be called again with the + // end-of-stream marker so that we can then finish the stream. + pending_cr_ = false; + ARROW_ASSIGN_OR_RAISE(auto cr, AllocateBuffer(1)); + cr->mutable_data()[0] = '\r'; + return TransformYield(std::shared_ptr(std::move(cr)), + /*ready_for_next=*/false); + } return TransformFinish(); } @@ -120,25 +130,53 @@ 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; + if (pending_cr_) { + // Reattach the '\r' withheld from the end of the previous buffer to the + // start of this one. This keeps a CRLF pair straddling the boundary + // contiguous inside a single buffer, where both the chunker and the + // parser handle it correctly -- whether it is a line separator or data + // inside a quoted field (GH-51368). + pending_cr_ = false; + const int64_t rest = buf->size() - offset; + ARROW_ASSIGN_OR_RAISE(auto joined, AllocateBuffer(1 + rest)); + uint8_t* out = joined->mutable_data(); + out[0] = '\r'; + if (rest > 0) { + memcpy(out + 1, buf->data() + offset, static_cast(rest)); + } + buf = std::shared_ptr(std::move(joined)); + offset = 0; + } + + if (buf->size() > offset && buf->data()[buf->size() - 1] == '\r') { + // Withhold a trailing '\r' so that downstream code never observes a + // buffer ending with '\r'. The chunker and the parser treat a '\r' at + // the end of a buffer as a complete line terminator; dropping the '\n' + // that may follow it (as was done previously) silently corrupts CRLF + // sequences inside quoted fields (GH-51368). + pending_cr_ = true; + buf = SliceBuffer(std::move(buf), offset, buf->size() - offset - 1); + } else if (offset != 0) { + buf = SliceBuffer(std::move(buf), offset); } - trailing_cr_ = (buf->data()[buf->size() - 1] == '\r'); - buf = SliceBuffer(std::move(buf), offset); if (buf->size() == 0) { + if (pending_cr_) { + // The withheld '\r' was the entire buffer contents; skip this empty + // buffer and wait for more data (or EOF, which flushes the '\r'). + return TransformSkip(); + } // EOF return TransformFinish(); - } else { - return TransformYield(std::move(buf)); } + 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; + // Whether a trailing '\r' was withheld from the end of the last emitted + // buffer, to be prepended to the next one (GH-51368) + bool pending_cr_ = false; }; struct CSVBlock { diff --git a/cpp/src/arrow/csv/reader_test.cc b/cpp/src/arrow/csv/reader_test.cc index 009bbd5fc25a..f710a3a3813a 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.h" #include "arrow/array/array_dict.h" #include "arrow/csv/options.h" #include "arrow/csv/test_common.h" @@ -135,6 +136,44 @@ void TestStraddling(TableReaderFactory reader_factory) { internal::GetCpuThreadPool()->WaitForIdle(); } +void TestQuotedCRLFStraddlingBlockBoundary(TableReaderFactory reader_factory) { + // GH-51368: a CRLF inside a quoted field must survive intact when the CR is + // the last byte of a read block and the LF is the first byte of the next + // block. Previously the LF was silently dropped. + constexpr int32_t kBlockSize = 1024; + const std::string prefix = "id,s,p\n0,\""; + const std::string embedded = "A\r\nB"; + const std::string suffix = "\",x\n1,plain,x\n2,plain,x\n"; + // Place 'A' at byte kBlockSize - 2 and '\r' at the last byte of the block + const int32_t pad = kBlockSize - static_cast(prefix.size()) - 2; + const std::string first_value = std::string(static_cast(pad), 'z') + embedded; + const std::string data = + prefix + std::string(static_cast(pad), 'z') + embedded + suffix; + + auto parse_options = ParseOptions::Defaults(); + parse_options.newlines_in_values = true; + + auto input = std::make_shared(std::make_shared(data)); + ASSERT_OK_AND_ASSIGN(auto reader, reader_factory(input, parse_options, kBlockSize)); + ASSERT_FINISHES_OK_AND_ASSIGN(auto table, reader->ReadAsync()); + + ASSERT_EQ(table->num_columns(), 3); + ASSERT_EQ(table->num_rows(), 3); + auto column = table->GetColumnByName("s"); + ASSERT_NE(column, nullptr); + std::vector values; + for (int i = 0; i < column->num_chunks(); ++i) { + auto arr = std::static_pointer_cast(column->chunk(i)); + for (int64_t j = 0; j < arr->length(); ++j) { + values.push_back(arr->GetString(j)); + } + } + ASSERT_EQ(values.size(), 3); + EXPECT_EQ(values[0], first_value); + EXPECT_EQ(values[1], "plain"); + EXPECT_EQ(values[2], "plain"); +} + void StressTableReader(TableReaderFactory reader_factory) { #ifdef ARROW_VALGRIND const int NTASKS = 10; @@ -258,6 +297,9 @@ TableReaderFactory MakeSerialFactory() { TEST(SerialReaderTests, Empty) { TestEmptyTable(MakeSerialFactory()); } TEST(SerialReaderTests, HeaderOnly) { TestHeaderOnly(MakeSerialFactory()); } TEST(SerialReaderTests, Straddling) { TestStraddling(MakeSerialFactory()); } +TEST(SerialReaderTests, QuotedCRLFStraddlingBlockBoundary) { + TestQuotedCRLFStraddlingBlockBoundary(MakeSerialFactory()); +} TEST(SerialReaderTests, Stress) { StressTableReader(MakeSerialFactory()); } TEST(SerialReaderTests, StressInvalid) { StressInvalidTableReader(MakeSerialFactory()); } TEST(SerialReaderTests, NestedParallelism) { @@ -294,6 +336,10 @@ TEST(AsyncReaderTests, HeaderOnly) { ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory()); TestHeaderOnly(table_factory); } +TEST(AsyncReaderTests, QuotedCRLFStraddlingBlockBoundary) { + ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory()); + TestQuotedCRLFStraddlingBlockBoundary(table_factory); +} TEST(AsyncReaderTests, Straddling) { ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory()); TestStraddling(table_factory); @@ -348,6 +394,9 @@ TEST(StreamingReaderTests, HeaderOnly) { TestHeaderOnlyStreaming(table_factory); } TEST(StreamingReaderTests, Straddling) { TestStraddling(MakeStreamingFactory()); } +TEST(StreamingReaderTests, QuotedCRLFStraddlingBlockBoundary) { + TestQuotedCRLFStraddlingBlockBoundary(MakeStreamingFactory()); +} TEST(StreamingReaderTests, Stress) { StressTableReader(MakeStreamingFactory()); } TEST(StreamingReaderTests, StressInvalid) { StressInvalidTableReader(MakeStreamingFactory()); From fdd0bfd82abe8c47123d8b882e63f8ec4c2ea28a Mon Sep 17 00:00:00 2001 From: Aditya Raj Date: Thu, 22 Oct 2026 22:24:55 +0530 Subject: [PATCH 2/3] GH-51368: [C++] Address review comments: fix use-after-move, preserve first-row terminator at block boundary --- cpp/src/arrow/csv/reader.cc | 45 +++++++++++++++++++++++++- cpp/src/arrow/csv/reader_test.cc | 54 ++++++++++++++++++++++++++++++++ 2 files changed, 98 insertions(+), 1 deletion(-) diff --git a/cpp/src/arrow/csv/reader.cc b/cpp/src/arrow/csv/reader.cc index a40b60c15ce3..0028b717c750 100644 --- a/cpp/src/arrow/csv/reader.cc +++ b/cpp/src/arrow/csv/reader.cc @@ -109,6 +109,13 @@ class CSVBufferIterator { Result>> operator()(std::shared_ptr buf) { if (buf == nullptr) { // EOF + if (held_first_buffer_ != nullptr) { + // The whole input was a single buffer ending with '\r', which was + // held back (see below). Emit it as-is: no buffer follows, so its + // trailing '\r' cannot be the first half of a straddling CRLF pair. + auto held = std::move(held_first_buffer_); + return TransformYield(std::move(held), /*ready_for_next=*/false); + } if (pending_cr_) { // A '\r' was withheld from the end of the previous buffer. Emit it // now so that no data is lost, and ask to be called again with the @@ -123,6 +130,7 @@ class CSVBufferIterator { } int64_t offset = 0; + const bool is_first_buffer = first_buffer_; if (first_buffer_) { ARROW_ASSIGN_OR_RAISE(auto data, util::SkipUTF8BOM(buf->data(), buf->size())); offset += data - buf->data(); @@ -148,14 +156,46 @@ class CSVBufferIterator { offset = 0; } + if (held_first_buffer_ != nullptr) { + // Prepend the first buffer, which was held back because it ended with + // '\r' (see below), so that the first emitted buffer contains the end + // of the first line whenever it falls inside the first read block. + auto held = std::move(held_first_buffer_); + const int64_t rest = buf->size() - offset; + ARROW_ASSIGN_OR_RAISE(auto joined, AllocateBuffer(held->size() + rest)); + uint8_t* out = joined->mutable_data(); + memcpy(out, held->data(), static_cast(held->size())); + if (rest > 0) { + memcpy(out + held->size(), buf->data() + offset, static_cast(rest)); + } + buf = std::shared_ptr(std::move(joined)); + offset = 0; + } + if (buf->size() > offset && buf->data()[buf->size() - 1] == '\r') { + if (is_first_buffer) { + // Do not withhold a trailing '\r' from the first buffer: the header + // parser (ReaderMixin::ProcessHeader) parses the first buffer on + // its own and needs to find the end of the first line in it, so + // emitting the buffer without its only line terminator makes the + // header unreadable (e.g. block_size=4 on "a,b\r1,2\r"). Hold the + // whole buffer back instead and prepend it to the next one above; + // the '\r' then sits inside the emitted buffer, where downstream + // code handles it like any other line terminator. + held_first_buffer_ = SliceBuffer(buf, offset); + return TransformSkip(); + } // Withhold a trailing '\r' so that downstream code never observes a // buffer ending with '\r'. The chunker and the parser treat a '\r' at // the end of a buffer as a complete line terminator; dropping the '\n' // that may follow it (as was done previously) silently corrupts CRLF // sequences inside quoted fields (GH-51368). pending_cr_ = true; - buf = SliceBuffer(std::move(buf), offset, buf->size() - offset - 1); + // Compute the slice length before moving `buf`: the evaluation order + // of function arguments is unspecified, so `buf->size()` could + // otherwise be read after `buf` has been moved from. + const int64_t length = buf->size() - offset - 1; + buf = SliceBuffer(std::move(buf), offset, length); } else if (offset != 0) { buf = SliceBuffer(std::move(buf), offset); } @@ -177,6 +217,9 @@ class CSVBufferIterator { // Whether a trailing '\r' was withheld from the end of the last emitted // buffer, to be prepended to the next one (GH-51368) bool pending_cr_ = false; + // The first buffer, if it ended with '\r' and was held back to be + // prepended to the second buffer (or emitted as-is at end of stream) + std::shared_ptr held_first_buffer_; }; struct CSVBlock { diff --git a/cpp/src/arrow/csv/reader_test.cc b/cpp/src/arrow/csv/reader_test.cc index f710a3a3813a..654fdba66201 100644 --- a/cpp/src/arrow/csv/reader_test.cc +++ b/cpp/src/arrow/csv/reader_test.cc @@ -174,6 +174,50 @@ void TestQuotedCRLFStraddlingBlockBoundary(TableReaderFactory reader_factory) { EXPECT_EQ(values[2], "plain"); } +void TestHeaderRowEndingAtBlockBoundary(TableReaderFactory reader_factory) { + // The header row's line terminator is the last byte of the first read + // block. ProcessHeader parses the first buffer on its own, so the buffer + // handed to it must still contain the end of the header line; withholding + // a trailing '\r' from the first buffer (as done for later buffers to fix + // GH-51368) makes the header unreadable here. + for (const std::string& data : + {std::string("a,b\r1,2\r"), std::string("a,b\r\n1,2\r\n")}) { + // block_size=4 splits the input right after the header terminator; + // block_size=64 makes it a single block (also exercising the case of a + // lone first buffer ending with '\r'). + for (const int32_t block_size : {4, 64}) { + auto input = std::make_shared(std::make_shared(data)); + ASSERT_OK_AND_ASSIGN(auto reader, + reader_factory(input, ParseOptions::Defaults(), block_size)); + ASSERT_FINISHES_OK_AND_ASSIGN(auto table, reader->ReadAsync()); + + ASSERT_EQ(table->num_columns(), 2); + ASSERT_EQ(table->num_rows(), 1); + auto col_a = table->GetColumnByName("a"); + auto col_b = table->GetColumnByName("b"); + ASSERT_NE(col_a, nullptr); + ASSERT_NE(col_b, nullptr); + std::vector values_a, values_b; + for (int i = 0; i < col_a->num_chunks(); ++i) { + auto arr = std::static_pointer_cast(col_a->chunk(i)); + for (int64_t j = 0; j < arr->length(); ++j) { + values_a.push_back(arr->Value(j)); + } + } + for (int i = 0; i < col_b->num_chunks(); ++i) { + auto arr = std::static_pointer_cast(col_b->chunk(i)); + for (int64_t j = 0; j < arr->length(); ++j) { + values_b.push_back(arr->Value(j)); + } + } + ASSERT_EQ(values_a.size(), 1); + ASSERT_EQ(values_b.size(), 1); + EXPECT_EQ(values_a[0], 1); + EXPECT_EQ(values_b[0], 2); + } + } +} + void StressTableReader(TableReaderFactory reader_factory) { #ifdef ARROW_VALGRIND const int NTASKS = 10; @@ -300,6 +344,9 @@ TEST(SerialReaderTests, Straddling) { TestStraddling(MakeSerialFactory()); } TEST(SerialReaderTests, QuotedCRLFStraddlingBlockBoundary) { TestQuotedCRLFStraddlingBlockBoundary(MakeSerialFactory()); } +TEST(SerialReaderTests, HeaderRowEndingAtBlockBoundary) { + TestHeaderRowEndingAtBlockBoundary(MakeSerialFactory()); +} TEST(SerialReaderTests, Stress) { StressTableReader(MakeSerialFactory()); } TEST(SerialReaderTests, StressInvalid) { StressInvalidTableReader(MakeSerialFactory()); } TEST(SerialReaderTests, NestedParallelism) { @@ -340,6 +387,10 @@ TEST(AsyncReaderTests, QuotedCRLFStraddlingBlockBoundary) { ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory()); TestQuotedCRLFStraddlingBlockBoundary(table_factory); } +TEST(AsyncReaderTests, HeaderRowEndingAtBlockBoundary) { + ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory()); + TestHeaderRowEndingAtBlockBoundary(table_factory); +} TEST(AsyncReaderTests, Straddling) { ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory()); TestStraddling(table_factory); @@ -397,6 +448,9 @@ TEST(StreamingReaderTests, Straddling) { TestStraddling(MakeStreamingFactory()); TEST(StreamingReaderTests, QuotedCRLFStraddlingBlockBoundary) { TestQuotedCRLFStraddlingBlockBoundary(MakeStreamingFactory()); } +TEST(StreamingReaderTests, HeaderRowEndingAtBlockBoundary) { + TestHeaderRowEndingAtBlockBoundary(MakeStreamingFactory()); +} TEST(StreamingReaderTests, Stress) { StressTableReader(MakeStreamingFactory()); } TEST(StreamingReaderTests, StressInvalid) { StressInvalidTableReader(MakeStreamingFactory()); From a81ba59486697836f3a636821f9864c745ca0e7a Mon Sep 17 00:00:00 2001 From: Aditya Raj Date: Fri, 23 Oct 2026 03:54:34 +0530 Subject: [PATCH 3/3] GH-51368: [C++] Hold CR-terminated blocks whole in CSVBufferIterator; add block-boundary test matrix --- cpp/src/arrow/csv/reader.cc | 143 ++++++++--------- cpp/src/arrow/csv/reader_test.cc | 255 +++++++++++++++++++++++++++++++ 2 files changed, 328 insertions(+), 70 deletions(-) diff --git a/cpp/src/arrow/csv/reader.cc b/cpp/src/arrow/csv/reader.cc index 0028b717c750..b4b662ceb3ea 100644 --- a/cpp/src/arrow/csv/reader.cc +++ b/cpp/src/arrow/csv/reader.cc @@ -109,12 +109,16 @@ class CSVBufferIterator { Result>> operator()(std::shared_ptr buf) { if (buf == nullptr) { // EOF - if (held_first_buffer_ != nullptr) { - // The whole input was a single buffer ending with '\r', which was - // held back (see below). Emit it as-is: no buffer follows, so its - // trailing '\r' cannot be the first half of a straddling CRLF pair. - auto held = std::move(held_first_buffer_); - return TransformYield(std::move(held), /*ready_for_next=*/false); + if (held_buffer_ != nullptr) { + // The last block ended with '\r' and was held back (see below). + // Emit it together with its trailing '\r': no block follows, so + // the '\r' cannot be the first half of a straddling CRLF pair. + auto held = std::move(held_buffer_); + ARROW_ASSIGN_OR_RAISE(auto out, AllocateBuffer(held->size() + 1)); + memcpy(out->mutable_data(), held->data(), static_cast(held->size())); + out->mutable_data()[held->size()] = '\r'; + return TransformYield(std::shared_ptr(std::move(out)), + /*ready_for_next=*/false); } if (pending_cr_) { // A '\r' was withheld from the end of the previous buffer. Emit it @@ -130,7 +134,6 @@ class CSVBufferIterator { } int64_t offset = 0; - const bool is_first_buffer = first_buffer_; if (first_buffer_) { ARROW_ASSIGN_OR_RAISE(auto data, util::SkipUTF8BOM(buf->data(), buf->size())); offset += data - buf->data(); @@ -138,78 +141,77 @@ class CSVBufferIterator { first_buffer_ = false; } - if (pending_cr_) { - // Reattach the '\r' withheld from the end of the previous buffer to the - // start of this one. This keeps a CRLF pair straddling the boundary - // contiguous inside a single buffer, where both the chunker and the - // parser handle it correctly -- whether it is a line separator or data - // inside a quoted field (GH-51368). - pending_cr_ = false; - const int64_t rest = buf->size() - offset; - ARROW_ASSIGN_OR_RAISE(auto joined, AllocateBuffer(1 + rest)); + // Fuse this block with whatever was withheld from the previous one, so + // that a CRLF pair straddling a block boundary stays contiguous inside + // a single buffer, where both the chunker and the parser handle it + // correctly -- whether it is a line separator or data inside a quoted + // field (GH-51368). + const int64_t incoming = buf->size() - offset; + bool fused = false; + std::shared_ptr current; + if (held_buffer_ != nullptr) { + auto held = std::move(held_buffer_); + ARROW_ASSIGN_OR_RAISE(auto joined, AllocateBuffer(held->size() + 1 + incoming)); uint8_t* out = joined->mutable_data(); - out[0] = '\r'; - if (rest > 0) { - memcpy(out + 1, buf->data() + offset, static_cast(rest)); + memcpy(out, held->data(), static_cast(held->size())); + out[held->size()] = '\r'; + if (incoming > 0) { + memcpy(out + held->size() + 1, buf->data() + offset, + static_cast(incoming)); } - buf = std::shared_ptr(std::move(joined)); - offset = 0; - } - - if (held_first_buffer_ != nullptr) { - // Prepend the first buffer, which was held back because it ended with - // '\r' (see below), so that the first emitted buffer contains the end - // of the first line whenever it falls inside the first read block. - auto held = std::move(held_first_buffer_); - const int64_t rest = buf->size() - offset; - ARROW_ASSIGN_OR_RAISE(auto joined, AllocateBuffer(held->size() + rest)); + current = std::shared_ptr(std::move(joined)); + fused = true; + } else if (pending_cr_) { + pending_cr_ = false; + ARROW_ASSIGN_OR_RAISE(auto joined, AllocateBuffer(1 + incoming)); uint8_t* out = joined->mutable_data(); - memcpy(out, held->data(), static_cast(held->size())); - if (rest > 0) { - memcpy(out + held->size(), buf->data() + offset, static_cast(rest)); + out[0] = '\r'; + if (incoming > 0) { + memcpy(out + 1, buf->data() + offset, static_cast(incoming)); } - buf = std::shared_ptr(std::move(joined)); - offset = 0; + current = std::shared_ptr(std::move(joined)); + fused = true; + } else if (offset != 0) { + current = SliceBuffer(buf, offset); + } else { + current = buf; } - if (buf->size() > offset && buf->data()[buf->size() - 1] == '\r') { - if (is_first_buffer) { - // Do not withhold a trailing '\r' from the first buffer: the header - // parser (ReaderMixin::ProcessHeader) parses the first buffer on - // its own and needs to find the end of the first line in it, so - // emitting the buffer without its only line terminator makes the - // header unreadable (e.g. block_size=4 on "a,b\r1,2\r"). Hold the - // whole buffer back instead and prepend it to the next one above; - // the '\r' then sits inside the emitted buffer, where downstream - // code handles it like any other line terminator. - held_first_buffer_ = SliceBuffer(buf, offset); + if (current->size() > 0 && current->data()[current->size() - 1] == '\r') { + if (fused && incoming > 0) { + // The fused buffer still contains a line terminator before its + // last byte (the re-inserted '\r' at the seam, or the one at its + // start), so it is safe to emit it without the trailing '\r' and + // let the next block complete the possible CRLF pair. + pending_cr_ = true; + // Compute the slice length before moving `current`: the + // evaluation order of function arguments is unspecified, so + // `current->size()` could otherwise be read after the move. + const int64_t length = current->size() - 1; + current = SliceBuffer(std::move(current), 0, length); + } else { + // Do not emit this block without its trailing '\r': downstream, + // the chunker requires the block that follows an unterminated row + // to contain a line terminator, and a block stripped of its only + // terminator (e.g. one holding just "z" from a split such as + // "...333,z" + "\r" + "\n") would be rejected with "straddling + // object straddles two block boundaries". Hold the block back + // instead and prepend it to the next one above; it is flushed + // as-is at end of stream. This also covers the first block: the + // first emitted buffer then contains the header line's + // terminator, which ReaderMixin::ProcessHeader relies on when it + // parses the first buffer on its own (e.g. block_size=4 on + // "a,b\r1,2\r"). + held_buffer_ = SliceBuffer(current, 0, current->size() - 1); return TransformSkip(); } - // Withhold a trailing '\r' so that downstream code never observes a - // buffer ending with '\r'. The chunker and the parser treat a '\r' at - // the end of a buffer as a complete line terminator; dropping the '\n' - // that may follow it (as was done previously) silently corrupts CRLF - // sequences inside quoted fields (GH-51368). - pending_cr_ = true; - // Compute the slice length before moving `buf`: the evaluation order - // of function arguments is unspecified, so `buf->size()` could - // otherwise be read after `buf` has been moved from. - const int64_t length = buf->size() - offset - 1; - buf = SliceBuffer(std::move(buf), offset, length); - } else if (offset != 0) { - buf = SliceBuffer(std::move(buf), offset); } - if (buf->size() == 0) { - if (pending_cr_) { - // The withheld '\r' was the entire buffer contents; skip this empty - // buffer and wait for more data (or EOF, which flushes the '\r'). - return TransformSkip(); - } - // EOF + if (current->size() == 0) { + // Nothing to emit (e.g. an input consisting only of a BOM) return TransformFinish(); } - return TransformYield(std::move(buf)); + return TransformYield(std::move(current)); } protected: @@ -217,9 +219,10 @@ class CSVBufferIterator { // Whether a trailing '\r' was withheld from the end of the last emitted // buffer, to be prepended to the next one (GH-51368) bool pending_cr_ = false; - // The first buffer, if it ended with '\r' and was held back to be - // prepended to the second buffer (or emitted as-is at end of stream) - std::shared_ptr held_first_buffer_; + // A block (without its trailing '\r') that ended with '\r' and was held + // back to be prepended to the next block, or flushed at end of stream; + // the '\r' that followed it is re-inserted when fusing + std::shared_ptr held_buffer_; }; struct CSVBlock { diff --git a/cpp/src/arrow/csv/reader_test.cc b/cpp/src/arrow/csv/reader_test.cc index 654fdba66201..c942eab21e43 100644 --- a/cpp/src/arrow/csv/reader_test.cc +++ b/cpp/src/arrow/csv/reader_test.cc @@ -218,6 +218,241 @@ void TestHeaderRowEndingAtBlockBoundary(TableReaderFactory reader_factory) { } } +// An InputStream that delivers its data in prescribed chunk sizes, to +// exercise short reads: a stream is allowed to return fewer bytes than +// requested, so block boundaries as seen by the reader do not necessarily +// fall on multiples of the configured block size. +class ChunkedInputStream : public io::InputStream { + public: + ChunkedInputStream(std::shared_ptr data, std::vector chunks) + : data_(std::move(data)), chunks_(std::move(chunks)) {} + + Status Close() override { + closed_ = true; + return Status::OK(); + } + bool closed() const override { return closed_; } + Result Tell() const override { return pos_; } + Result Read(int64_t nbytes, void* out) override { + int64_t to_read = NextChunk(nbytes); + if (to_read > 0) { + memcpy(out, data_->data() + pos_, static_cast(to_read)); + pos_ += to_read; + } + return to_read; + } + Result> Read(int64_t nbytes) override { + int64_t start = pos_; + int64_t to_read = NextChunk(nbytes); + pos_ += to_read; + return SliceBuffer(data_, start, to_read); + } + + private: + int64_t NextChunk(int64_t nbytes) { + int64_t chunk = chunks_[std::min(chunk_idx_, chunks_.size() - 1)]; + int64_t to_read = std::min({nbytes, chunk, data_->size() - pos_}); + if (to_read == chunk) ++chunk_idx_; + return to_read; + } + std::shared_ptr data_; + std::vector chunks_; + size_t chunk_idx_ = 0; + int64_t pos_ = 0; + bool closed_ = false; +}; + +// Check a parsed two-column table ("a": int64, "b": string) against the +// expected values. Returns "" on success, a description of the mismatch +// otherwise (so that loop-driven tests can report every failing case). +std::string CheckTableContents(const std::shared_ptr& table, + const std::vector& expected_a, + const std::vector& expected_b) { + if (table->num_columns() != 2) { + return "expected 2 columns, got " + std::to_string(table->num_columns()); + } + if (table->num_rows() != static_cast(expected_a.size())) { + return "expected " + std::to_string(expected_a.size()) + " rows, got " + + std::to_string(table->num_rows()); + } + std::vector values_a; + for (int i = 0; i < table->column(0)->num_chunks(); ++i) { + auto arr = std::dynamic_pointer_cast(table->column(0)->chunk(i)); + if (arr == nullptr) { + return "column 'a' is not int64"; + } + for (int64_t j = 0; j < arr->length(); ++j) { + values_a.push_back(arr->Value(j)); + } + } + std::vector values_b; + for (int i = 0; i < table->column(1)->num_chunks(); ++i) { + auto arr = std::dynamic_pointer_cast(table->column(1)->chunk(i)); + if (arr == nullptr) { + return "column 'b' is not a string column"; + } + for (int64_t j = 0; j < arr->length(); ++j) { + values_b.push_back(arr->GetString(j)); + } + } + if (values_a != expected_a) { + return "column 'a' values differ"; + } + if (values_b != expected_b) { + return "column 'b' values differ"; + } + return ""; +} + +// Read a whole table from `input` and check its contents, with the same +// return convention as CheckTableContents. +std::string ReadAndCheckTable(TableReaderFactory reader_factory, + std::shared_ptr input, + const ParseOptions& parse_options, int32_t block_size, + const std::vector& expected_a, + const std::vector& expected_b) { + auto maybe_reader = reader_factory(input, parse_options, block_size); + if (!maybe_reader.ok()) { + return "could not create reader: " + maybe_reader.status().message(); + } + auto maybe_table = (*maybe_reader)->Read(); + if (!maybe_table.ok()) { + return "read failed: " + maybe_table.status().message(); + } + return CheckTableContents(*maybe_table, expected_a, expected_b); +} + +void TestMixedLineEndingsStraddling(TableReaderFactory reader_factory) { + // LF-terminated rows followed by a final CRLF-terminated row, read with + // block_size=8. The last row ("333,z\r\n") is 7 bytes: it must parse + // even when short reads split the stream so that the '\r' arrives in a + // different read than the data it terminates (e.g. chunk sizes 8, 8, 2, 1 + // deliver "...333," / "z\r" / "\n"). Splitting the '\r' from the "z" + // used to make the chunker reject the file with "straddling object + // straddles two block boundaries", even though the row is smaller than + // the block size. + const std::string data = "a,b\n1,x\n2,y\n333,z\r\n"; + const std::vector expected_a = {1, 2, 333}; + const std::vector expected_b = {"x", "y", "z"}; + + // Aligned reads of exactly block_size bytes + { + auto input = std::make_shared(std::make_shared(data)); + ASSERT_EQ("", ReadAndCheckTable(reader_factory, input, ParseOptions::Defaults(), + /*block_size=*/8, expected_a, expected_b)); + } + // Short reads in various chunk patterns (each read still returns at most + // block_size bytes). The first read of each pattern covers the header + // line: the first block handed to the reader must contain its end (the + // header parser reads the first block on its own). + for (const auto& chunks : + {std::vector{8, 8, 2, 1}, std::vector{16, 2, 1}, + std::vector{4, 4, 8, 2, 1}, std::vector{5, 7}}) { + auto input = + std::make_shared(std::make_shared(data), chunks); + ASSERT_EQ("", ReadAndCheckTable(reader_factory, input, ParseOptions::Defaults(), + /*block_size=*/8, expected_a, expected_b)) + << "with chunk pattern starting with " << chunks[0]; + } +} + +void TestSmallInputsAcrossBlockSizes(TableReaderFactory reader_factory) { + // Systematically read small inputs across line-ending styles, presence + // or absence of a final line ending, small block sizes and field styles, + // checking the exact parsed values each time. Block boundaries must not + // change the parse result (GH-51368 and follow-ups): in particular, a + // block must never lose the line terminator of the data it carries. + // + // The block sizes are chosen per field style so that every row still + // completes within the block that follows the one it starts in: the + // chunker legitimately rejects rows straddling more than that ("straddling + // object straddles two block boundaries", see TestStraddling), on the + // unmodified reader too. + const std::vector ending_names = {"LF", "CR", "CRLF"}; + const std::vector line_endings = {"\n", "\r", "\r\n"}; + struct FieldStyle { + std::string name; + std::vector rows; + std::vector expected_a; + std::vector expected_b; + bool newlines_in_values; + std::vector block_sizes; + }; + const std::vector styles = { + {"plain", + {"1,x", "2,y", "33,z"}, + {1, 2, 33}, + {"x", "y", "z"}, + /*newlines_in_values=*/false, + {4, 5, 6, 7, 8, 9, 11, 16}}, + {"quoted", + {"1,\"p\nq\"", "2,\"r\rs\"", "3,\"t\r\nu\""}, + {1, 2, 3}, + {"p\nq", "r\rs", "t\r\nu"}, + /*newlines_in_values=*/true, + {8, 9, 11, 16}}, + }; + + std::vector failures; + auto check = [&](const FieldStyle& style, size_t ending_index, bool final_ending, + const std::string& data, int32_t block_size) { + auto input = std::make_shared(std::make_shared(data)); + auto parse_options = ParseOptions::Defaults(); + parse_options.newlines_in_values = style.newlines_in_values; + auto problem = ReadAndCheckTable(reader_factory, input, parse_options, block_size, + style.expected_a, style.expected_b); + if (!problem.empty()) { + failures.push_back("style=" + style.name + + " endings=" + ending_names[ending_index] + + " final_ending=" + (final_ending ? "yes" : "no") + + " block_size=" + std::to_string(block_size) + ": " + problem); + } + }; + + for (const auto& style : styles) { + for (size_t e = 0; e < line_endings.size(); ++e) { + for (const bool final_ending : {true, false}) { + std::string data = "a,b"; + for (const auto& row : style.rows) { + data += line_endings[e] + row; + } + if (final_ending) { + data += line_endings[e]; + } + for (const int32_t block_size : style.block_sizes) { + check(style, e, final_ending, data, block_size); + } + } + } + } + + // Quoted fields with embedded CRLF at block_size=7: a block can end + // right after an embedded '\r', whose '\n' must not be dropped + // (GH-51368; the unmodified reader silently corrupts the values here). + { + const auto& style = styles[1]; + const size_t e = 2; // CRLF + for (const bool final_ending : {true, false}) { + std::string data = "a,b"; + for (const auto& row : style.rows) { + data += line_endings[e] + row; + } + if (final_ending) { + data += line_endings[e]; + } + check(style, e, final_ending, data, /*block_size=*/7); + } + } + + EXPECT_TRUE(failures.empty()) << [&] { + std::string out; + for (const auto& failure : failures) { + out += "\n " + failure; + } + return out; + }(); +} + void StressTableReader(TableReaderFactory reader_factory) { #ifdef ARROW_VALGRIND const int NTASKS = 10; @@ -347,6 +582,12 @@ TEST(SerialReaderTests, QuotedCRLFStraddlingBlockBoundary) { TEST(SerialReaderTests, HeaderRowEndingAtBlockBoundary) { TestHeaderRowEndingAtBlockBoundary(MakeSerialFactory()); } +TEST(SerialReaderTests, MixedLineEndingsStraddling) { + TestMixedLineEndingsStraddling(MakeSerialFactory()); +} +TEST(SerialReaderTests, SmallInputsAcrossBlockSizes) { + TestSmallInputsAcrossBlockSizes(MakeSerialFactory()); +} TEST(SerialReaderTests, Stress) { StressTableReader(MakeSerialFactory()); } TEST(SerialReaderTests, StressInvalid) { StressInvalidTableReader(MakeSerialFactory()); } TEST(SerialReaderTests, NestedParallelism) { @@ -391,6 +632,14 @@ TEST(AsyncReaderTests, HeaderRowEndingAtBlockBoundary) { ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory()); TestHeaderRowEndingAtBlockBoundary(table_factory); } +TEST(AsyncReaderTests, MixedLineEndingsStraddling) { + ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory()); + TestMixedLineEndingsStraddling(table_factory); +} +TEST(AsyncReaderTests, SmallInputsAcrossBlockSizes) { + ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory()); + TestSmallInputsAcrossBlockSizes(table_factory); +} TEST(AsyncReaderTests, Straddling) { ASSERT_OK_AND_ASSIGN(auto table_factory, MakeAsyncFactory()); TestStraddling(table_factory); @@ -451,6 +700,12 @@ TEST(StreamingReaderTests, QuotedCRLFStraddlingBlockBoundary) { TEST(StreamingReaderTests, HeaderRowEndingAtBlockBoundary) { TestHeaderRowEndingAtBlockBoundary(MakeStreamingFactory()); } +TEST(StreamingReaderTests, MixedLineEndingsStraddling) { + TestMixedLineEndingsStraddling(MakeStreamingFactory()); +} +TEST(StreamingReaderTests, SmallInputsAcrossBlockSizes) { + TestSmallInputsAcrossBlockSizes(MakeStreamingFactory()); +} TEST(StreamingReaderTests, Stress) { StressTableReader(MakeStreamingFactory()); } TEST(StreamingReaderTests, StressInvalid) { StressInvalidTableReader(MakeStreamingFactory());