This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-11791-9ffd63d528e924d9c0faceef0027e3766b615c98 in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
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: // +--------------------------------------------------------+
