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-11910-81392e3e9925e099b28a0895bf668ef868f3e5b0 in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
commit 6939e904071fb162f656e38b79ab3409e81805da Author: Y Ethan Guo <[email protected]> AuthorDate: Wed Oct 28 17:44:40 2026 +0000 perf(parquet): convert FIXED_LEN_BYTE_ARRAY decimals with a width-specialized loop (#11910) # Which issue does this PR close? - Closes #11908 # Rationale for this change Converting `FIXED_LEN_BYTE_ARRAY` decimals to native integers dominates reading them, because `sign_extend_be` copies a runtime length and compiles to a `memcpy` call per value. # What changes are included in this PR? `consume_batch` matches the column's byte width once per batch and converts with a loop specialized to it (`as_chunks::<W>()` plus a new `sign_extend_be_fixed`), so each copy is a constant-size move. Every width each decimal type accepts is covered, which is 60 small monomorphized loops, and the existing path stays as a fallback. # Are these changes tested? `test_read_decimal_of_every_width` round-trips the minimal width for every precision from 1 to 76 at a fractional scale, with positive and negative values, boundaries and nulls, as each decimal type that can hold it. `arrow_reader` benchmarks (including those from #11909) against `main`, Apple M3 Max, mean of two alternating runs. Each iteration reads 80,000 values (2 row groups of 4 pages of 10,000 values): | Benchmark (mandatory, no NULLs unless noted) | main | this PR | change | | --- | --- | --- | --- | | FLBA(16) as Decimal128, plain | 273 µs | 65 µs | -76% | | FLBA(16) as Decimal128, plain, half NULLs | 410 µs | 211 µs | -49% | | FLBA(16) as Decimal128, byte_stream_split | 629 µs | 416 µs | -34% | | FLBA(7) as Decimal128, plain | 638 µs | 67 µs | -89% | | FLBA(7) as Decimal128, plain, half NULLs | 765 µs | 260 µs | -66% | | FLBA(7) as Decimal64, plain | 341 µs | 55 µs | -84% | | BYTE_ARRAY as Decimal128, plain (unchanged path) | 826 µs | 865 µs | noise | The `BYTE_ARRAY` control moved +5%, but its loop compiles to the same instructions in both builds, only placed at a different address, so that is code layout noise. End to end, this speeds up DuckDB reading Hudi tables through hudi-rs, which reads the tables' Parquet files with this reader. Hudi's Spark writer stores every decimal in them as `FIXED_LEN_BYTE_ARRAY`. Locally (same M3 Max, Snappy Parquet files on local disk, 4 threads, so the queries are CPU-bound), TPC-H gets faster in proportion to how many decimal values a query decodes: | Query | SF1 | SF10 | | --- | --- | --- | | Q6 | -35% | -37% | | Q15 | -28% | -25% | | Q14 | -24% | -24% | | Q17 | -22% | -20% | | Q1 | -17% | -18% | | Q19 | -15% | -18% | | All 22 queries | 2.37 s to 2.08 s (-12%) | 17.7 s to 15.7 s (-11%) | Queries that decode few decimals (Q4, Q12, Q13, Q16) are unchanged. DataFusion 55.2 shows the same on plain Parquet files, with this change backported to parquet 59.3.2. Spark wrote them with `spark.sql.parquet.writeLegacyFormat=true`, which also stores every decimal as `FIXED_LEN_BYTE_ARRAY` (7 bytes here). Same machine, default release build, 4 partitions at SF1 and 8 at SF10, median of 5 rounds with the builds alternating, results checked against an `INT64` copy of the data: | Query | SF1 | SF10 | | --- | --- | --- | | Q6 | -30% | -31% | | Q15 | -30% | -26% | | Q14 | -26% | -25% | | Q17 | -15% | -13% | | Q1 | -18% | -19% | | Q19 | -23% | -23% | | All 22 queries | 1.82 s to 1.53 s (-16%) | 9.27 s to 7.97 s (-14%) | Building with `-C target-cpu=native` moves neither build by more than about 1%, since NEON is already in the default `aarch64` target. # Are there any significant user-facing changes? No. AI usage: written with an AI coding assistant under my direction; I reviewed every line and ran the tests and benchmarks. --- .../src/arrow/array_reader/fixed_len_byte_array.rs | 153 ++++++++++++++++++--- parquet/src/arrow/buffer/bit_util.rs | 16 +++ 2 files changed, 151 insertions(+), 18 deletions(-) diff --git a/parquet/src/arrow/array_reader/fixed_len_byte_array.rs b/parquet/src/arrow/array_reader/fixed_len_byte_array.rs index c2efb1c5d6..e81e4dec70 100644 --- a/parquet/src/arrow/array_reader/fixed_len_byte_array.rs +++ b/parquet/src/arrow/array_reader/fixed_len_byte_array.rs @@ -16,7 +16,7 @@ // under the License. use crate::arrow::array_reader::{ArrayReader, read_records, skip_records}; -use crate::arrow::buffer::bit_util::{iter_set_bits_rev, sign_extend_be}; +use crate::arrow::buffer::bit_util::{iter_set_bits_rev, sign_extend_be, sign_extend_be_fixed}; use crate::arrow::buffer::offset_buffer::OffsetBuffer; use crate::arrow::decoder::{DeltaByteArrayDecoder, DictIndexDecoder}; use crate::arrow::record_reader::GenericRecordReader; @@ -27,10 +27,10 @@ use crate::column::page::PageIterator; use crate::column::reader::decoder::{ColumnValueDecoder, normalize_fixed_len_byte_array_payload}; use crate::errors::{ParquetError, Result}; use crate::schema::types::ColumnDescPtr; +use arrow_array::types::{Decimal32Type, Decimal64Type, Decimal128Type, Decimal256Type}; use arrow_array::{ - ArrayRef, Decimal32Array, Decimal64Array, Decimal128Array, Decimal256Array, - FixedSizeBinaryArray, Float16Array, IntervalDayTimeArray, IntervalYearMonthArray, - OffsetSizeTrait, + Array, ArrayRef, ArrowPrimitiveType, FixedSizeBinaryArray, Float16Array, IntervalDayTimeArray, + IntervalYearMonthArray, OffsetSizeTrait, PrimitiveArray, }; use arrow_buffer::{Buffer, IntervalDayTime, i256}; use arrow_data::ArrayDataBuilder; @@ -144,6 +144,36 @@ struct FixedLenByteArrayReader { record_reader: GenericRecordReader<FixedLenByteArrayBuffer, ValueDecoder>, } +/// Converts the big-endian two's complement values of `binary`, each `W` bytes wide, +/// with `from_be`. +fn decimal_from_be<T: ArrowPrimitiveType, const W: usize, const N: usize>( + binary: &FixedSizeBinaryArray, + from_be: impl Fn([u8; N]) -> T::Native, +) -> PrimitiveArray<T> { + let (values, _) = binary.value_data()[..binary.len() * W].as_chunks::<W>(); + let values: Vec<T::Native> = values + .iter() + .map(|b| from_be(sign_extend_be_fixed(b))) + .collect(); + PrimitiveArray::new(values.into(), binary.nulls().cloned()) +} + +/// Converts the values of a [`FixedSizeBinaryArray`] to the decimal type `$t`, with a +/// loop specialized to the value width when it is at most `$size`, the size of `$native`. +macro_rules! decimal_array { + ($binary:expr, $t:ty, $native:ty, $size:literal) => {{ + let binary = &$binary; + seq_macro::seq!(W in 1..=$size { + match binary.value_length() { + #(W => decimal_from_be::<$t, W, $size>(binary, <$native>::from_be_bytes),)* + _ => PrimitiveArray::<$t>::from_unary(binary, |b: &[u8]| { + <$native>::from_be_bytes(sign_extend_be(b)) + }), + } + }) + }}; +} + impl FixedLenByteArrayReader { fn new( pages: Box<dyn PageIterator>, @@ -208,24 +238,20 @@ impl ArrayReader for FixedLenByteArrayReader { // the inner loop (see docs for `PrimitiveArray::from_unary`). let array: ArrayRef = match &self.data_type { ArrowType::Decimal32(p, s) => { - let f = |b: &[u8]| i32::from_be_bytes(sign_extend_be(b)); - Arc::new(Decimal32Array::from_unary(&binary, f).with_precision_and_scale(*p, *s)?) - as ArrayRef + let array = decimal_array!(binary, Decimal32Type, i32, 4); + Arc::new(array.with_precision_and_scale(*p, *s)?) as ArrayRef } ArrowType::Decimal64(p, s) => { - let f = |b: &[u8]| i64::from_be_bytes(sign_extend_be(b)); - Arc::new(Decimal64Array::from_unary(&binary, f).with_precision_and_scale(*p, *s)?) - as ArrayRef + let array = decimal_array!(binary, Decimal64Type, i64, 8); + Arc::new(array.with_precision_and_scale(*p, *s)?) as ArrayRef } ArrowType::Decimal128(p, s) => { - let f = |b: &[u8]| i128::from_be_bytes(sign_extend_be(b)); - Arc::new(Decimal128Array::from_unary(&binary, f).with_precision_and_scale(*p, *s)?) - as ArrayRef + let array = decimal_array!(binary, Decimal128Type, i128, 16); + Arc::new(array.with_precision_and_scale(*p, *s)?) as ArrayRef } ArrowType::Decimal256(p, s) => { - let f = |b: &[u8]| i256::from_be_bytes(sign_extend_be(b)); - Arc::new(Decimal256Array::from_unary(&binary, f).with_precision_and_scale(*p, *s)?) - as ArrayRef + let array = decimal_array!(binary, Decimal256Type, i256, 32); + Arc::new(array.with_precision_and_scale(*p, *s)?) as ArrayRef } ArrowType::Interval(unit) => { // An interval is stored as 3x 32-bit unsigned integers storing months, days, @@ -608,8 +634,12 @@ enum Decoder { mod tests { use super::*; use crate::arrow::ArrowWriter; - use crate::arrow::arrow_reader::ParquetRecordBatchReader; - use arrow::datatypes::Field; + use crate::arrow::arrow_reader::{ + ArrowReaderOptions, ParquetRecordBatchReader, ParquetRecordBatchReaderBuilder, + }; + use crate::data_type::{FixedLenByteArray, FixedLenByteArrayType}; + use crate::file::writer::SerializedFileWriter; + use arrow::datatypes::{Field, Schema}; use arrow::error::Result as ArrowResult; use arrow_array::{Array, ListArray}; use arrow_array::{Decimal256Array, RecordBatch}; @@ -664,4 +694,91 @@ mod tests { assert_eq!(&written.slice(3, 3), &read[1]); assert_eq!(&written.slice(6, 1), &read[2]); } + + /// Writes `values` as an optional `FIXED_LEN_BYTE_ARRAY` column of `DECIMAL(precision, scale)` + /// with the minimal width for `precision`. + fn write_decimal_column(precision: u8, scale: i8, values: &[Option<i256>]) -> Bytes { + let width = crate::arrow::schema::decimal_length_from_precision(precision); + let schema = crate::schema::parser::parse_message_type(&format!( + "message test {{\ + OPTIONAL FIXED_LEN_BYTE_ARRAY ({width}) d (DECIMAL({precision}, {scale})); }}" + )) + .unwrap(); + let mut buffer = Vec::new(); + let mut writer = + SerializedFileWriter::new(&mut buffer, Arc::new(schema), Default::default()).unwrap(); + let mut row_group = writer.next_row_group().unwrap(); + let mut column = row_group.next_column().unwrap().unwrap(); + let def_levels: Vec<i16> = values.iter().map(|v| v.is_some() as i16).collect(); + let data: Vec<FixedLenByteArray> = values + .iter() + .flatten() + .map(|v| FixedLenByteArray::from(v.to_be_bytes()[32 - width..].to_vec())) + .collect(); + column + .typed::<FixedLenByteArrayType>() + .write_batch(&data, Some(&def_levels), None) + .unwrap(); + column.close().unwrap(); + row_group.close().unwrap(); + writer.close().unwrap(); + Bytes::from(buffer) + } + + #[test] + fn test_read_decimal_of_every_width() { + let ten = i256::from_i128(10); + for precision in 1..=76u8 { + let scale = precision.div_ceil(2) as i8; + let max = (0..precision) + .fold(i256::ONE, |acc, _| acc.wrapping_mul(ten)) + .wrapping_sub(i256::ONE); + let values = [ + Some(i256::ZERO), + Some(i256::ONE), + Some(i256::MINUS_ONE), + None, + Some(max), + Some(max.wrapping_neg()), + None, + Some(i256::from_i128(-7)), + ]; + let data_types = [ + (9, ArrowType::Decimal32(precision, scale)), + (18, ArrowType::Decimal64(precision, scale)), + (38, ArrowType::Decimal128(precision, scale)), + (76, ArrowType::Decimal256(precision, scale)), + ]; + let file = write_decimal_column(precision, scale, &values); + for (_, data_type) in data_types + .iter() + .filter(|(max_precision, _)| precision <= *max_precision) + { + let schema = Arc::new(Schema::new(vec![Field::new("d", data_type.clone(), true)])); + let options = ArrowReaderOptions::new().with_schema(schema); + let batches = + ParquetRecordBatchReaderBuilder::try_new_with_options(file.clone(), options) + .unwrap() + .with_batch_size(4) + .build() + .unwrap() + .collect::<ArrowResult<Vec<_>>>() + .unwrap(); + let columns: Vec<&dyn Array> = + batches.iter().map(|b| b.column(0).as_ref()).collect(); + let read = arrow::compute::concat(&columns).unwrap(); + assert_eq!(read.data_type(), data_type); + let read = + arrow::compute::cast(&read, &ArrowType::Decimal256(precision, scale)).unwrap(); + let expected = Decimal256Array::from_iter(values) + .with_precision_and_scale(precision, scale) + .unwrap(); + assert_eq!( + read.as_ref(), + &expected as &dyn Array, + "precision {precision}, {data_type}" + ); + } + } + } } diff --git a/parquet/src/arrow/buffer/bit_util.rs b/parquet/src/arrow/buffer/bit_util.rs index 3a288bf7d1..f81cad02d6 100644 --- a/parquet/src/arrow/buffer/bit_util.rs +++ b/parquet/src/arrow/buffer/bit_util.rs @@ -62,6 +62,22 @@ pub fn sign_extend_be<const N: usize>(b: &[u8]) -> [u8; N] { result } +/// Sign extends `b`, a big-endian two's complement value of `W` bytes, to `N` bytes. +/// +/// Equivalent to [`sign_extend_be`] for a width known at compile time, which keeps the +/// copy a constant-size move. +#[inline] +pub fn sign_extend_be_fixed<const W: usize, const N: usize>(b: &[u8; W]) -> [u8; N] { + const { assert!(W > 0 && W <= N, "invalid width for sign extension") }; + let mut result = if b[0] & 0x80 != 0 { + [u8::MAX; N] + } else { + [0; N] + }; + result[N - W..].copy_from_slice(b); + result +} + #[cfg(test)] mod tests { use super::*;
