The problem

arrow::csv::TableReader can silently mis-split a row and then fail with a misleading
"Expected N columns, got M" error, or in some cases succeed with corrupted column boundaries,
when a text field contains a raw embedded NUL (0x00) byte partway through a file. Rows before
the failure parse correctly, only later rows break, which makes this very easy to misdiagnose as
a data problem rather than a parser bug.

Root cause

SSE42Filter::Matches in cpp/src/arrow/csv/lexing_internal.h uses _mm_cmpistrc, an
implicit-length SSE4.2 string-compare intrinsic:

https://github.com/apache/arrow/blob/main/cpp/src/arrow/csv/lexing_internal.h#L135-L138

cpp

bool Matches(WordType w) const {
  // Look up every byte in `w` in the SIMD filter.
  return _mm_cmpistrc(_mm_set1_epi64x(w), filter_,
                      _SIDD_UBYTE_OPS | _SIDD_CMP_EQUAL_ANY);
}

_mm_cmpistrc treats 0x00 as an implicit string terminator and stops scanning at the first one
in either operand. Since the caller feeds it 8 raw CSV bytes at a time (RunBulkFilter in
cpp/src/arrow/csv/parser.cc):

https://github.com/apache/arrow/blob/main/cpp/src/arrow/csv/parser.cc#L473-L493

a real quote/comma/newline that shares an 8-byte word with an embedded NUL becomes invisible to
the filter. RunBulkFilter then trusts the filter's "no special chars here" answer and bulk-copies
the whole 8-byte word verbatim, silently swallowing the structural character:

if (bulk_filter.Matches(word)) {
  return data;
}
// No special chars
data_writer->PushFieldWord(word);   // <-- copies the real delimiter/quote too, unexamined
data += sizeof(WordType);

The identical pattern also exists in the row-chunker (Lexer::RunBulkFilter in
cpp/src/arrow/csv/chunker.cc, used when newlines_in_values=true to find safe row-cut points in
raw reads), so which row actually exhibits the corruption is also sensitive to chunk/block
boundaries.

Why it only shows up partway through a file

BlockParser only switches to the buggy SIMD path once it decides the data is "worth" the fixed
cost of the bulk filter — a per-instance flag re-evaluated after each
max(32768/num_cols, 512)-row internal chunk, based on that chunk's own average bytes/value:

https://github.com/apache/arrow/blob/main/cpp/src/arrow/csv/parser.cc#L539-L541

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;
and the chunk size itself:
https://github.com/apache/arrow/blob/main/cpp/src/arrow/csv/parser.cc#L600
constexpr int32_t kTargetChunkSize = 32768;  // in number of values

So the first internal chunk of a fresh parser always runs in the safe (non-SIMD) scanning mode;
only once average-bytes-per-value exceeds 10 does the next chunk switch to the buggy SIMD path.
For a typical CSV with reasonably wide text columns, that's essentially every file past its first
~32768/num_cols rows — which is why a NUL byte anywhere in the first chunk is harmless, but the
identical byte pattern later in the same file corrupts parsing.

Reproducing (standalone C++, public API only)

#include <arrow/csv/api.h>
#include <arrow/io/api.h>
#include <arrow/api.h>
#include <iostream>
#include <sstream>

int main() {
    const int num_cols = 16;
    const int num_rows = 3000;
    const int nul_row = 2048;  // must land after the first ~2048-row chunk

    std::ostringstream csv;
    for (int c = 0; c < num_cols; ++c) csv << (c ? "," : "") << "col" << c;
    csv << "\n";
    for (int r = 0; r < num_rows; ++r) {
        if (r == nul_row) {
            csv << "\"abc" << '\0' << "def\"";
        } else {
            csv << "val_" << r << "_0";
        }
        for (int c = 1; c < num_cols; ++c) csv << ",val_" << r << "_" << c;
        csv << "\n";
    }

    auto buffer = arrow::Buffer::FromString(csv.str());
    auto input = std::make_shared<arrow::io::BufferReader>(buffer);
    auto maybe_reader = arrow::csv::TableReader::Make(
        arrow::io::default_io_context(), input,
        arrow::csv::ReadOptions::Defaults(),
        arrow::csv::ParseOptions::Defaults(),
        arrow::csv::ConvertOptions::Defaults());
    auto maybe_table = (*maybe_reader)->Read();
    if (!maybe_table.ok()) {
        std::cerr << "Read failed: " << maybe_table.status().ToString() << std::endl;
        return 1;
    }
    std::cout << "OK - num_rows: " << (*maybe_table)->num_rows() << std::endl;
    return 0;
}

Actual:
Read failed: Invalid: CSV parse error: Expected 16 columns, got 1: "abc def",val_2048_1,val_2048_2,val_2048_3,val_2048_4,val_2048_5,val_2048_6,val_2048_7,val_2048_ ...

Expected: a table with 3000 rows, row 2048's first column equal to "abc\0def" (the exact value
with the embedded NUL preserved), same as every row before it.

Suggested fix

replace _mm_cmpistrc with the explicit-length variant _mm_cmpestrc
(which takes explicit byte counts for both operands instead of relying on NUL-termination), so
a 0x00 inside the 8-byte word is treated as ordinary data rather than end-of-string.

Component(s)

C++