This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 44f6371167 fix(parquet): return byte array overflow and map decoding
errors (#11791)
44f6371167 is described below
commit 44f6371167979c83d634fbe04f6a5c60d79a4bbb
Author: Huang Qiwei <[email protected]>
AuthorDate: Fri Oct 30 16:18:20 2026 +0000
fix(parquet): return byte array overflow and map decoding errors (#11791)
# Which issue does this PR close?
- Relates to #7973 and #11478.
First part(1/4) of the solution. Whole solution can be found here:
https://github.com/apache/arrow-rs/issues/11478#issuecomment-6333545397
This addresses panic handling only. Reading oversized `Utf8`/`Binary`
batches successfully and bounding Mask decoding memory are separate
changes.
# Rationale for this change
Reading `DELTA_LENGTH_BYTE_ARRAY` data can panic when the cumulative
payload exceeds the output offset range. Map decoding can also panic
when an underlying reader returns an error.
These failures should return explicit errors through the normal reader
error path.
# What changes are included in this PR?
- Use checked arithmetic when validating delta-length page lengths.
- Check cumulative output offsets before reserving or copying payloads.
On overflow, return an error without modifying the output or advancing
the decoder.
- Preserve the existing bulk-copy path for valid input.
- Propagate Map child-reader errors instead of unwrapping them.
# Are these changes tested?
Yes. Added regression tests cover:
- Offset boundaries, integer overflow, and values that fit 64-bit
offsets but exceed 32-bit offsets.
- Appending decoded values, empty strings, skipping, and end of input.
- Propagation of errors from malformed Map data.
- A real large-buffer overflow, verifying that the output and decoder
state remain unchanged and subsequent reads still work. This
memory-intensive test is ignored by default and was run explicitly.
Both panic regressions fail against the original implementation and pass
with these changes. Workspace tests and the additional Parquet
async/experimental tests also pass.
# Are there any significant user-facing changes?
No additional changes beyond the error handling described above. There
are no public API changes.
AI assistance: OpenAI Codex assisted with implementation and regression
tests.
---
parquet/src/arrow/array_reader/byte_array.rs | 102 +++++++++++++++++++++++++--
parquet/src/arrow/array_reader/map_array.rs | 36 +++++++++-
2 files changed, 133 insertions(+), 5 deletions(-)
diff --git a/parquet/src/arrow/array_reader/byte_array.rs
b/parquet/src/arrow/array_reader/byte_array.rs
index 1641fd0219..444163f743 100644
--- a/parquet/src/arrow/array_reader/byte_array.rs
+++ b/parquet/src/arrow/array_reader/byte_array.rs
@@ -459,7 +459,7 @@ impl ByteArrayDecoderDeltaLength {
let mut lengths = vec![0; values];
len_decoder.get(&mut lengths)?;
- let mut total_bytes = 0;
+ let mut total_bytes = len_decoder.get_offset();
for l in &lengths {
if *l < 0 {
@@ -467,10 +467,12 @@ impl ByteArrayDecoderDeltaLength {
"negative delta length byte array length".to_string(),
));
}
- total_bytes += *l as usize;
+ total_bytes = total_bytes
+ .checked_add(*l as usize)
+ .ok_or_else(|| general_err!("delta length byte array length
overflow"))?;
}
- if total_bytes + len_decoder.get_offset() > data.len() {
+ if total_bytes > data.len() {
return Err(ParquetError::General(
"Insufficient delta length byte array bytes".to_string(),
));
@@ -496,6 +498,8 @@ impl ByteArrayDecoderDeltaLength {
let src_lengths = &self.lengths[self.length_offset..self.length_offset
+ to_read];
let total_bytes: usize = src_lengths.iter().map(|x| *x as usize).sum();
+ Self::checked_offset::<I>(initial_values_length, total_bytes)?;
+
// Reserve capacity for both offsets and values upfront
output.offsets.reserve(to_read);
output.values.reserve(total_bytes);
@@ -511,7 +515,8 @@ impl ByteArrayDecoderDeltaLength {
let mut running = base_offset;
output.offsets.extend(src_lengths.iter().map(|length| {
running += *length as usize;
- I::from_usize(running).expect("index overflow decoding byte array")
+ // Nonnegative lengths and the checked final offset make every
offset valid.
+ I::usize_as(running)
}));
self.data_offset = data_end;
@@ -523,6 +528,13 @@ impl ByteArrayDecoderDeltaLength {
Ok(to_read)
}
+ fn checked_offset<I: OffsetSizeTrait>(offset: usize, additional: usize) ->
Result<I> {
+ offset
+ .checked_add(additional)
+ .and_then(I::from_usize)
+ .ok_or_else(|| general_err!("index overflow decoding byte array"))
+ }
+
fn skip(&mut self, to_skip: usize) -> Result<usize> {
let remain_values = self.lengths.len() - self.length_offset;
let to_skip = remain_values.min(to_skip);
@@ -767,4 +779,86 @@ mod tests {
assert_eq!(output.offsets.as_slice(), &[0, 3]);
assert_eq!(decoder.max_remaining_values, 1);
}
+
+ #[test]
+ fn test_delta_length_checked_offset() {
+ let max = i32::MAX as usize;
+ assert_eq!(
+ ByteArrayDecoderDeltaLength::checked_offset::<i32>(max - 1,
1).unwrap(),
+ i32::MAX
+ );
+ assert_eq!(
+ ByteArrayDecoderDeltaLength::checked_offset::<i32>(max,
0).unwrap(),
+ i32::MAX
+ );
+ for (offset, additional) in [(max, 1), (0, max + 1), (usize::MAX, 1)] {
+ let error =
+ ByteArrayDecoderDeltaLength::checked_offset::<i32>(offset,
additional).unwrap_err();
+ assert_eq!(
+ error.to_string(),
+ "Parquet error: index overflow decoding byte array"
+ );
+ }
+ assert_eq!(
+ ByteArrayDecoderDeltaLength::checked_offset::<i64>(max,
1).unwrap(),
+ i32::MAX as i64 + 1
+ );
+ assert!(ByteArrayDecoderDeltaLength::checked_offset::<i64>(usize::MAX,
1).is_err());
+ }
+
+ #[test]
+ fn test_delta_length_read_append_and_skip() {
+ let (pages, _) = byte_array_all_encodings(vec!["ab", "", "cde", "f"]);
+ let (_, page) = pages
+ .into_iter()
+ .find(|(encoding, _)| *encoding ==
Encoding::DELTA_LENGTH_BYTE_ARRAY)
+ .unwrap();
+ let mut decoder = ByteArrayDecoderDeltaLength::new(page,
true).unwrap();
+ let mut output = OffsetBuffer::<i32>::with_capacity(0);
+ output.try_push(b"prefix", true).unwrap();
+
+ assert_eq!(decoder.read(&mut output, 2).unwrap(), 2);
+ assert_eq!(output.values.as_slice(), b"prefixab");
+ assert_eq!(output.offsets.as_slice(), &[0, 6, 8, 8]);
+ assert_eq!(decoder.skip(1).unwrap(), 1);
+ assert_eq!(decoder.read(&mut output, 1).unwrap(), 1);
+ assert_eq!(output.values.as_slice(), b"prefixabf");
+ assert_eq!(output.offsets.as_slice(), &[0, 6, 8, 8, 9]);
+ assert_eq!(decoder.read(&mut output, 1).unwrap(), 0);
+ }
+
+ // The checked_offset unit tests cover arithmetic; this also verifies that
+ // read rejects a real overflow before changing output or decoder state.
+ // Keep this regression available for manual runs without requiring an
+ // allocation of about 2 GiB in routine CI.
+ #[test]
+ #[ignore = "allocates about 2 GiB; run --ignored --test-threads=1"]
+ fn test_delta_length_offset_overflow_preserves_state() {
+ let (pages, _) = byte_array_all_encodings(vec!["", "a"]);
+ let (_, page) = pages
+ .into_iter()
+ .find(|(encoding, _)| *encoding ==
Encoding::DELTA_LENGTH_BYTE_ARRAY)
+ .unwrap();
+ let mut decoder = ByteArrayDecoderDeltaLength::new(page,
false).unwrap();
+ let data_offset = decoder.data_offset;
+ let mut output = OffsetBuffer::<i32> {
+ offsets: vec![0, i32::MAX],
+ values: vec![0; i32::MAX as usize],
+ };
+
+ let error = decoder.read(&mut output, 2).unwrap_err();
+ assert_eq!(
+ error.to_string(),
+ "Parquet error: index overflow decoding byte array"
+ );
+ assert_eq!(output.values.len(), i32::MAX as usize);
+ assert_eq!(output.offsets, vec![0, i32::MAX]);
+ assert_eq!(decoder.data_offset, data_offset);
+ assert_eq!(decoder.length_offset, 0);
+
+ assert_eq!(decoder.read(&mut output, 1).unwrap(), 1);
+ assert_eq!(output.offsets, vec![0, i32::MAX, i32::MAX]);
+ assert_eq!(decoder.skip(1).unwrap(), 1);
+ assert_eq!(decoder.read(&mut output, 1).unwrap(), 0);
+ }
}
diff --git a/parquet/src/arrow/array_reader/map_array.rs
b/parquet/src/arrow/array_reader/map_array.rs
index 8bc847b386..c8489f6df7 100644
--- a/parquet/src/arrow/array_reader/map_array.rs
+++ b/parquet/src/arrow/array_reader/map_array.rs
@@ -98,7 +98,7 @@ impl ArrayReader for MapArrayReader {
fn consume_batch(&mut self) -> Result<ArrayRef> {
// A MapArray is just a ListArray with a StructArray child
// we can therefore just alter the ArrayData
- let array = self.reader.consume_batch().unwrap();
+ let array = self.reader.consume_batch()?;
let data = array.to_data();
let builder = data.into_builder().data_type(self.data_type.clone());
@@ -139,6 +139,40 @@ mod tests {
use arrow_schema::Fields;
use bytes::Bytes;
+ #[test]
+ fn test_map_reader_propagates_mismatched_child_lengths() {
+ use crate::arrow::array_reader::test_util::make_int32_page_reader;
+ use crate::errors::ParquetError;
+
+ // One map row has two keys but only one value.
+ let keys = make_int32_page_reader(&[1, 2], &[1, 1], &[0, 1], 1, 1,
Some(1));
+ let values = make_int32_page_reader(&[10], &[1], &[0], 1, 1, Some(1));
+ let entries = Arc::new(Field::new(
+ "entries",
+ ArrowType::Struct(Fields::from(vec![
+ Field::new("key", ArrowType::Int32, false),
+ Field::new("value", ArrowType::Int32, false),
+ ])),
+ false,
+ ));
+ let mut reader = MapArrayReader::new(
+ keys,
+ values,
+ ArrowType::Map(entries, false),
+ 1,
+ 1,
+ false,
+ None,
+ );
+
+ assert_eq!(reader.read_records(1).unwrap(), 1);
+ let error = reader.consume_batch().unwrap_err();
+ assert!(matches!(
+ error,
+ ParquetError::General(message) if message == "Not all children
array length are the same!"
+ ));
+ }
+
#[test]
// This test writes a parquet file with the following data:
// +--------------------------------------------------------+
