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 3c5bc5f4ce Allow configuring DELTA_BINARY_PACKED block layout (#11051)
3c5bc5f4ce is described below
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]
