This is an automated email from the ASF dual-hosted git repository.
alamb 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 44141e61bb fix(parquet): skip miniblocks wider than 64 values instead
of erroring (#11021)
44141e61bb is described below
commit 44141e61bbc3cf1d470c0437ee3bbbbe92379904
Author: Bharadwaj Pendyala <[email protected]>
AuthorDate: Thu Sep 10 03:32:41 2026 -0500
fix(parquet): skip miniblocks wider than 64 values instead of erroring
(#11021)
# Which issue does this PR close?
- Closes #11018.
# Rationale for this change
`DeltaBitPackDecoder::skip` sized its scratch buffer with a `match` on
`values_per_mini_block` that only accepted 32 and 64, and returned
`cannot skip miniblock of size N` for anything else. The spec only
requires a multiple of 32, and Databricks Photon 0.2 writes 256, so
valid files from that engine fail to read as soon as a
`RowSelector::skip` is involved. `get` never had the restriction because
it chunks against the caller's buffer.
The scratch buffer only exists to walk `last_value` forward through the
deltas. `BitReader::get_batch` carries its own `bit_offset` and
`byte_offset`, so reading a 256 value miniblock as four calls of 64
leaves the reader exactly where one call of 256 would. That means the
buffer can stay small and the miniblock can be consumed a chunk at a
time, which was the concern that closed #9793 without a fix.
# What changes are included in this PR?
- Size `skip_buffer` at `values_per_mini_block.min(64)` and loop over
the miniblock in chunks of it, instead of rejecting sizes other than 32
and 64.
- Flip `non_standard_delta_blocks` in `bad_data.rs` from asserting the
error to asserting the skip lands on the right rows. That test already
carried a commented-out `skip should succeed` and a TODO for this
change.
For 32 and 64 the buffer is the same length it was before and the new
loop runs exactly once, so the path this crate's own writer produces
allocates and reads what it did before.
# Are these changes tested?
- New `test_delta_bit_packed_skip_wide_miniblocks` builds a hand-written
page with `block_size = 1024` over 4 miniblocks, so 256 values each. It
fails on `main` with `cannot skip miniblock of size 256` and passes
here. It checks the values after a skip that stops mid-miniblock, and
after one that crosses into a `bit_width = 0` miniblock, since both
depend on `last_value` surviving the chunk boundary.
- `non_standard_delta_blocks` now reads `bigdelta.parquet` twice, once
with the row selection and once without, and compares the 5 selected
rows against `all.slice(1000, 5)`. Asserting only `num_rows() == 5`
would pass on wrong values.
- `cargo test -p parquet --all-features`: 1426 lib tests and 456
integration tests pass, 0 failures.
- `cargo clippy -p parquet --all-features --all-targets`: clean.
- No benchmark numbers here because the change doesn't claim a speedup.
The argument that the common path is untouched is the buffer length and
loop count above, not a measurement.
# Are there any user-facing changes?
Reading a page with miniblocks wider than 64 values under a row
selection now returns data instead of an error. No public API changes.
There's also an existing overflow bug in `skip`. The `bit_width == 0`
branch of `skip` computes `min_delta * n` in `i64` and then converts to
`T::T`, so a valid INT32 progression can fail the conversion even though
every decoded value fits. A page with `first_value = -2000000000`,
`min_delta = 10000000` and zero-width miniblocks reads fine through
`get`, giving 560000000 at index 256, but `skip(257)` returns `delta*n
overflow in skip`. That's independent of miniblock width and predates
this change, so I've left it for a separate issue and fix.
# AI usage
Claude wrote the change and the tests, and Codex reviewed the diff
adversarially before it was pushed. I checked every finding myself: the
split-read equivalence against `BitReader::get_batch`, the
truncated-page error path, and the overflow above, which I reproduced
before writing it down. The PR description is AI-assisted.
---------
Co-authored-by: Ed Seidl <[email protected]>
---
parquet/src/encodings/decoding.rs | 129 +++++++++++++++++++++++----------
parquet/tests/arrow_reader/bad_data.rs | 29 +++++---
2 files changed, 107 insertions(+), 51 deletions(-)
diff --git a/parquet/src/encodings/decoding.rs
b/parquet/src/encodings/decoding.rs
index c884f3c2e0..9830cd9d5e 100644
--- a/parquet/src/encodings/decoding.rs
+++ b/parquet/src/encodings/decoding.rs
@@ -515,6 +515,11 @@ where
// ----------------------------------------------------------------------
// DELTA_BINARY_PACKED Decoding
+/// Upper bound on the scratch buffer [`DeltaBitPackDecoder::skip`] decodes
into.
+/// Matches the widest miniblock this crate writes, and the widest unpack
+/// [`BitReader::get_batch`] performs in one step for a 64 bit type.
+const MAX_SKIP_BUFFER_VALUES: usize = 64;
+
/// Delta binary packed decoder.
/// Supports INT32 and INT64 types.
/// See [`DeltaBitPackEncoder`](crate::encoding::DeltaBitPackEncoder) for more
@@ -852,23 +857,12 @@ where
}
// See https://github.com/apache/arrow-rs/pull/9794.
- // The parquet spec actually allows for miniblock sizes other than 32
or 64, but
- // no current writers use anything else. Using values_per_mini_block
directly
- // for the skip_buffer doesn't allow stack allocation and leads to a
significant
- // drop in performance. We'll settle for erroring out here and come up
with a
- // better fix if writers ever start getting creative with block sizes.
- let mini_block_batch_size = match self.values_per_mini_block {
- 32 => 32,
- 64 => 64,
- _ => {
- return Err(general_err!(
- "cannot skip miniblock of size {}",
- self.values_per_mini_block
- ));
- }
- };
-
- let mut skip_buffer = vec![T::T::default(); mini_block_batch_size];
+ // The parquet spec allows miniblock sizes other than the 32 and 64
this crate
+ // writes, and sizing skip_buffer off values_per_mini_block costs us
the stack
+ // allocation that keeps skip fast. The buffer only exists to walk
last_value
+ // forward, so keep it fixed at the widest size the common cases need
and
+ // consume wider miniblocks a chunk at a time.
+ let mut skip_buffer = [T::T::default(); MAX_SKIP_BUFFER_VALUES];
while skip < to_skip {
if self.mini_block_remaining == 0 {
self.next_mini_block()?;
@@ -894,30 +888,37 @@ where
// bit_width=0 payloads occupy zero bytes; no bit_reader
advancement needed.
} else {
// bw>0: must decode to track last_value for subsequent get()
calls.
- let skip_count = self
- .bit_reader
- .get_batch(&mut skip_buffer[0..mini_block_to_skip],
bit_width);
-
- if skip_count != mini_block_to_skip {
- return Err(general_err!(
- "Expected to skip {} values from mini block got {}.",
- mini_block_to_skip,
- skip_count
- ));
- }
-
- if min_delta == 0 {
- for v in &mut skip_buffer[0..skip_count] {
- *v = v.wrapping_add(&self.last_value);
- self.last_value = *v;
+ let mut skipped_in_mini_block = 0;
+ while skipped_in_mini_block < mini_block_to_skip {
+ let batch_to_skip =
+ (mini_block_to_skip -
skipped_in_mini_block).min(skip_buffer.len());
+ let skip_count = self
+ .bit_reader
+ .get_batch(&mut skip_buffer[0..batch_to_skip],
bit_width);
+
+ if skip_count != batch_to_skip {
+ return Err(general_err!(
+ "Expected to skip {} values from mini block got
{}.",
+ batch_to_skip,
+ skip_count
+ ));
}
- } else {
- for v in &mut skip_buffer[0..skip_count] {
- *v = v
- .wrapping_add(&self.min_delta)
- .wrapping_add(&self.last_value);
- self.last_value = *v;
+
+ if min_delta == 0 {
+ for v in &mut skip_buffer[0..skip_count] {
+ *v = v.wrapping_add(&self.last_value);
+ self.last_value = *v;
+ }
+ } else {
+ for v in &mut skip_buffer[0..skip_count] {
+ *v = v
+ .wrapping_add(&self.min_delta)
+ .wrapping_add(&self.last_value);
+ self.last_value = *v;
+ }
}
+
+ skipped_in_mini_block += batch_to_skip;
}
}
@@ -1912,6 +1913,56 @@ mod tests {
assert_eq!(result, vec![29, 43, 89]);
}
+ #[test]
+ fn test_delta_bit_packed_skip_wide_miniblocks() {
+ // 1024 values per block over 4 miniblocks is 256 values per
miniblock, which is
+ // spec legal but wider than anything this crate writes. Databricks
Photon 0.2
+ // emits pages shaped like this. See
+ // https://github.com/apache/arrow-rs/issues/11018.
+ let header = vec![
+ 128, 8, // block_size = 1024
+ 4, // mini_blocks_per_block = 4
+ 172, 2, // total_values = 300
+ 0, // first_value = 0
+ ];
+
+ let block_header = vec![
+ 0, // min_delta = 0
+ 1, 0, 0, 0, // bit widths
+ ];
+
+ // Miniblock 1 - bit width 1, all deltas 1 => 256 bits
+ // Miniblocks 2 to 4 - bit width 0 => no bytes
+ let block = vec![0xFF; 32];
+
+ let data: Vec<u8> = header
+ .into_iter()
+ .chain(block_header)
+ .chain(block)
+ .collect();
+ let data = Bytes::from(data);
+
+ // Values are 0..=256 followed by 43 more copies of 256.
+ let expected: Vec<i32> = (200..=256).chain(std::iter::repeat_n(256,
43)).collect();
+
+ let mut decoder = DeltaBitPackDecoder::<Int32Type>::new();
+ decoder.set_data(data.clone(), 0).unwrap();
+ assert_eq!(decoder.skip(200).unwrap(), 200);
+
+ let mut output = vec![0_i32; 100];
+ assert_eq!(decoder.get(&mut output).unwrap(), 100);
+ assert_eq!(output, expected);
+
+ // Skipping across the miniblock boundary must leave last_value on the
plateau.
+ let mut decoder = DeltaBitPackDecoder::<Int32Type>::new();
+ decoder.set_data(data, 0).unwrap();
+ assert_eq!(decoder.skip(299).unwrap(), 299);
+
+ let mut output = vec![0_i32; 1];
+ assert_eq!(decoder.get(&mut output).unwrap(), 1);
+ assert_eq!(output, vec![256]);
+ }
+
#[test]
fn test_delta_bit_packed_padding() {
// Page header
diff --git a/parquet/tests/arrow_reader/bad_data.rs
b/parquet/tests/arrow_reader/bad_data.rs
index 8587d15230..b22e78a32b 100644
--- a/parquet/tests/arrow_reader/bad_data.rs
+++ b/parquet/tests/arrow_reader/bad_data.rs
@@ -17,7 +17,9 @@
//! Tests that reading invalid parquet files returns an error
+use arrow::compute::concat_batches;
use arrow::util::test_util::parquet_test_data;
+use arrow_array::RecordBatchReader;
use bytes::Bytes;
use parquet::arrow::arrow_reader::ArrowReaderBuilder;
use parquet::errors::ParquetError;
@@ -171,23 +173,26 @@ fn non_standard_delta_blocks() {
let selectors = vec![RowSelector::skip(1000), RowSelector::select(5)];
let selection: RowSelection = selectors.into();
- let reader = ArrowReaderBuilder::try_new(file)
+ let reader = ArrowReaderBuilder::try_new(file.clone())
.unwrap()
.with_row_selection(selection)
.build()
.unwrap();
- if let Some(maybe_batch) = reader.into_iter().next() {
- // TODO: uncomment if we ever allow skipping miniblocks > 64 elements
- //let batch = maybe_batch.expect("skip should succeed");
- //assert_eq!(batch.num_rows(), 5);
- assert!(
- maybe_batch
- .unwrap_err()
- .to_string()
- .contains("cannot skip miniblock of size 128")
- );
- }
+ let selected = concat_batches(
+ &reader.schema(),
+ &reader.collect::<Result<Vec<_>, _>>().unwrap(),
+ )
+ .unwrap();
+ assert_eq!(selected.num_rows(), 5);
+
+ let reader = ArrowReaderBuilder::try_new(file).unwrap().build().unwrap();
+ let all = concat_batches(
+ &reader.schema(),
+ &reader.collect::<Result<Vec<_>, _>>().unwrap(),
+ )
+ .unwrap();
+ assert_eq!(selected, all.slice(1000, 5));
}
#[test]
