This is an automated email from the ASF dual-hosted git repository.
Jefffrey pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new cc49a83f44 fix(arrow-avro): fix split sync marker assembly and
validate per-block sync markers (#10497)
cc49a83f44 is described below
commit cc49a83f4422f8bcbcb06f908505112cf96bce3d
Author: ranflarion <[email protected]>
AuthorDate: Mon Aug 3 19:38:27 2026 -0400
fix(arrow-avro): fix split sync marker assembly and validate per-block sync
markers (#10497)
# Which issue does this PR close?
- Closes #10494.
- Closes #10495.
# Rationale for this change
Both `BlockDecoder` and `HeaderDecoder` assemble the 16-byte sync marker
with an offset derived from the size of the incoming fragment (`sync[16
- to_decode..]`), which only works when all 16 bytes arrive in one
`decode()` call; a marker straddling a chunk boundary gets scrambled
(#10494). For `HeaderDecoder` this is live today: a header larger than
one fetch whose trailing marker straddles the boundary yields a garbled
`Header::sync()`, the async reader's sync marker scan never matches, and
a valid file silently reads as an empty stream.
With assembly fixed, `Block.sync` becomes trustworthy, enabling the
validation the Avro spec intends the marker for: both readers now
compare each block's trailing marker against the file header's and
reject a mismatch, matching the Java reference reader (`DataFileStream`
throws `Invalid sync!`) (#10495). This continues the recent hardening of
the container decode path (#10237, #10407). Without it, a desynced or
corrupted stream can decode silently into wrong values; avro-deflate
block data is raw DEFLATE with no checksum, so this is a real
silent-corruption window with transiently faulty object stores.
# What changes are included in this PR?
- `BlockDecoder` and `HeaderDecoder` fill the sync marker from the front
using the consumed count (`offset = 16 - bytes_remaining`).
- The sync `Reader` and the async reader error with `ParseError("Avro
block sync marker does not match file header")` when a flushed block's
marker differs from the header's.
# Are these changes tested?
- A `BlockDecoder` unit test feeds a block in every chunk size from 1
byte up and asserts the assembled marker; it fails on main.
- Both readers get a test that flips the final byte of
`alltypes_plain.avro` (the last block's sync marker) and asserts the
read fails; on main the flip is accepted silently.
- Existing suites pass unchanged, including the roundtrip and range-read
tests.
# Are there any user-facing changes?
Corrupt files that previously decoded silently (or, for the header case,
silently produced an empty stream) now error, matching the Java reader.
Appended files reuse the file's marker, so any file the reference reader
accepts still reads fine.
Co-authored-by: Jeffrey Vo <[email protected]>
---
arrow-avro/src/reader/async_reader/mod.rs | 23 +++++++++++++++++++++++
arrow-avro/src/reader/block.rs | 25 +++++++++++++++++++++++--
arrow-avro/src/reader/header.rs | 5 +++--
arrow-avro/src/reader/mod.rs | 22 ++++++++++++++++++++++
4 files changed, 71 insertions(+), 4 deletions(-)
diff --git a/arrow-avro/src/reader/async_reader/mod.rs
b/arrow-avro/src/reader/async_reader/mod.rs
index 84ba1c873b..7ab0ee3efb 100644
--- a/arrow-avro/src/reader/async_reader/mod.rs
+++ b/arrow-avro/src/reader/async_reader/mod.rs
@@ -389,6 +389,11 @@ impl<R: AsyncFileReader + Unpin + 'static>
AsyncAvroFileReader<R> {
// If we reached the end of the block, flush it, and move
to read batches.
if let Some(block) = self.block_decoder.flush() {
// Successfully decoded a block.
+ if block.sync != self.sync_marker {
+ return
self.finish_with_error(AvroError::ParseError(
+ "Avro block sync marker does not match file
header".to_string(),
+ ));
+ }
let block_count = block.count;
let block_data = Bytes::from_owner(if let Some(ref
codec) = self.codec {
match codec.decompress(&block.data) {
@@ -1125,6 +1130,24 @@ mod tests {
assert!(batch.num_rows() > 0);
}
+ #[tokio::test]
+ async fn test_block_sync_marker_mismatch_errors() {
+ use tempfile::tempdir;
+ let file = arrow_test_data("avro/alltypes_plain.avro");
+ let mut bytes = std::fs::read(&file).unwrap();
+ // The file ends with the final block's 16-byte sync marker.
+ let last = bytes.len() - 1;
+ bytes[last] ^= 0xFF;
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("corrupt_sync.avro");
+ std::fs::write(&path, bytes).unwrap();
+ let schema = get_alltypes_schema();
+ let err = read_async_file(path.to_str().unwrap(), 1024, None,
Some(schema), None)
+ .await
+ .expect_err("corrupted block sync marker should fail the read");
+ assert!(err.to_string().contains("sync marker"), "{err}");
+ }
+
#[tokio::test]
async fn test_range_no_sync_marker() {
// Small range unlikely to contain sync marker
diff --git a/arrow-avro/src/reader/block.rs b/arrow-avro/src/reader/block.rs
index 3ec6e507ea..44161e2646 100644
--- a/arrow-avro/src/reader/block.rs
+++ b/arrow-avro/src/reader/block.rs
@@ -118,8 +118,10 @@ impl BlockDecoder {
}
BlockDecoderState::Sync => {
let to_decode = buf.len().min(self.bytes_remaining);
- let write = &mut self.in_progress.sync[16 - to_decode..];
- write[..to_decode].copy_from_slice(&buf[..to_decode]);
+ // Fill from the front: the marker may arrive split across
decode() calls.
+ let offset = 16 - self.bytes_remaining;
+ self.in_progress.sync[offset..offset + to_decode]
+ .copy_from_slice(&buf[..to_decode]);
self.bytes_remaining -= to_decode;
buf = &buf[to_decode..];
if self.bytes_remaining == 0 {
@@ -222,4 +224,23 @@ mod tests {
assert_eq!(block.data, payload);
assert_eq!(block.sync, sync);
}
+
+ #[test]
+ fn test_sync_marker_split_across_decode_calls() {
+ // count=1 (zig-zag 0x02), size=1 (0x02), one data byte, 16-byte sync
marker
+ let sync: [u8; 16] = core::array::from_fn(|i| i as u8);
+ let mut block_bytes = vec![0x02, 0x02, 0xAA];
+ block_bytes.extend_from_slice(&sync);
+
+ for chunk_size in 1..block_bytes.len() {
+ let mut decoder = BlockDecoder::default();
+ for chunk in block_bytes.chunks(chunk_size) {
+ decoder.decode(chunk).unwrap();
+ }
+ let block = decoder.flush().expect("complete block");
+ assert_eq!(block.count, 1);
+ assert_eq!(block.data, vec![0xAA]);
+ assert_eq!(block.sync, sync, "chunk_size {chunk_size}");
+ }
+ }
}
diff --git a/arrow-avro/src/reader/header.rs b/arrow-avro/src/reader/header.rs
index 235166bb76..9c28ef64fe 100644
--- a/arrow-avro/src/reader/header.rs
+++ b/arrow-avro/src/reader/header.rs
@@ -314,8 +314,9 @@ impl HeaderDecoder {
}
HeaderDecoderState::Sync => {
let to_decode = buf.len().min(self.bytes_remaining);
- let write = &mut self.sync_marker[16 - to_decode..];
- write[..to_decode].copy_from_slice(&buf[..to_decode]);
+ // Fill from the front: the marker may arrive split across
decode() calls.
+ let offset = 16 - self.bytes_remaining;
+ self.sync_marker[offset..offset +
to_decode].copy_from_slice(&buf[..to_decode]);
self.bytes_remaining -= to_decode;
buf = &buf[to_decode..];
if self.bytes_remaining == 0 {
diff --git a/arrow-avro/src/reader/mod.rs b/arrow-avro/src/reader/mod.rs
index 1b966060e4..a9ec035a70 100644
--- a/arrow-avro/src/reader/mod.rs
+++ b/arrow-avro/src/reader/mod.rs
@@ -1371,6 +1371,11 @@ impl<R: BufRead> Reader<R> {
self.reader.consume(consumed);
if let Some(block) = self.block_decoder.flush() {
// Successfully decoded a block.
+ if block.sync != self.header.sync() {
+ return Err(AvroError::ParseError(
+ "Avro block sync marker does not match file
header".to_string(),
+ ));
+ }
self.block_data = if let Some(ref codec) =
self.header.compression()? {
let decompressed: Vec<u8> =
codec.decompress(&block.data)?;
decompressed
@@ -1492,6 +1497,23 @@ mod test {
arrow::compute::concat_batches(&schema, &batches).unwrap()
}
+ #[test]
+ fn test_block_sync_marker_mismatch_errors() {
+ let path = arrow_test_data("avro/alltypes_plain.avro");
+ let mut bytes = std::fs::read(&path).unwrap();
+ // The file ends with the final block's 16-byte sync marker.
+ let last = bytes.len() - 1;
+ bytes[last] ^= 0xFF;
+ let reader = ReaderBuilder::new()
+ .with_batch_size(1024)
+ .build(std::io::Cursor::new(bytes))
+ .unwrap();
+ let err = reader
+ .collect::<Result<Vec<_>, _>>()
+ .expect_err("corrupted block sync marker should fail the read");
+ assert!(err.to_string().contains("sync marker"), "{err}");
+ }
+
fn read_file_strict(
path: &str,
batch_size: usize,
