| 36 | using ::arrow::io::InputStream; |
| 37 | |
| 38 | Status FuzzCsvReader(const uint8_t* data, int64_t size) { |
| 39 | // Since the Fuzz-allocated data is not owned, any task that outlives the TableReader |
| 40 | // may try to read memory that has been deallocated. Hence we wait for all pending |
| 41 | // tasks to end before leaving. |
| 42 | struct TaskGuard { |
| 43 | ~TaskGuard() { GetCpuThreadPool()->WaitForIdle(); } |
| 44 | }; |
| 45 | |
| 46 | auto io_context = arrow::io::default_io_context(); |
| 47 | |
| 48 | auto read_options = ReadOptions::Defaults(); |
| 49 | // Make chunking more likely to exercise chunked reading and optional parallelization. |
| 50 | // Most files in the seed corpus are currently in the 4-10 kB range. |
| 51 | read_options.block_size = 1000; |
| 52 | auto parse_options = ParseOptions::Defaults(); |
| 53 | auto convert_options = ConvertOptions::Defaults(); |
| 54 | convert_options.auto_dict_encode = true; |
| 55 | // This is the default value, but we might want to turn this knob to have a better |
| 56 | // mix of dict-encoded and non-dict-encoded columns when reading. |
| 57 | convert_options.auto_dict_max_cardinality = 50; |
| 58 | |
| 59 | // TODO should we also test non-inferring table read? |
| 60 | |
| 61 | auto read_table_serial = [=](std::shared_ptr<InputStream> input) mutable -> Status { |
| 62 | read_options.use_threads = false; |
| 63 | ARROW_ASSIGN_OR_RAISE(auto table_reader, |
| 64 | TableReader::Make(io_context, input, read_options, |
| 65 | parse_options, convert_options)); |
| 66 | ARROW_ASSIGN_OR_RAISE(auto table, table_reader->Read()); |
| 67 | return table->ValidateFull(); |
| 68 | }; |
| 69 | |
| 70 | auto read_table_threaded = [=](std::shared_ptr<InputStream> input) mutable -> Status { |
| 71 | read_options.use_threads = true; |
| 72 | ARROW_ASSIGN_OR_RAISE(auto table_reader, |
| 73 | TableReader::Make(io_context, input, read_options, |
| 74 | parse_options, convert_options)); |
| 75 | ARROW_ASSIGN_OR_RAISE(auto table, table_reader->Read()); |
| 76 | return table->ValidateFull(); |
| 77 | }; |
| 78 | |
| 79 | auto read_streaming = [=](std::shared_ptr<InputStream> input) mutable -> Status { |
| 80 | read_options.use_threads = true; |
| 81 | ARROW_ASSIGN_OR_RAISE( |
| 82 | auto reader, StreamingReader::Make(io_context, input, read_options, parse_options, |
| 83 | convert_options)); |
| 84 | ARROW_ASSIGN_OR_RAISE(auto table, reader->ToTable()); |
| 85 | return table->ValidateFull(); |
| 86 | }; |
| 87 | |
| 88 | auto count_rows = [=](std::shared_ptr<InputStream> input) mutable -> Status { |
| 89 | read_options.use_threads = true; |
| 90 | parse_options.newlines_in_values = false; |
| 91 | auto fut = CountRowsAsync(io_context, input, GetCpuThreadPool(), read_options, |
| 92 | parse_options); |
| 93 | return fut.status(); |
| 94 | }; |
| 95 |
no test coverage detected