diff --git a/cpp/build-support/fuzzing/dicts/arrow-csv-fuzz.dict b/cpp/build-support/fuzzing/dicts/arrow-csv-fuzz.dict new file mode 100644 index 000000000000..bbda8108e016 --- /dev/null +++ b/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" \ No newline at end of file diff --git a/cpp/build-support/fuzzing/generate_corpuses.sh b/cpp/build-support/fuzzing/generate_corpuses.sh index 2d95e110b516..25fa735a51ec 100755 --- a/cpp/build-support/fuzzing/generate_corpuses.sh +++ b/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 +# ".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}/" \; diff --git a/cpp/src/arrow/csv/fuzz.cc b/cpp/src/arrow/csv/fuzz.cc index 2219efa2cb59..beabfc4c8415 100644 --- a/cpp/src/arrow/csv/fuzz.cc +++ b/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 input) { + Status st; - auto read_table_serial = [=](std::shared_ptr 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 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 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 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 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(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(payload, payload_size))); } } // namespace arrow::csv diff --git a/cpp/src/arrow/csv/generate_fuzz_corpus.cc b/cpp/src/arrow/csv/generate_fuzz_corpus.cc index eeba49e87f81..dc369121eb96 100644 --- a/cpp/src/arrow/csv/generate_fuzz_corpus.cc +++ b/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 option_variants; + auto append_variant = [&](std::function 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(); }