pitrou commented on code in PR #54912:
URL: https://github.com/apache/arrow/pull/54912#discussion_r4346467289
##########
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.
Review Comment:
Hmm, can `pack_corpus.py` be changed to not fail on duplicate hashes then?
Perhaps it should just emit a warning instead of failing?
##########
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.
Review Comment:
Ok, it's not clear to me how dictionaries are used by the fuzzing engine.
Are inputs restricted to a combination of tokens from the dictionaries? Or
are random mutations still applied on top?
##########
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) {
Review Comment:
Just like in
https://github.com/apache/arrow/blob/373d64c02cb23663b63c5853099fca43a46c39fc/cpp/src/parquet/arrow/fuzz_encoding_internal.cc#L119,
it would be nicer for this function to take a `std::span<const uint8_t>` IMHO.
##########
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);
Review Comment:
Hmm, instead of all the manual bit-twiddling here, can we use a dedicated
packed structure with bit-fields?
Can get inspiration from the Parquet encoding fuzzer:
https://github.com/apache/arrow/blob/373d64c02cb23663b63c5853099fca43a46c39fc/cpp/src/parquet/arrow/fuzz_encoding_internal.cc#L70-L88
##########
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.
Review Comment:
I'm not sure we care about exercising the *writer* here. We're just
generating files for the reader.
And since those are CSV files, there is no inherent chunking in the file
contents, so I don't understand what this is achieving.
##########
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));
Review Comment:
At this point, the input has been exhausted by the previous table reader
call, so you'd just getting an empty table here?
##########
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,
+ // 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);
- return fut.status();
- };
+ 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.
Review Comment:
Well, the header is small (8 bytes), and most legacy inputs will actually go
through the other code path, and get a non-sensical header as well as an
unexpected non-header payload.
##########
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));
Review Comment:
We should also be writing the serialized `FuzzCsvOptions` header, no?
See the Parquet encoding fuzzer for how this can be done.
##########
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();
+ }
Review Comment:
Note: if all three table reads are successful, we may want to add
differential fuzzing, i.e. test that all three tables are equal (but there may
be legitimate reasons why they are not?).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
