| 212 | } |
| 213 | |
| 214 | void TestInvalidRowsSkipped(TableReaderFactory reader_factory, bool async) { |
| 215 | const int NROWS = 1000; |
| 216 | const int INVALID_EVERY = 20; |
| 217 | const int NINVALID = 50; |
| 218 | |
| 219 | auto opts = ParseOptions::Defaults(); |
| 220 | std::atomic<int> num_invalid_rows(0); |
| 221 | opts.invalid_row_handler = [&](const InvalidRow& row) { |
| 222 | auto cur_invalid_rows = ++num_invalid_rows; |
| 223 | if (async) { |
| 224 | // Row numbers are not counted in batches during async processing |
| 225 | EXPECT_EQ(-1, row.number); |
| 226 | } else { |
| 227 | // actual data starts at row #2 in the CSV "file" |
| 228 | EXPECT_EQ(cur_invalid_rows * INVALID_EVERY + 1, row.number); |
| 229 | } |
| 230 | return InvalidRowResult::Skip; |
| 231 | }; |
| 232 | |
| 233 | ASSERT_OK_AND_ASSIGN(auto table_buffer, MakeSampleCsvBuffer(NROWS, [=](size_t row_num) { |
| 234 | // row_num is 0-based |
| 235 | return (row_num + 1) % static_cast<size_t>(INVALID_EVERY) != 0; |
| 236 | })); |
| 237 | auto input = std::make_shared<io::BufferReader>(table_buffer); |
| 238 | ASSERT_OK_AND_ASSIGN(auto reader, |
| 239 | reader_factory(input, std::move(opts), /*block_size=*/{})); |
| 240 | ASSERT_OK_AND_ASSIGN(auto table, reader->Read()); |
| 241 | ASSERT_EQ(NROWS - NINVALID, table->num_rows()); |
| 242 | ASSERT_EQ(NINVALID, num_invalid_rows); |
| 243 | } |
| 244 | |
| 245 | TableReaderFactory MakeSerialFactory() { |
| 246 | return [](std::shared_ptr<io::InputStream> input_stream, ParseOptions parse_options, |