You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by ap...@apache.org on 2026/03/10 15:28:10 UTC
(arrow) branch main updated: GH-49448: [C++][CI] Detect mismatching schema in differential IPC fuzzing (#49451)
This is an automated email from the ASF dual-hosted git repository.
apitrou pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/main by this push:
new 425666c8b7 GH-49448: [C++][CI] Detect mismatching schema in differential IPC fuzzing (#49451)
425666c8b7 is described below
commit 425666c8b726dd81401d56305bc5138f755f3663
Author: Antoine Pitrou <an...@python.org>
AuthorDate: Tue Mar 10 16:28:02 2026 +0100
GH-49448: [C++][CI] Detect mismatching schema in differential IPC fuzzing (#49451)
### Rationale for this change
In https://github.com/apache/arrow/issues/49311 we added differential fuzzing for the IPC file fuzzer, where we compare the contents as read by the IPC file reader and the IPC stream reader, respectively.
However, as the IPC file footer carries a duplicate definition of the file's schema, the fuzzer could mutate it and make it mismatch the IPC stream schema. The fuzz target would currently fail on such a file, even though it's technically not a bug in the IPC implementation.
This was detected by OSS-Fuzz in https://issues.oss-fuzz.com/issues/486127061
### What changes are included in this PR?
* Detect a mismatching schema between IPC stream and IPC file, and skip the batches comparison in that case.
* Compare results between IPC file reads with pre-buffering enabled/disabled
* When comparing IPC read results, also compare custom metadata, not just the data batches themselves
### Are these changes tested?
Yes, by new regression test.
### Are there any user-facing changes?
No.
* GitHub Issue: #49448
Authored-by: Antoine Pitrou <an...@python.org>
Signed-off-by: Antoine Pitrou <an...@python.org>
---
cpp/src/arrow/ipc/reader.cc | 112 ++++++++++++++++++++++++++++++++++----------
testing | 2 +-
2 files changed, 88 insertions(+), 26 deletions(-)
diff --git a/cpp/src/arrow/ipc/reader.cc b/cpp/src/arrow/ipc/reader.cc
index 991d238240..6aa502dc27 100644
--- a/cpp/src/arrow/ipc/reader.cc
+++ b/cpp/src/arrow/ipc/reader.cc
@@ -23,6 +23,7 @@
#include <memory>
#include <numeric>
#include <optional>
+#include <sstream>
#include <string>
#include <type_traits>
#include <unordered_map>
@@ -42,6 +43,7 @@
#include "arrow/ipc/metadata_internal.h"
#include "arrow/ipc/reader_internal.h"
#include "arrow/ipc/writer.h"
+#include "arrow/pretty_print.h"
#include "arrow/record_batch.h"
#include "arrow/sparse_tensor.h"
#include "arrow/status.h"
@@ -2694,13 +2696,49 @@ Status ValidateFuzzBatch(const RecordBatchWithMetadata& batch) {
Status CompareFuzzBatches(const RecordBatchWithMetadata& left,
const RecordBatchWithMetadata& right) {
+ if ((left.custom_metadata != nullptr) != (right.custom_metadata != nullptr) ||
+ (left.custom_metadata && !left.custom_metadata->Equals(*right.custom_metadata))) {
+ std::stringstream ss;
+ auto print_metadata = [&](const RecordBatchWithMetadata& batch) {
+ if (batch.custom_metadata) {
+ ss << batch.custom_metadata->ToString();
+ } else {
+ ss << "nullptr";
+ }
+ };
+ ss << "Custom metadata unequal: left = ";
+ print_metadata(left);
+ ss << "\nright = ";
+ print_metadata(right);
+ return Status::Invalid(ss.str());
+ }
+
bool ok = true;
if ((left.batch != nullptr) != (right.batch != nullptr)) {
ok = false;
} else if (left.batch) {
ok &= left.batch->Equals(*right.batch, EqualOptions{}.nans_equal(true));
}
- return ok ? Status::OK() : Status::Invalid("Batches unequal");
+ if (ok) {
+ return Status::OK();
+ }
+ std::stringstream ss;
+ PrettyPrintOptions options{};
+ options.show_field_metadata = true;
+ options.show_schema_metadata = true;
+ auto print_batch = [&](const RecordBatchWithMetadata& batch) {
+ if (batch.batch) {
+ ss << "\n";
+ ARROW_UNUSED(PrettyPrint(*batch.batch, options, &ss));
+ } else {
+ ss << "nullptr";
+ }
+ };
+ ss << "Batches unequal: left = ";
+ print_batch(left);
+ ss << "\nright = ";
+ print_batch(right);
+ return Status::Invalid(ss.str());
}
Status CompareFuzzBatches(const std::vector<RecordBatchWithMetadata>& left,
@@ -2748,8 +2786,13 @@ Status FuzzIpcFile(const uint8_t* data, int64_t 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<std::vector<RecordBatchWithMetadata>> {
+ auto do_stream_read = [&]() -> Result<IpcReadResult> {
io::BufferReader buffer_reader(buffer);
// Skip magic bytes at the beginning
RETURN_NOT_OK(
@@ -2766,11 +2809,10 @@ Status FuzzIpcFile(const uint8_t* data, int64_t size) {
}
batches.push_back(batch);
}
- return batches;
+ return IpcReadResult{batch_reader->schema(), batches};
};
- auto do_file_read =
- [&](bool pre_buffer) -> Result<std::vector<RecordBatchWithMetadata>> {
+ 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()));
@@ -2779,46 +2821,66 @@ Status FuzzIpcFile(const uint8_t* data, int64_t size) {
RETURN_NOT_OK(batch_reader->PreBufferMetadata(/*indices=*/{}));
}
+ IpcReadResult result = {.schema = batch_reader->schema()};
const int n_batches = batch_reader->num_record_batches();
- std::vector<RecordBatchWithMetadata> 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);
- batches.push_back(batch);
+ result.batches.push_back(batch);
}
RETURN_NOT_OK(st);
- return batches;
+ return result;
};
- // Lazily-initialized if the IPC reader succeeds
- std::optional<Result<std::vector<RecordBatchWithMetadata>>> maybe_stream_batches;
+ // Persistent read result for differential fuzzing.
+ std::optional<IpcReadResult> maybe_read_result;
+
+ auto compare_result = [&](IpcReadResult new_result) {
+ // Differential fuzzing: compare results read using different APIs or options
+ // with one another.
+ if (maybe_read_result.has_value()) {
+ ARROW_CHECK_OK(CompareFuzzBatches(maybe_read_result->batches, new_result.batches));
+ } else {
+ maybe_read_result = std::move(new_result);
+ }
+ };
for (const bool pre_buffer : {false, true}) {
- auto maybe_file_batches = do_file_read(pre_buffer);
- final_status &= maybe_file_batches.status();
- if (maybe_file_batches.ok()) {
- // IPC file read successful: differential fuzzing with IPC stream reader,
- // if possible.
- // NOTE: some valid IPC files may not be readable as IPC streams,
- // for example because of excess spacing between IPC messages.
- // A regular IPC file writer would not produce them, but fuzzing might.
- if (!maybe_stream_batches.has_value()) {
- maybe_stream_batches = do_stream_read();
- final_status &= maybe_stream_batches->status();
- }
- if (maybe_stream_batches->ok()) {
+ auto maybe_file_result = do_file_read(pre_buffer);
+ final_status &= maybe_file_result.status();
+ if (maybe_file_result.ok()) {
+ compare_result(*std::move(maybe_file_result));
+ }
+ }
+
+ if (maybe_read_result.has_value()) {
+ // IPC file read successful: compare results with IPC stream reader,
+ // if possible.
+ // NOTE: some valid IPC files may not be readable as IPC streams,
+ // for example because of excess spacing between IPC messages.
+ // A regular IPC file writer would not produce them, but fuzzing might.
+ auto maybe_stream_result = do_stream_read();
+ final_status &= maybe_stream_result.status();
+ if (maybe_stream_result.ok()) {
+ if (maybe_read_result->schema->Equals(maybe_stream_result->schema,
+ /*check_metadata=*/true)) {
// XXX: in some rare cases, an IPC file might read unequal to the enclosed
// IPC stream, for example if the footer skips some batches or orders the
// batches differently. We should revisit this if the fuzzer generates such
// files.
- ARROW_CHECK_OK(CompareFuzzBatches(*maybe_file_batches, **maybe_stream_batches));
+ compare_result(*maybe_stream_result);
+ } else {
+ // The fuzzer might have mutated the schema definition that is duplicated
+ // in the IPC file footer, in which case the comparison above would fail.
+ final_status &= Status::TypeError(
+ "Schema mismatch between IPC stream and IPC file footer, skipping "
+ "comparison");
}
}
}
-
return final_status;
}
diff --git a/testing b/testing
index ca49b7795c..fff99f68b7 160000
--- a/testing
+++ b/testing
@@ -1 +1 @@
-Subproject commit ca49b7795c09181c2915b0a5e762a8fac70f9556
+Subproject commit fff99f68b7085b7ac1c210f79f1087f6a085a06b