From abeddb857dadcd130644bf4073b55f94287d9016 Mon Sep 17 00:00:00 2001 From: Aditya Raj Date: Sat, 17 Oct 2026 19:25:33 +0530 Subject: [PATCH] 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());