GH-50481: [C++] Fix CSV reader mis-parsing rows with an embedded NUL byte

cpp/src/arrow/csv/chunker.cc

@@ -56,6 +56,10 @@ class Lexer {
 
   // Decide whether it's worth using a bulk filter over the given data area
   bool ShouldUseBulkFilter(const char* data, const char* data_end) {
+    if (!bulk_filter_.CanUseOnBlock(
+            std::string_view(data, static_cast<size_t>(data_end - data)))) {
+      return false;
+    }
     constexpr int32_t kWordSize = static_cast<int32_t>(sizeof(BulkWordType));
 
     // Only probe the 32 first words and assume they are representative of the rest

cpp/src/arrow/csv/chunker_test.cc

@@ -341,6 +341,26 @@ TEST_P(BaseChunkerTest, EscapingAndQuoting) {
   }
 }
 
+TEST_P(BaseChunkerTest, EmbeddedNulBytesDisableBulkFilter) {
+  // Regression test for GH-50481 in Lexer's own bulk filter (used for
+  // chunking, separately from BlockParser's).
+  //
+  // Lead-in with no structural bytes so ShouldUseBulkFilter's probe of the
+  // first 256 bytes decides to use the bulk filter here.
+  std::string lead_in(300, 'x');
+  // NUL right before the closing quote: the misaligned-SIMD-scan trigger.
+  std::string trigger_field = "abc";
+  trigger_field += '\0';
+  trigger_field += "def";
+  std::string first_row = lead_in + ",\"" + trigger_field + "\",tail\n";
+  // No trailing newline on the second row, so Process() must split it out
+  // as the partial remainder -- proving the first row's boundary was found.
+  std::string csv = first_row + "next";
+
+  MakeChunker();
+  AssertChunkSize(*chunker_, csv, static_cast<uint32_t>(first_row.size()));
+}
+
 TEST_P(BaseChunkerTest, ParseSkip) {
   {
     auto csv = MakeCSVData({"ab,c,\n", "def,,gh\n", ",ij,kl\n"});

cpp/src/arrow/csv/lexing_internal.h

@@ -18,6 +18,8 @@
 #pragma once
 
 #include <cstdint>
+#include <cstring>
+#include <string_view>
 
 #include "arrow/csv/options.h"
 #include "arrow/util/simd.h"
@@ -45,6 +47,10 @@ class BaseBloomFilter {
  public:
   explicit BaseBloomFilter(const ParseOptions& options) : filter_(MakeFilter(options)) {}
 
+  // Bloom filters match bytes individually and have no implicit-length
+  // scanning, so an embedded NUL byte doesn't affect their correctness.
+  bool CanUseOnBlock(std::string_view) const { return true; }
+
  protected:
   using FilterType = uint64_t;
   // 63 for uint64_t
@@ -138,6 +144,17 @@ class SSE42Filter {
                         _SIDD_UBYTE_OPS | _SIDD_CMP_EQUAL_ANY);
   }
 
+  // _mm_cmpistrc is an implicit-length compare: it treats a NUL byte as a
+  // terminator and can miss a real delimiter/quote/newline sharing an 8-byte
+  // word with one. Never use this filter on a block that contains a NUL.
+  // We could instead use the explicit-length compare _mm_cmpestrc but
+  // it comes with a massive performance cost on some CPUs
+  // (some benchmarks were measured to be twice slower in
+  // https://github.com/apache/arrow/pull/50483).
+  bool CanUseOnBlock(std::string_view data) const {
+    return data.empty() || std::memchr(data.data(), '\0', data.size()) == nullptr;
+  }
+
  protected:
   using BulkFilterType = __m128i;
 
@@ -190,6 +207,10 @@ class NeonFilter {
     return r != 0;
   }
 
+  // NEON compares each byte independently (no implicit-length scanning), so
+  // an embedded NUL byte doesn't affect correctness here.
+  bool CanUseOnBlock(std::string_view) const { return true; }
+
  private:
   const uint8x8_t delim_, quote_, escape_;
 };

cpp/src/arrow/csv/parser.cc

@@ -533,12 +533,12 @@ class BlockParserImpl {
     }
 
     if (batch_.num_rows_ > start_num_rows && batch_.num_cols_ > 0) {
-      // Use bulk filter only if average value length is >= 10 bytes,
-      // as the bulk filter has a fixed cost that isn't compensated
-      // when values are too short.
+      // Use bulk filter only if average value length is >= 10 bytes
+      // (its fixed cost isn't compensated for short values), and the block
+      // has no embedded NUL bytes (see block_has_nul_).
       const int64_t bulk_filter_threshold = static_cast<int64_t>(batch_.num_cols_) *
                                             (batch_.num_rows_ - start_num_rows) * 10;
-      use_bulk_filter_ = (data - *out_data) > bulk_filter_threshold;
+      use_bulk_filter_ = !block_has_nul_ && (data - *out_data) > bulk_filter_threshold;
     }
 
     // Append new buffers and update size
@@ -561,8 +561,16 @@ class BlockParserImpl {
     values_size_ = 0;
 
     size_t total_view_length = 0;
+    block_has_nul_ = false;
     for (const auto& view : views) {
       total_view_length += view.length();
+      if (!block_has_nul_ && !bulk_filter.CanUseOnBlock(view)) {
+        block_has_nul_ = true;
+      }
+    }
+    if (block_has_nul_) {
+      // Clear a bulk filter left on by an earlier NUL-free block.
+      use_bulk_filter_ = false;
     }
     if (total_view_length > std::numeric_limits<uint32_t>::max()) {
       return Status::Invalid("CSV block too large");
@@ -691,6 +699,7 @@ class BlockParserImpl {
   int32_t max_num_rows_;
 
   bool use_bulk_filter_ = false;
+  bool block_has_nul_ = false;
 
   // Unparsed data size
   int32_t values_size_;

cpp/src/arrow/csv/parser_test.cc

@@ -936,5 +936,49 @@ TEST(BlockParser, RowNumberAppendedToError) {
   }
 }
 
+TEST(BlockParser, EmbeddedNulBytesDisableBulkFilter) {
+  // Regression test for GH-50481: disables the bulk filter for any block
+  // with an embedded NUL, so every cell here carries one.
+  constexpr int32_t num_cols = 64;
+  // 4x the ~512-row ParseChunk cap for num_cols == 64, so the filler block
+  // spans multiple calls even if that internal constant changes.
+  constexpr int32_t num_filler_rows = 4 * 512;
+
+  // 12 bytes/value, above the bulk filter's activation threshold.
+  std::string filler_cell = "xxxxxxxxxxx";
+  filler_cell += '\0';
+
+  std::string csv;
+  for (int32_t r = 0; r < num_filler_rows; ++r) {
+    for (int32_t c = 0; c < num_cols; ++c) {
+      if (c) csv += ',';
+      csv += filler_cell;
+    }
+    csv += '\n';
+  }
+  // NUL right before the closing quote: the misaligned-SIMD-scan trigger.
+  csv += "\"abc";
+  csv += '\0';
+  csv += "def\"";
+  for (int32_t c = 1; c < num_cols; ++c) {
+    csv += ',';
+    csv += filler_cell;
+  }
+  csv += '\n';
+
+  BlockParser parser(ParseOptions::Defaults(), num_cols, /*first_row=*/0);
+  AssertParseFinal(parser, csv);
+  ASSERT_EQ(parser.num_rows(), num_filler_rows + 1);
+
+  std::vector<std::string> last_row;
+  GetLastRow(parser, &last_row);
+  ASSERT_EQ(last_row.size(), static_cast<size_t>(num_cols));
+  ASSERT_EQ(last_row[0], std::string("abc\0def", 7));
+  // Other NUL-bearing fields in the row must come through unmangled too.
+  for (size_t c = 1; c < last_row.size(); ++c) {
+    ASSERT_EQ(last_row[c], filler_cell) << "column " << c;
+  }
+}
+
 }  // namespace csv
 }  // namespace arrow