Function FuzzIpcFile

cpp/src/arrow/ipc/reader.cc:2785–2887

}

Status FuzzIpcFile(const uint8_t* data, int64_t size) {
  auto buffer = std::make_shared<Buffer>(data, size);

  Status final_status;

  struct IpcReadResult {
    std::shared_ptr<Schema> schema;
    std::vector<RecordBatchWithMetadata> batches = {};
  };

  // Try to read the IPC file as a stream to compare the results (differential fuzzing)
  auto do_stream_read = [&]() -> Result<IpcReadResult> {
    io::BufferReader buffer_reader(buffer);
    // Skip magic bytes at the beginning
    RETURN_NOT_OK(
        buffer_reader.Advance(bit_util::RoundUpToMultipleOf8(kArrowMagicBytes.length())));
    ARROW_ASSIGN_OR_RAISE(auto batch_reader, RecordBatchStreamReader::Open(
                                                 &buffer_reader, FuzzingOptions()));

    std::vector<RecordBatchWithMetadata> batches;
    while (true) {
      ARROW_ASSIGN_OR_RAISE(auto batch, batch_reader->ReadNext());
      if (!batch.batch && !batch.custom_metadata) {
        // EOS
        break;
      }
      RETURN_NOT_OK(ValidateFuzzBatch(batch));
      batches.push_back(batch);
    }
    return IpcReadResult{batch_reader->schema(), batches};
  };

  auto do_file_read = [&](bool pre_buffer) -> Result<IpcReadResult> {
    io::BufferReader buffer_reader(buffer);
    ARROW_ASSIGN_OR_RAISE(auto batch_reader,
                          RecordBatchFileReader::Open(&buffer_reader, FuzzingOptions()));
    if (pre_buffer) {
      // Pre-buffer all record batches
      RETURN_NOT_OK(batch_reader->PreBufferMetadata(/*indices=*/{}));
    }

    IpcReadResult result = {.schema = batch_reader->schema()};
    const int n_batches = batch_reader->num_record_batches();
    // Delay error return until the end, as we want to access all record batches
    Status st;
    for (int i = 0; i < n_batches; ++i) {
      RecordBatchWithMetadata batch;
      st &= batch_reader->ReadRecordBatchWithCustomMetadata(i).Value(&batch);
      st &= ValidateFuzzBatch(batch);
      result.batches.push_back(batch);
    }
    RETURN_NOT_OK(st);
    return result;
  };

  // Persistent read result for differential fuzzing.
  std::optional<IpcReadResult> maybe_read_result;