GH-54903: [C++][OSS-Fuzz fuzzing] Update csv fuzzer and corpora/dict preparation

cpp/build-support/fuzzing/dicts/arrow-csv-fuzz.dict

@@ -0,0 +1,60 @@
+# Tokens for arrow-csv-fuzz: CSV dialect characters, null/bool spellings,
+# numeric edge cases and timestamp formats understood by the Arrow CSV reader.
+","
+";"
+"\x09"
+"|"
+" "
+"\""
+"'"
+"\\"
+"\x0d\x0a"
+"\x0a"
+"\x0d"
+"NULL"
+"null"
+"None"
+"N/A"
+"n/a"
+"NaN"
+"nan"
+"inf"
+"+inf"
+"-inf"
+"Infinity"
+"true"
+"false"
+"True"
+"False"
+"TRUE"
+"FALSE"
+"1"
+"0"
+"01"
+"0.5"
+"-0.0"
+"-1.25"
+"1e10"
+"1E+10"
+"1.5e-300"
+"18446744073709551615"
+"-9223372036854775808"
+"9223372036854775807"
+"2147483648"
+"2024-12-31"
+"2024-12-31T23:59:59"
+"2024-12-31T23:59:59.123456789"
+"2024-12-31 23:59:59"
+"12/31/2024"
+"1970-01-01T00:00:00Z"
+"2024-12-31T23:59:59+0100"
+"23:59:59"
+"12:34:56.789"
+"00:00:00"
+"a,b"
+"\"quoted\""
+"\"with,\"delimiter"
+"\"escaped\\\"quote\""
+"\"\""
+"\xc3\xa9"
+"\xe6\x97\xa5\xe6\x9c\xac\xe8\xaa\x9e"

cpp/build-support/fuzzing/generate_corpuses.sh

@@ -89,4 +89,25 @@ rm -rf "${CORPUS_DIR}"
 cp "${ARROW_ROOT}"/testing/data/csv/*.csv "${CORPUS_DIR}"
 # Add examples from Pandas test suite
 find "${PANDAS_DIR}/" -name "*.csv" -exec cp --backup=numbered '{}' "${CORPUS_DIR}" \;
+# Some Pandas CSV files are exact duplicates of each other under different
+# names, and pack_corpus.py fails on duplicate hashes; keep only one copy.
+declare -A seen=()
+while IFS= read -r -d '' f; do
+  h=$(sha1sum "$f" | cut -d' ' -f1)
+  if [[ -n "${seen[$h]}" ]]; then
+    rm -f "$f"
+  else
+    seen[$h]=1
+  fi
+done < <(find "${CORPUS_DIR}" -maxdepth 1 -type f -print0)
+
 "${ARROW_CPP}/build-support/fuzzing/pack_corpus.py" "${CORPUS_DIR}" "${OUT}/arrow-csv-fuzz_seed_corpus.zip"
+
+# Fuzz target dictionaries
+#
+# Dictionaries are shipped under build-support/fuzzing/dicts as
+# "<FUZZ TARGET>.dict" and copied next to the seed corpus zips, so that
+# OSS-Fuzz's build script can pick them up like the corpora above.
+
+find "${ARROW_CPP}/build-support/fuzzing/dicts" -name '*.dict' \
+  -exec cp '{}' "${OUT}/" \;

cpp/src/arrow/csv/fuzz.cc

@@ -29,92 +29,160 @@
 #include "arrow/util/fuzz_internal.h"
 #include "arrow/util/macros.h"
 #include "arrow/util/thread_pool.h"
+#include "arrow/util/value_parsing.h"
 
 namespace arrow::csv {
 
 using ::arrow::internal::GetCpuThreadPool;
 using ::arrow::io::InputStream;
 
-Status FuzzCsvReader(const uint8_t* data, int64_t size) {
-  // Since the Fuzz-allocated data is not owned, any task that outlives the TableReader
-  // may try to read memory that has been deallocated. Hence we wait for all pending
-  // tasks to end before leaving.
-  struct TaskGuard {
-    ~TaskGuard() { GetCpuThreadPool()->WaitForIdle(); }
-  };
+namespace {
 
-  auto io_context = arrow::io::default_io_context();
+// Packed header prefixing the fuzz payload, used to explore the CSV reader's
+// option space (see https://github.com/google/oss-fuzz/issues/14437 for why
+// fuzz target parameters have to be encoded inside the payload itself).
+//
+// All option values are total functions of the header bytes, so every possible
+// input leads to a well-defined option set. Inputs shorter than the header are
+// read with the original hardcoded options, preserving the fuzzer's historical
+// behavior on legacy corpus entries.
+constexpr int64_t kFuzzHeaderSize = 8;
+
+struct FuzzCsvOptions {
+  ReadOptions read;
+  ParseOptions parse;
+  ConvertOptions convert;
+};
+
+FuzzCsvOptions MakeFuzzCsvOptions(const uint8_t* header) {
+  FuzzCsvOptions opts;
+
+  // --- ParseOptions ---
+  const char kDelimiters[] = {',', '\t', ';', '|', ' '};
+  opts.parse.delimiter = kDelimiters[header[0] % 5];
+  opts.parse.quoting = !(header[1] & 1);
+  opts.parse.quote_char = (header[1] & 2) ? '\'' : '"';
+  opts.parse.double_quote = !(header[1] & 4);
+  opts.parse.escaping = (header[2] & 1);
+  opts.parse.escape_char = (header[2] & 2) ? '"' : '\\';
+  opts.parse.newlines_in_values = (header[4] & 1);
+  opts.parse.ignore_empty_lines = !(header[4] & 2);
+  opts.parse.pad_short_rows = (header[4] & 4);
+  opts.parse.ignore_extra_columns = (header[4] & 8);
+
+  // --- ReadOptions ---
+  const int32_t kBlockSizes[] = {1024, 4096, 65536, 1 << 20};
+  opts.read.block_size = kBlockSizes[(header[3] >> 1) % 4];
+  opts.read.use_threads = (header[3] & 1);
+  opts.read.skip_rows = (header[7] & 1) ? 1 : 0;
+  opts.read.autogenerate_column_names = (header[7] & 2);
+
+  // --- ConvertOptions ---
+  opts.convert.strings_can_be_null = (header[5] & 1);
+  opts.convert.quoted_strings_can_be_null = !(header[5] & 2);
+  opts.convert.auto_dict_encode = (header[5] & 4);
+  const int32_t kDictMaxCardinalities[] = {50, 1000, 1000000};
+  opts.convert.auto_dict_max_cardinality = kDictMaxCardinalities[(header[5] >> 3) % 3];
+  opts.convert.decimal_point = (header[6] & 1) ? ',' : '.';
+  opts.convert.check_utf8 = !(header[6] & 2);
+  if (header[6] & 4) {
+    opts.convert.timestamp_parsers.push_back(TimestampParser::MakeStrptime("%Y-%m-%d"));
+  }
+  if (header[6] & 8) {
+    opts.convert.timestamp_parsers.push_back(TimestampParser::MakeStrptime("%m/%d/%Y"));
+  }
 
-  auto read_options = ReadOptions::Defaults();
-  // Make chunking more likely to exercise chunked reading and optional parallelization.
-  // Most files in the seed corpus are currently in the 4-10 kB range.
-  read_options.block_size = 1000;
-  auto parse_options = ParseOptions::Defaults();
-  auto convert_options = ConvertOptions::Defaults();
-  convert_options.auto_dict_encode = true;
-  // This is the default value, but we might want to turn this knob to have a better
-  // mix of dict-encoded and non-dict-encoded columns when reading.
-  convert_options.auto_dict_max_cardinality = 50;
+  return opts;
+}
 
-  // TODO should we also test non-inferring table read?
+Status ReadAllWays(io::IOContext io_context, const FuzzCsvOptions& opts,
+                   std::shared_ptr<InputStream> input) {
+  Status st;
 
-  auto read_table_serial = [=](std::shared_ptr<InputStream> input) mutable -> Status {
+  // 1. Full table read, single-threaded
+  {
+    auto read_options = opts.read;
     read_options.use_threads = false;
-    ARROW_ASSIGN_OR_RAISE(auto table_reader,
-                          TableReader::Make(io_context, input, read_options,
-                                            parse_options, convert_options));
+    ARROW_ASSIGN_OR_RAISE(
+        auto table_reader,
+        TableReader::Make(io_context, input, read_options, opts.parse, opts.convert));
     ARROW_ASSIGN_OR_RAISE(auto table, table_reader->Read());
-    return table->ValidateFull();
-  };
+    st &= table->ValidateFull();
+  }
 
-  auto read_table_threaded = [=](std::shared_ptr<InputStream> input) mutable -> Status {
+  // 2. Full table read, multi-threaded
+  {
+    auto read_options = opts.read;
     read_options.use_threads = true;
-    ARROW_ASSIGN_OR_RAISE(auto table_reader,
-                          TableReader::Make(io_context, input, read_options,
-                                            parse_options, convert_options));
+    ARROW_ASSIGN_OR_RAISE(
+        auto table_reader,
+        TableReader::Make(io_context, input, read_options, opts.parse, opts.convert));
     ARROW_ASSIGN_OR_RAISE(auto table, table_reader->Read());
-    return table->ValidateFull();
-  };
+    st &= table->ValidateFull();
+  }
 
-  auto read_streaming = [=](std::shared_ptr<InputStream> input) mutable -> Status {
+  // 3. Streaming (incremental) read
+  {
+    auto read_options = opts.read;
     read_options.use_threads = true;
     ARROW_ASSIGN_OR_RAISE(
-        auto reader, StreamingReader::Make(io_context, input, read_options, parse_options,
-                                           convert_options));
+        auto reader,
+        StreamingReader::Make(io_context, input, read_options, opts.parse, opts.convert));
     ARROW_ASSIGN_OR_RAISE(auto table, reader->ToTable());
-    return table->ValidateFull();
-  };
+    st &= table->ValidateFull();
+  }
 
-  auto count_rows = [=](std::shared_ptr<InputStream> input) mutable -> Status {
-    read_options.use_threads = true;
-    parse_options.newlines_in_values = false;
-    auto fut = CountRowsAsync(io_context, input, GetCpuThreadPool(), read_options,
-                              parse_options);
-    return fut.status();
-  };
+  // 4. Row counting, with both values for `newlines_in_values` (row counting
+  //    scans rows without conversion, so it exercises parsing on its own)
+  for (const bool newlines_in_values : {false, true}) {
+    auto parse_options = opts.parse;
+    parse_options.newlines_in_values = newlines_in_values;
+    auto fut =
+        CountRowsAsync(io_context, input, GetCpuThreadPool(), opts.read, parse_options);
+    st &= fut.status();
+  }
 
-  auto count_rows_allow_newlines =
-      [=](std::shared_ptr<InputStream> input) mutable -> Status {
-    read_options.use_threads = true;
-    parse_options.newlines_in_values = true;
-    auto fut = CountRowsAsync(io_context, input, GetCpuThreadPool(), read_options,
-                              parse_options);
-    return fut.status();
+  return st;
+}
+
+}  // namespace
+
+Status FuzzCsvReader(const uint8_t* data, int64_t size) {
+  // Since the Fuzz-allocated data is not owned, any task that outlives the
+  // TableReader may try to read memory that has been deallocated. Hence we
+  // wait for all pending tasks to end before leaving.
+  struct TaskGuard {
+    ~TaskGuard() { GetCpuThreadPool()->WaitForIdle(); }
   };
+  TaskGuard task_guard;
 
-  using ReadFunc = decltype(std::function(read_table_serial));
+  auto io_context = arrow::io::default_io_context();
 
-  // Test all reader types regardless of outcome
-  Status st;
-  for (auto read_func : {ReadFunc(read_table_serial), ReadFunc(read_table_threaded),
-                         ReadFunc(read_streaming), ReadFunc(count_rows),
-                         ReadFunc(count_rows_allow_newlines)}) {
-    auto input =
-        std::make_shared<::arrow::io::BufferReader>(std::make_shared<Buffer>(data, size));
-    TaskGuard task_guard;
-    st &= read_func(input);
+  FuzzCsvOptions opts;
+  const uint8_t* payload = data;
+  int64_t payload_size = size;
+  if (size >= kFuzzHeaderSize) {
+    opts = MakeFuzzCsvOptions(data);
+    payload += kFuzzHeaderSize;
+    payload_size -= kFuzzHeaderSize;
+    RETURN_NOT_OK(opts.parse.Validate());
+    RETURN_NOT_OK(opts.read.Validate());
+    RETURN_NOT_OK(opts.convert.Validate());
+  } else {
+    // Legacy behavior for inputs without a full header: the original
+    // hardcoded options. Small block sizes make chunking more likely to be
+    // exercised, as most legacy corpus entries are in the 4-10 kB range.
+    opts.read = ReadOptions::Defaults();
+    opts.read.block_size = 1000;
+    opts.parse = ParseOptions::Defaults();
+    opts.convert = ConvertOptions::Defaults();
+    opts.convert.auto_dict_encode = true;
+    opts.convert.auto_dict_max_cardinality = 50;
   }
-  return st;
+
+  return ReadAllWays(io_context, opts,
+                     std::make_shared<::arrow::io::BufferReader>(
+                         std::make_shared<Buffer>(payload, payload_size)));
 }
 
 }  // namespace arrow::csv

cpp/src/arrow/csv/generate_fuzz_corpus.cc

@@ -163,18 +163,37 @@ Status DoMain(const std::string& out_dir) {
 
   ARROW_ASSIGN_OR_RAISE(auto batches, Batches());
 
-  auto options = WriteOptions::Defaults();
-  RETURN_NOT_OK(options.Validate());
+  // Generate several dialect variants of each batch, matching the option
+  // rotations performed by the fuzz target (see csv/fuzz.cc).
+  std::vector<WriteOptions> option_variants;
+  auto append_variant = [&](std::function<void(WriteOptions&)> fn) {
+    auto options = WriteOptions::Defaults();
+    fn(options);
+    if (options.Validate().ok()) {
+      option_variants.push_back(std::move(options));
+    }
+  };
+  append_variant([](WriteOptions&) {});
+  append_variant([](WriteOptions& o) { o.delimiter = ';'; });
+  append_variant([](WriteOptions& o) { o.delimiter = '\t'; });
+  append_variant([](WriteOptions& o) { o.quoting_style = QuotingStyle::AllValid; });
+  append_variant([](WriteOptions& o) { o.include_header = false; });
+  // Small batch sizes exercise the writer's chunking and produce files that
+  // make the reader merge many batches.
+  append_variant([](WriteOptions& o) { o.batch_size = 7; });
+  append_variant([](WriteOptions& o) { o.null_string = "N/A"; });
 
   for (const auto& batch : batches) {
     RETURN_NOT_OK(batch->ValidateFull());
-    ARROW_ASSIGN_OR_RAISE(auto buffer, WriteRecordBatch(batch, options));
-
-    ARROW_ASSIGN_OR_RAISE(auto sample_fn, dir_fn.Join(sample_name()));
-    std::cerr << sample_fn.ToString() << std::endl;
-    ARROW_ASSIGN_OR_RAISE(auto file, io::FileOutputStream::Open(sample_fn.ToString()));
-    RETURN_NOT_OK(file->Write(buffer));
-    RETURN_NOT_OK(file->Close());
+    for (const auto& options : option_variants) {
+      ARROW_ASSIGN_OR_RAISE(auto buffer, WriteRecordBatch(batch, options));
+
+      ARROW_ASSIGN_OR_RAISE(auto sample_fn, dir_fn.Join(sample_name()));
+      std::cerr << sample_fn.ToString() << std::endl;
+      ARROW_ASSIGN_OR_RAISE(auto file, io::FileOutputStream::Open(sample_fn.ToString()));
+      RETURN_NOT_OK(file->Write(buffer));
+      RETURN_NOT_OK(file->Close());
+    }
   }
   return Status::OK();
 }