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-11051-efb652c27242870f35e009df26d7d84a0c1ee06a in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
commit 3c5bc5f4ce8f13f0974e5f01fdbed965d206172e Author: Yifan Chen <[email protected]> AuthorDate: Wed Oct 7 04:27:44 2026 +0000 Allow configuring DELTA_BINARY_PACKED block layout (#11051) # Which issue does this PR close? - Closes #2282. # Rationale for this change The DELTA_BINARY_PACKED encoder currently fixes its layout at 128 values per block and four mini blocks for INT32, and 256 values per block and four mini blocks for INT64. The Parquet stream records this layout, and representative workloads can benefit from different compression and throughput trade-offs. # What changes are included in this PR? This adds validated DeltaBinaryPackedEncoderOptions and global/per-column WriterProperties setters. The column writer passes the full resolved column properties to the encoder factory. Custom layouts now configure direct DELTA_BINARY_PACKED values as well as the length, prefix-length, and suffix-length integer encoders inside DELTA_LENGTH_BYTE_ARRAY and DELTA_BYTE_ARRAY. The custom encoder constructor is infallible because it only accepts already validated options. Existing get_encoder callers and type-specific defaults remain unchanged. Public documentation links to the Parquet format and explains the block-size and mini-block trade-offs. # Benchmark evidence The committed Criterion benchmark compares four layouts on 16,384 deterministic INT32 values and reports encoded bytes in each benchmark ID. On this arm64 macOS host, comparing the INT32 default 128/4 layout with 256/4: - Uniform-width random deltas: 66,699 to 66,123 bytes (0.86% smaller), median encode time 66.76 to 62.71 us (6.1% faster), and median decode time 9.71 to 9.16 us (5.7% faster). - Locally varying bit widths: 35,131 to 42,161 bytes (20.0% larger), median encode time was effectively unchanged at 61.70 versus 61.68 us, and median decode time fell from 10.59 to 9.22 us (12.9% faster). These results support retaining the defaults generally. A larger layout can help long inputs with stable delta distributions, but bursty local ranges can materially increase size; users should benchmark representative data. Results are machine and workload specific. Benchmark command: cargo bench --locked -p parquet --bench bit_packing --features "experimental default" -- delta_binary_packed_layout --sample-size 20 --warm-up-time 1 --measurement-time 2 # Are these changes tested? Yes. - cargo test --locked -p parquet --all-features --lib --tests - cargo clippy --locked -p parquet --all-targets --all-features -- -D warnings - cargo fmt --all -- --check - git diff --check - The focused layout benchmark above # Are there any user-facing changes? Yes. This adds an additive public configuration API for all encoders that use DELTA_BINARY_PACKED internally. Existing defaults and behavior are unchanged unless users opt in. # AI assistance This PR was prepared with OpenAI Codex assistance for investigation, implementation, benchmark construction, test execution, and review-response drafting. --------- Co-authored-by: Ed Seidl <[email protected]> --- parquet/benches/bit_packing.rs | 73 +++++++- parquet/src/column/writer/encoder.rs | 5 +- parquet/src/encodings/encoding/mod.rs | 324 ++++++++++++++++++++++++++++++++-- parquet/src/file/properties.rs | 105 +++++++++++ 4 files changed, 485 insertions(+), 22 deletions(-) diff --git a/parquet/benches/bit_packing.rs b/parquet/benches/bit_packing.rs index 079928d540..52ffb6dfad 100644 --- a/parquet/benches/bit_packing.rs +++ b/parquet/benches/bit_packing.rs @@ -20,12 +20,15 @@ use std::sync::Arc; use arrow_array::{ArrayRef, StringArray}; use arrow_schema::{DataType as ArrowDataType, Field, Schema}; -use criterion::{BatchSize, Criterion, Throughput, criterion_group, criterion_main}; +use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; use parquet::arrow::ArrowSchemaConverter; use parquet::arrow::arrow_writer::{ArrowRowGroupWriterFactory, compute_leaves}; use parquet::basic::Encoding; use parquet::data_type::{BoolType, ByteArray, ByteArrayType, DataType, Int32Type, Int64Type}; -use parquet::encoding::{DictEncoder, Encoder, get_encoder}; +use parquet::decoding::{Decoder, DeltaBitPackDecoder}; +use parquet::encoding::{ + DeltaBinaryPackedEncoderOptions, DeltaBitPackEncoder, DictEncoder, Encoder, get_encoder, +}; use parquet::encodings::levels::LevelEncoder; use parquet::file::properties::WriterProperties; use parquet::file::writer::SerializedFileWriter; @@ -61,6 +64,57 @@ fn bench_encoding<T: DataType>(c: &mut Criterion, name: &str, values: &[T::T], e group.finish(); } +fn bench_delta_binary_packed_layouts(c: &mut Criterion, name: &str, values: &[i32]) { + let layouts = [(128, 4), (256, 8), (256, 4), (512, 8)]; + let mut group = c.benchmark_group(format!("delta_binary_packed_layout/i32/{name}")); + group.throughput(Throughput::Elements(values.len() as u64)); + + for (block_size, mini_blocks_per_block) in layouts { + let options = + DeltaBinaryPackedEncoderOptions::try_new(block_size, mini_blocks_per_block).unwrap(); + let mut encoder = DeltaBitPackEncoder::<Int32Type>::new_with_options(options); + encoder.put(values).unwrap(); + let encoded = encoder.flush_buffer().unwrap(); + let layout = format!( + "{block_size}/{mini_blocks_per_block}/{} bytes", + encoded.len() + ); + + group.bench_with_input( + BenchmarkId::new("encode", &layout), + &options, + |b, options| { + b.iter(|| { + let mut encoder = DeltaBitPackEncoder::<Int32Type>::new_with_options(*options); + encoder.put(black_box(values)).unwrap(); + black_box(encoder.flush_buffer().unwrap()); + }); + }, + ); + + group.bench_with_input( + BenchmarkId::new("decode", &layout), + &encoded, + |b, encoded| { + b.iter_batched( + || { + ( + DeltaBitPackDecoder::<Int32Type>::new(), + vec![0; values.len()], + ) + }, + |(mut decoder, mut output)| { + decoder.set_data(encoded.clone(), values.len()).unwrap(); + black_box(decoder.get(&mut output).unwrap()); + }, + BatchSize::SmallInput, + ); + }, + ); + } + group.finish(); +} + /// Dictionary values are encoded separately from their indices. Populate the /// dictionary outside the timed section and measure the affected RLE/bit-packed /// index-writing path. @@ -141,6 +195,7 @@ fn criterion_benchmark(c: &mut Criterion) { let mut rng = StdRng::seed_from_u64(0); let mut bools = Vec::with_capacity(NUM_VALUES); let mut i32s = Vec::with_capacity(NUM_VALUES); + let mut locally_varying_i32s = Vec::with_capacity(NUM_VALUES); let mut i64s = Vec::with_capacity(NUM_VALUES); let mut byte_arrays = Vec::with_capacity(NUM_VALUES); let mut dictionary_values = Vec::with_capacity(NUM_VALUES); @@ -162,6 +217,18 @@ fn criterion_benchmark(c: &mut Criterion) { levels.push(rng.random_range(0..=3)); } + let mut locally_varying_rng = StdRng::seed_from_u64(1); + let mut locally_varying_value = 0_i32; + for i in 0..NUM_VALUES { + let delta = if (i / 32) % 8 == 0 { + locally_varying_rng.random_range(-1_000_000..=1_000_000) + } else { + locally_varying_rng.random_range(-4..=4) + }; + locally_varying_value = locally_varying_value.wrapping_add(delta); + locally_varying_i32s.push(locally_varying_value); + } + // Direct BitWriter and RLE/bit-packed hybrid users. bench_encoding::<BoolType>(c, "plain/bool/bit_packed", &bools, Encoding::PLAIN); bench_encoding::<BoolType>(c, "rle/bool/bit_packed", &bools, Encoding::RLE); @@ -179,6 +246,8 @@ fn criterion_benchmark(c: &mut Criterion) { &i64s, Encoding::DELTA_BINARY_PACKED, ); + bench_delta_binary_packed_layouts(c, "uniform_bit_width", &i32s); + bench_delta_binary_packed_layouts(c, "locally_varying_bit_width", &locally_varying_i32s); // These byte-array encodings transitively use DELTA_BINARY_PACKED for // lengths, prefix lengths, and suffix lengths. diff --git a/parquet/src/column/writer/encoder.rs b/parquet/src/column/writer/encoder.rs index cd0d2a267c..4939e19ce4 100644 --- a/parquet/src/column/writer/encoder.rs +++ b/parquet/src/column/writer/encoder.rs @@ -24,7 +24,7 @@ use crate::column::writer::{ }; use crate::data_type::DataType; use crate::data_type::private::ParquetValueType; -use crate::encodings::encoding::{DictEncoder, Encoder, get_encoder}; +use crate::encodings::encoding::{DictEncoder, Encoder, get_encoder_with_properties}; use crate::errors::{ParquetError, Result}; use crate::file::properties::{EnabledStatistics, ResolvedColumnProperties, WriterProperties}; use crate::geospatial::accumulator::{GeoStatsAccumulator, try_new_geo_stats_accumulator}; @@ -281,11 +281,12 @@ impl<T: DataType> ColumnValueEncoder for ColumnValueEncoderImpl<T> { let dict_encoder = dict_supported.then(|| DictEncoder::new(descr.clone())); // Set either main encoder or fallback encoder. - let encoder = get_encoder( + let encoder = get_encoder_with_properties( column_props .encoding .unwrap_or_else(|| fallback_encoding(T::get_physical_type(), props)), descr, + column_props, )?; let statistics_enabled = column_props.statistics_enabled; diff --git a/parquet/src/encodings/encoding/mod.rs b/parquet/src/encodings/encoding/mod.rs index 8b38d4940f..be53e94a22 100644 --- a/parquet/src/encodings/encoding/mod.rs +++ b/parquet/src/encodings/encoding/mod.rs @@ -24,6 +24,7 @@ use crate::data_type::private::ParquetValueType; use crate::data_type::*; use crate::encodings::rle::RleEncoder; use crate::errors::{ParquetError, Result}; +use crate::file::properties::ResolvedColumnProperties; use crate::schema::types::ColumnDescPtr; use crate::util::bit_util::{BitWriter, num_required_bits}; use crate::util::prefix::common_prefix_length; @@ -87,7 +88,15 @@ pub fn get_encoder<T: DataType>( encoding: Encoding, descr: &ColumnDescPtr, ) -> Result<Box<dyn Encoder<T>>> { - <T::T as private::GetEncoder>::get_encoder(descr, encoding) + <T::T as private::GetEncoder>::get_encoder(descr, encoding, None) +} + +pub(crate) fn get_encoder_with_properties<T: DataType>( + encoding: Encoding, + descr: &ColumnDescPtr, + column_props: &ResolvedColumnProperties, +) -> Result<Box<dyn Encoder<T>>> { + <T::T as private::GetEncoder>::get_encoder(descr, encoding, Some(column_props)) } pub(crate) mod private { @@ -101,17 +110,23 @@ pub(crate) mod private { /// /// [`ParquetValueType`]: crate::data_type::private::ParquetValueType pub trait GetEncoder { + #[expect( + private_interfaces, + reason = "this trait is not nameable outside the crate" + )] fn get_encoder<T: DataType<T = Self>>( descr: &ColumnDescPtr, encoding: Encoding, + options: Option<&ResolvedColumnProperties>, ) -> Result<Box<dyn Encoder<T>>> { - get_encoder_default(descr, encoding) + get_encoder_default(descr, encoding, options) } } fn get_encoder_default<T: DataType>( descr: &ColumnDescPtr, encoding: Encoding, + options: Option<&ResolvedColumnProperties>, ) -> Result<Box<dyn Encoder<T>>> { let encoder: Box<dyn Encoder<T>> = match encoding { Encoding::PLAIN => Box::new(PlainEncoder::new()), @@ -125,17 +140,30 @@ pub(crate) mod private { physical_type => return Err(unsupported_column_encoding(encoding, physical_type)), }, Encoding::DELTA_BINARY_PACKED => match T::get_physical_type() { - Type::INT32 | Type::INT64 => Box::new(DeltaBitPackEncoder::new()), + Type::INT32 | Type::INT64 => Box::new( + match options.and_then(|props| props.delta_binary_packed_encoder_options) { + Some(options) => DeltaBitPackEncoder::new_with_options(options), + None => DeltaBitPackEncoder::new(), + }, + ), physical_type => return Err(unsupported_column_encoding(encoding, physical_type)), }, Encoding::DELTA_LENGTH_BYTE_ARRAY => match T::get_physical_type() { - Type::BYTE_ARRAY => Box::new(DeltaLengthByteArrayEncoder::new()), + Type::BYTE_ARRAY => Box::new( + match options.and_then(|props| props.delta_binary_packed_encoder_options) { + Some(options) => DeltaLengthByteArrayEncoder::new_with_options(options), + None => DeltaLengthByteArrayEncoder::new(), + }, + ), physical_type => return Err(unsupported_column_encoding(encoding, physical_type)), }, Encoding::DELTA_BYTE_ARRAY => match T::get_physical_type() { - Type::BYTE_ARRAY | Type::FIXED_LEN_BYTE_ARRAY => { - Box::new(DeltaByteArrayEncoder::new()) - } + Type::BYTE_ARRAY | Type::FIXED_LEN_BYTE_ARRAY => Box::new( + match options.and_then(|props| props.delta_binary_packed_encoder_options) { + Some(options) => DeltaByteArrayEncoder::new_with_options(options), + None => DeltaByteArrayEncoder::new(), + }, + ), physical_type => return Err(unsupported_column_encoding(encoding, physical_type)), }, Encoding::BYTE_STREAM_SPLIT => match T::get_physical_type() { @@ -168,25 +196,35 @@ pub(crate) mod private { impl GetEncoder for FixedLenByteArray {} impl GetEncoder for f32 { + #[expect( + private_interfaces, + reason = "this trait is not nameable outside the crate" + )] fn get_encoder<T: DataType<T = Self>>( descr: &ColumnDescPtr, encoding: Encoding, + options: Option<&ResolvedColumnProperties>, ) -> Result<Box<dyn Encoder<T>>> { match encoding { Encoding::ALP => Ok(Box::new(AlpEncoder::new())), - _ => get_encoder_default(descr, encoding), + _ => get_encoder_default(descr, encoding, options), } } } impl GetEncoder for f64 { + #[expect( + private_interfaces, + reason = "this trait is not nameable outside the crate" + )] fn get_encoder<T: DataType<T = Self>>( descr: &ColumnDescPtr, encoding: Encoding, + options: Option<&ResolvedColumnProperties>, ) -> Result<Box<dyn Encoder<T>>> { match encoding { Encoding::ALP => Ok(Box::new(AlpEncoder::new())), - _ => get_encoder_default(descr, encoding), + _ => get_encoder_default(descr, encoding, options), } } } @@ -372,6 +410,68 @@ impl<T: DataType> Encoder<T> for RleValueEncoder<T> { const MAX_PAGE_HEADER_WRITER_SIZE: usize = 32; const DEFAULT_NUM_MINI_BLOCKS: usize = 4; +/// Controls the block layout emitted by [`DeltaBitPackEncoder`] and the integer sub-encoders used +/// by [`DeltaLengthByteArrayEncoder`] and [`DeltaByteArrayEncoder`]. +/// +/// In the [Parquet DELTA_BINARY_PACKED encoding], each block records a minimum delta and each mini +/// block records its own bit width. The defaults use 128 values per block and 4 mini blocks for +/// `INT32`, and 256 values per block and 4 mini blocks for `INT64`. +/// +/// Smaller blocks or mini blocks can reduce encoded size for bursty data, where the required bit +/// width changes over short ranges, and for short pages that would otherwise need more padding. +/// They also add more minimum-delta and bit-width metadata. Larger blocks or mini blocks can reduce +/// that metadata and block-processing overhead for long inputs with stable delta distributions, +/// improving compression or throughput, but an outlier then affects more values and the encoder +/// uses a larger scratch buffer. Benchmark representative data before changing the defaults. +/// +/// The format requires block sizes to be multiples of 128 and mini block sizes to be multiples +/// of 32. +/// +/// [Parquet DELTA_BINARY_PACKED encoding]: https://github.com/apache/parquet-format/blob/master/Encodings.md#delta-encoding-delta_binary_packed--5 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DeltaBinaryPackedEncoderOptions { + block_size: usize, + mini_blocks_per_block: usize, +} + +impl DeltaBinaryPackedEncoderOptions { + /// Creates validated delta binary packed encoder options. + pub fn try_new(block_size: usize, mini_blocks_per_block: usize) -> Result<Self> { + if block_size == 0 || !block_size.is_multiple_of(128) { + return Err(general_err!( + "delta binary packed block size must be a positive multiple of 128, got {block_size}" + )); + } + if mini_blocks_per_block == 0 || !block_size.is_multiple_of(mini_blocks_per_block) { + return Err(general_err!( + "delta binary packed mini block count must be a positive divisor of the block size, got {mini_blocks_per_block} for block size {block_size}" + )); + } + + let mini_block_size = block_size / mini_blocks_per_block; + if !mini_block_size.is_multiple_of(32) { + return Err(general_err!( + "delta binary packed mini block size must be a multiple of 32, got {mini_block_size}" + )); + } + + Ok(Self { + block_size, + mini_blocks_per_block, + }) + } + + /// Returns the number of values in each block. + pub fn block_size(&self) -> usize { + self.block_size + } + + /// Returns the number of mini blocks in each block. + pub fn mini_blocks_per_block(&self) -> usize { + self.mini_blocks_per_block + } +} + /// Delta bit packed encoder. /// Consists of a header followed by blocks of delta encoded values binary packed. /// @@ -431,6 +531,10 @@ impl<T: DataType> DeltaBitPackEncoder<T> { let block_size = mini_block_size * num_mini_blocks; assert_eq!(block_size % 128, 0); + Self::new_with_layout(block_size, num_mini_blocks) + } + + fn new_with_layout(block_size: usize, num_mini_blocks: usize) -> Self { DeltaBitPackEncoder { page_header_writer: BitWriter::new(MAX_PAGE_HEADER_WRITER_SIZE), // Don't pre-allocate: encoders are often created eagerly (e.g. as a @@ -441,7 +545,7 @@ impl<T: DataType> DeltaBitPackEncoder<T> { first_value: 0, current_value: 0, // current value to keep adding deltas block_size, // can write fewer values than block size for last block - mini_block_size, + mini_block_size: block_size / num_mini_blocks, num_mini_blocks, values_in_block: 0, // will be at most block_size deltas: vec![0; block_size], @@ -449,6 +553,15 @@ impl<T: DataType> DeltaBitPackEncoder<T> { } } + /// Creates a delta bit packed encoder with a custom block layout. + pub fn new_with_options(options: DeltaBinaryPackedEncoderOptions) -> Self { + Self::assert_supported_type(); + + let block_size = options.block_size(); + let num_mini_blocks = options.mini_blocks_per_block(); + Self::new_with_layout(block_size, num_mini_blocks) + } + /// Writes page header for blocks, this method is invoked when we are done encoding /// values. It is also okay to encode when no values have been provided fn write_page_header(&mut self) { @@ -692,6 +805,16 @@ impl<T: DataType> DeltaLengthByteArrayEncoder<T> { _phantom: PhantomData, } } + + /// Creates a delta length byte array encoder with a custom layout for its length encoder. + pub fn new_with_options(options: DeltaBinaryPackedEncoderOptions) -> Self { + Self { + len_encoder: DeltaBitPackEncoder::new_with_options(options), + data: vec![], + encoded_size: 0, + _phantom: PhantomData, + } + } } impl<T: DataType> Encoder<T> for DeltaLengthByteArrayEncoder<T> { @@ -781,6 +904,17 @@ impl<T: DataType> DeltaByteArrayEncoder<T> { _phantom: PhantomData, } } + + /// Creates a delta byte array encoder with a custom layout for its prefix and suffix length + /// encoders. + pub fn new_with_options(options: DeltaBinaryPackedEncoderOptions) -> Self { + Self { + prefix_len_encoder: DeltaBitPackEncoder::new_with_options(options), + suffix_writer: DeltaLengthByteArrayEncoder::new_with_options(options), + previous: vec![], + _phantom: PhantomData, + } + } } impl<T: DataType> Encoder<T> for DeltaByteArrayEncoder<T> { @@ -865,12 +999,146 @@ mod tests { use std::sync::Arc; use crate::encodings::decoding::{Decoder, DictDecoder, PlainDecoder, get_decoder}; + use crate::file::properties::WriterProperties; use crate::schema::types::{ColumnDescPtr, ColumnDescriptor, ColumnPath, Type as SchemaType}; use crate::util::bit_util; use crate::util::test_common::rand_gen::{RandGen, random_bytes}; const TEST_SET_SIZE: usize = 1024; + #[test] + fn test_delta_binary_packed_encoder_options() { + let options = DeltaBinaryPackedEncoderOptions::try_new(256, 4).unwrap(); + assert_eq!(options.block_size(), 256); + assert_eq!(options.mini_blocks_per_block(), 4); + + let encoder = DeltaBitPackEncoder::<Int32Type>::new_with_options(options); + assert_eq!(encoder.block_size, 256); + assert_eq!(encoder.mini_block_size, 64); + assert_eq!(encoder.num_mini_blocks, 4); + + let props = WriterProperties::builder() + .set_delta_binary_packed_encoder_options(options) + .build(); + let column_props = props.resolve_column_properties(&ColumnPath::new(vec![])); + let desc = create_test_col_desc_ptr(-1, Type::INT32); + let mut encoder = get_encoder_with_properties::<Int32Type>( + Encoding::DELTA_BINARY_PACKED, + &desc, + &column_props, + ) + .unwrap(); + let input: Vec<i32> = (0..600).map(|value| value * value).collect(); + encoder.put(&input).unwrap(); + let encoded = encoder.flush_buffer().unwrap(); + let mut header = bit_util::BitReader::new(encoded.clone()); + assert_eq!(header.get_vlq_int(), Some(256)); + assert_eq!(header.get_vlq_int(), Some(4)); + + let mut decoder = create_test_decoder::<Int32Type>(-1, Encoding::DELTA_BINARY_PACKED); + decoder.set_data(encoded, input.len()).unwrap(); + let mut output = vec![0; input.len()]; + assert_eq!(decoder.get(&mut output).unwrap(), input.len()); + assert_eq!(output, input); + + let encoder = DeltaLengthByteArrayEncoder::<ByteArrayType>::new_with_options(options); + assert_eq!(encoder.len_encoder.block_size, 256); + let encoder = DeltaByteArrayEncoder::<ByteArrayType>::new_with_options(options); + assert_eq!(encoder.prefix_len_encoder.block_size, 256); + assert_eq!(encoder.suffix_writer.len_encoder.block_size, 256); + + let input: Vec<ByteArray> = (0..600) + .map(|value| ByteArray::from(format!("prefix-{value:04}-suffix").into_bytes())) + .collect(); + let desc = create_test_col_desc_ptr(-1, Type::BYTE_ARRAY); + for encoding in [ + Encoding::DELTA_LENGTH_BYTE_ARRAY, + Encoding::DELTA_BYTE_ARRAY, + ] { + let mut encoder = + get_encoder_with_properties::<ByteArrayType>(encoding, &desc, &column_props) + .unwrap(); + encoder.put(&input).unwrap(); + let encoded = encoder.flush_buffer().unwrap(); + let mut header = bit_util::BitReader::new(encoded.clone()); + assert_eq!(header.get_vlq_int(), Some(256)); + assert_eq!(header.get_vlq_int(), Some(4)); + + let mut decoder = create_test_decoder::<ByteArrayType>(-1, encoding); + decoder.set_data(encoded, input.len()).unwrap(); + let mut output = vec![ByteArray::default(); input.len()]; + assert_eq!(decoder.get(&mut output).unwrap(), input.len()); + assert_eq!(output, input); + } + } + + #[test] + fn test_delta_binary_packed_encoder_options_validation() { + for (block_size, mini_blocks_per_block, message) in [ + (0, 4, "positive multiple of 128"), + (127, 4, "positive multiple of 128"), + (128, 0, "positive divisor"), + (128, 3, "positive divisor"), + (256, 16, "must be a multiple of 32"), + ] { + let error = DeltaBinaryPackedEncoderOptions::try_new(block_size, mini_blocks_per_block) + .unwrap_err(); + assert!(error.to_string().contains(message), "{error}"); + } + } + + #[test] + fn test_get_encoders_with_properties() { + fn check<T: DataType>( + encoding: Encoding, + values: &[T::T], + props: &ResolvedColumnProperties, + ) { + let desc = create_test_col_desc_ptr(-1, T::get_physical_type()); + let mut default = get_encoder::<T>(encoding, &desc).unwrap(); + let mut resolved = get_encoder_with_properties::<T>(encoding, &desc, props).unwrap(); + default.put(values).unwrap(); + resolved.put(values).unwrap(); + assert_eq!(resolved.encoding(), encoding); + assert_eq!( + resolved.flush_buffer().unwrap(), + default.flush_buffer().unwrap() + ); + } + + let path = ColumnPath::from("col"); + let props = WriterProperties::builder().build(); + let defaults = props.resolve_column_properties(&path); + let ints: Vec<i32> = (0..600).map(|value| value * value).collect(); + check::<Int32Type>(Encoding::DELTA_BINARY_PACKED, &ints, &defaults); + let longs: Vec<i64> = ints.iter().map(|&value| i64::from(value)).collect(); + check::<Int64Type>(Encoding::DELTA_BINARY_PACKED, &longs, &defaults); + let bytes: Vec<ByteArray> = ints + .iter() + .map(|value| ByteArray::from(format!("prefix-{value}").into_bytes())) + .collect(); + for encoding in [ + Encoding::DELTA_LENGTH_BYTE_ARRAY, + Encoding::DELTA_BYTE_ARRAY, + ] { + check::<ByteArrayType>(encoding, &bytes, &defaults); + } + + let props = WriterProperties::builder() + .set_delta_binary_packed_encoder_options( + DeltaBinaryPackedEncoderOptions::try_new(256, 4).unwrap(), + ) + .build(); + let custom = props.resolve_column_properties(&path); + for props in [&defaults, &custom] { + check::<Int32Type>(Encoding::PLAIN, &ints, props); + for encoding in [Encoding::PLAIN, Encoding::BYTE_STREAM_SPLIT, Encoding::ALP] { + check::<FloatType>(encoding, &[1.25, -2.5, 3.75], props); + check::<DoubleType>(encoding, &[1.25, -2.5, 3.75], props); + } + } + } + #[test] fn test_get_encoders() { // supported encodings @@ -1396,15 +1664,35 @@ mod tests { err: Option<ParquetError>, ) { let desc = create_test_col_desc_ptr(type_length, T::get_physical_type()); - let encoder = get_encoder::<T>(encoding, &desc); - match err { - Some(parquet_error) => { - assert_eq!( - encoder.err().unwrap().to_string(), - parquet_error.to_string() - ) + let defaults = WriterProperties::builder().build(); + let custom = WriterProperties::builder() + .set_delta_binary_packed_encoder_options( + DeltaBinaryPackedEncoderOptions::try_new(256, 4).unwrap(), + ) + .build(); + let path = ColumnPath::from("t"); + for encoder in [ + get_encoder::<T>(encoding, &desc), + get_encoder_with_properties::<T>( + encoding, + &desc, + &defaults.resolve_column_properties(&path), + ), + get_encoder_with_properties::<T>( + encoding, + &desc, + &custom.resolve_column_properties(&path), + ), + ] { + match &err { + Some(parquet_error) => { + assert_eq!( + encoder.err().unwrap().to_string(), + parquet_error.to_string() + ) + } + None => assert_eq!(encoder.unwrap().encoding(), encoding), } - None => assert_eq!(encoder.unwrap().encoding(), encoding), } } diff --git a/parquet/src/file/properties.rs b/parquet/src/file/properties.rs index e658c7edd0..435e542085 100644 --- a/parquet/src/file/properties.rs +++ b/parquet/src/file/properties.rs @@ -18,6 +18,7 @@ //! Configuration via [`WriterProperties`] and [`ReaderProperties`] use crate::basic::{Compression, Encoding}; use crate::compression::{CodecOptions, CodecOptionsBuilder}; +pub use crate::encodings::encoding::DeltaBinaryPackedEncoderOptions; #[cfg(feature = "encryption")] use crate::encryption::encrypt::FileEncryptionProperties; use crate::errors::{ParquetError, Result}; @@ -485,6 +486,20 @@ impl WriterProperties { ) } + /// Returns custom delta binary packed encoder options for a specific column. + /// + /// See [`DeltaBinaryPackedEncoderOptions`] for layout trade-offs. These options also apply to + /// the integer sub-encoders used by `DELTA_LENGTH_BYTE_ARRAY` and `DELTA_BYTE_ARRAY`. + pub fn delta_binary_packed_encoder_options( + &self, + col: &ColumnPath, + ) -> Option<DeltaBinaryPackedEncoderOptions> { + resolve_delta_binary_packed_encoder_options( + self.column_override(col), + &self.default_column_properties, + ) + } + /// Returns encoding for a data page, when dictionary encoding is enabled. /// /// This is not configurable. @@ -588,6 +603,9 @@ impl WriterProperties { data_page_v2_compression_ratio_threshold: resolve_data_page_v2_compression_ratio_threshold(column, default), bloom_filter_properties: resolve_bloom_filter_properties(column, default).cloned(), + delta_binary_packed_encoder_options: resolve_delta_binary_packed_encoder_options( + column, default, + ), } } @@ -1064,6 +1082,19 @@ impl WriterPropertiesBuilder { self } + /// Sets the default delta binary packed encoder block layout for all columns. + /// + /// See [`DeltaBinaryPackedEncoderOptions`] for layout trade-offs. These options also apply to + /// the integer sub-encoders used by `DELTA_LENGTH_BYTE_ARRAY` and `DELTA_BYTE_ARRAY`. + pub fn set_delta_binary_packed_encoder_options( + mut self, + value: DeltaBinaryPackedEncoderOptions, + ) -> Self { + self.default_column_properties + .set_delta_binary_packed_encoder_options(value); + self + } + /// Sets FileEncryptionProperties (defaults to `None`) #[cfg(feature = "encryption")] pub fn with_file_encryption_properties( @@ -1356,6 +1387,21 @@ impl WriterPropertiesBuilder { self } + /// Sets the delta binary packed encoder block layout for a specific column. + /// + /// Takes precedence over [`Self::set_delta_binary_packed_encoder_options`]. + /// See [`DeltaBinaryPackedEncoderOptions`] for layout trade-offs. These options also apply to + /// the integer sub-encoders used by `DELTA_LENGTH_BYTE_ARRAY` and `DELTA_BYTE_ARRAY`. + pub fn set_column_delta_binary_packed_encoder_options( + mut self, + col: ColumnPath, + value: DeltaBinaryPackedEncoderOptions, + ) -> Self { + self.get_mut_props(col) + .set_delta_binary_packed_encoder_options(value); + self + } + /// Deprecated alias for [`Self::set_column_bloom_filter_max_ndv`]. #[deprecated( since = "59.0.0", @@ -1667,6 +1713,7 @@ struct ColumnProperties { /// Whether the bloom filter NDV was explicitly set by the user bloom_filter_ndv_is_set: bool, data_page_v2_compression_ratio_threshold: Option<f64>, + delta_binary_packed_encoder_options: Option<DeltaBinaryPackedEncoderOptions>, } impl ColumnProperties { @@ -1774,6 +1821,10 @@ impl ColumnProperties { self.data_page_v2_compression_ratio_threshold = Some(value); } + fn set_delta_binary_packed_encoder_options(&mut self, value: DeltaBinaryPackedEncoderOptions) { + self.delta_binary_packed_encoder_options = Some(value); + } + /// Returns optional encoding for this column. fn encoding(&self) -> Option<Encoding> { self.encoding @@ -1825,6 +1876,10 @@ impl ColumnProperties { self.data_page_v2_compression_ratio_threshold } + fn delta_binary_packed_encoder_options(&self) -> Option<DeltaBinaryPackedEncoderOptions> { + self.delta_binary_packed_encoder_options + } + /// If bloom filter is enabled and NDV was not explicitly set, resolve it to the /// given `default_ndv` (typically derived from `max_row_group_row_count`). fn resolve_bloom_filter_ndv(&mut self, default_ndv: u64) { @@ -1860,6 +1915,8 @@ pub(crate) struct ResolvedColumnProperties { pub(crate) data_page_v2_compression_ratio_threshold: f64, /// See [`WriterProperties::bloom_filter_properties`]. pub(crate) bloom_filter_properties: Option<BloomFilterProperties>, + /// See [`WriterProperties::delta_binary_packed_encoder_options`]. + pub(crate) delta_binary_packed_encoder_options: Option<DeltaBinaryPackedEncoderOptions>, } /// Returns the setting read by `get` for `column` if it sets one, otherwise the @@ -1958,6 +2015,17 @@ fn resolve_bloom_filter_properties<'a>( .or_else(|| default.bloom_filter_properties()) } +fn resolve_delta_binary_packed_encoder_options( + column: Option<&ColumnProperties>, + default: &ColumnProperties, +) -> Option<DeltaBinaryPackedEncoderOptions> { + column_or_default( + column, + default, + ColumnProperties::delta_binary_packed_encoder_options, + ) +} + /// Reference counted reader properties. pub type ReaderPropertiesPtr = Arc<ReaderProperties>; @@ -2204,6 +2272,43 @@ mod tests { .bloom_filter_properties(&ColumnPath::from("col")) .is_none() ); + assert!( + props + .delta_binary_packed_encoder_options(&ColumnPath::from("col")) + .is_none() + ); + } + + #[test] + fn test_writer_properties_delta_binary_packed_encoder_options() { + let default = DeltaBinaryPackedEncoderOptions::try_new(256, 4).unwrap(); + let overridden = DeltaBinaryPackedEncoderOptions::try_new(128, 4).unwrap(); + let column = ColumnPath::from("column"); + let props = WriterProperties::builder() + .set_delta_binary_packed_encoder_options(default) + .set_column_delta_binary_packed_encoder_options(column.clone(), overridden) + .build(); + + assert_eq!( + props.delta_binary_packed_encoder_options(&column), + Some(overridden) + ); + assert_eq!( + props.delta_binary_packed_encoder_options(&ColumnPath::from("other")), + Some(default) + ); + assert_eq!( + props + .resolve_column_properties(&column) + .delta_binary_packed_encoder_options, + Some(overridden) + ); + + let rebuilt = props.into_builder().build(); + assert_eq!( + rebuilt.delta_binary_packed_encoder_options(&column), + Some(overridden) + ); } #[test]
