Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 18 additions & 4 deletions parquet/src/encodings/decoding.rs
Original file line number Diff line number Diff line change
Expand Up @@ -847,10 +847,21 @@ where
self.values_left -= 1;
}

let mini_block_batch_size = match T::T::PHYSICAL_TYPE {
Type::INT32 => 32,
Type::INT64 => 64,
_ => unreachable!(),
// 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 {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

an error certainly seems better than a panic

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];
Expand All @@ -863,10 +874,13 @@ where
self.check_bit_width(bit_width)?;
let mini_block_to_skip = self.mini_block_remaining.min(to_skip - skip);

// see commentary in self.get() above regarding optimizations
let min_delta = self.min_delta.as_i64()?;
if bit_width == 0 {
// All remainders are zero: every delta equals min_delta exactly.
// Advance last_value by n * min_delta with no bit reads.
// When min_delta == 0 there is nothing to do: last_value is
// unchanged and no bytes are consumed from the bit reader.
if min_delta != 0 {
let total = min_delta.wrapping_mul(mini_block_to_skip as i64);
let step = T::T::from_i64(total)
Expand Down
29 changes: 28 additions & 1 deletion parquet/tests/arrow_reader/bad_data.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
//! Tests that reading invalid parquet files returns an error

use arrow::util::test_util::parquet_test_data;
use bytes::Bytes;
use parquet::arrow::arrow_reader::ArrowReaderBuilder;
use parquet::errors::ParquetError;
use std::collections::HashSet;
Expand Down Expand Up @@ -147,11 +148,37 @@ fn read_file(name: &str) -> Result<usize, ParquetError> {
Ok(num_rows)
}

#[test]
fn non_standard_delta_blocks() {
let file = Bytes::from_static(include_bytes!("bigdelta.parquet"));
use parquet::arrow::arrow_reader::{RowSelection, RowSelector};

let selectors = vec![RowSelector::skip(1000), RowSelector::select(5)];

let selection: RowSelection = selectors.into();
let reader = ArrowReaderBuilder::try_new(file)
.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")
);
}
}

#[cfg(feature = "async")]
#[tokio::test]
#[allow(deprecated)]
async fn bad_metadata_err() {
use bytes::Bytes;
use parquet::file::metadata::ParquetMetaDataReader;

let metadata_buffer = Bytes::from_static(include_bytes!("bad_raw_metadata.bin"));
Expand Down
Binary file added parquet/tests/arrow_reader/bigdelta.parquet
Comment thread
alamb marked this conversation as resolved.
Binary file not shown.
Loading