GH-51368: [C++] Fix silent LF drop when quoted CRLF straddles CSV blo…

cpp/src/arrow/csv/reader.cc

@@ -109,6 +109,16 @@ class CSVBufferIterator {
   Result<TransformFlow<std::shared_ptr<Buffer>>> operator()(std::shared_ptr<Buffer> 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<Buffer>(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<size_t>(rest));
+      }
+      buf = std::shared_ptr<Buffer>(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 {

cpp/src/arrow/csv/reader_test.cc

@@ -29,6 +29,7 @@
 #include <utility>
 #include <vector>
 
+#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<int32_t>(prefix.size()) - 2;
+  const std::string first_value = std::string(static_cast<size_t>(pad), 'z') + embedded;
+  const std::string data =
+      prefix + std::string(static_cast<size_t>(pad), 'z') + embedded + suffix;
+
+  auto parse_options = ParseOptions::Defaults();
+  parse_options.newlines_in_values = true;
+
+  auto input = std::make_shared<io::BufferReader>(std::make_shared<Buffer>(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<std::string> values;
+  for (int i = 0; i < column->num_chunks(); ++i) {
+    auto arr = std::static_pointer_cast<StringArray>(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());