perf(parquet): slice up contiguous buffer for decimals and fsb (#10364)

MassivePizza · web-flow · commit 09ebd4eaeb8b · 2026-07-25T19:20:07.000-04:00
# Which issue does this PR close?  N/A # Rationale for this change  Converting decimals and FixedLenByteArrays to individual allocs requires a lot of alloc+drop time and memory overhead. # What changes are included in this PR?  Allocate one big buffer, and cut it up into slices of `bytes::Bytes`. This amortizes the alloc upfront, and effectively eliminates the drop overhead by turning a dealloc to an Arc decrement per slice. This still has moderate memory overhead. `Bytes` is 32 bytes, while the slices usually expose less than 16 bytes at a time e.g. UUIDs or decimal128, but a lot more effort would be required to implement a (perhaps borrowed?) alternative that behaves like `FixedLenByteArray`. There are a few other ideas worth exploring, primarily custom `ColumnValueEncoder`s that converts on the fly to avoid the temporary allocs all together. ~~There is also a small outlining+inlining in `compare_greater` to help out in the `get_min_max` loop.~~ Extracted to #10389. # Are these changes tested?  Should be covered by existing tests. There are also new benches included (extracted to #10388): <details> <summary>decimal bench</summary> ``` Benchmarking decimal/default: Warming up for 3.0000 s Warning: Unable to complete 100 samples in 5.0s. You may wish to increase target time to 5.4s, or reduce sample count to 90. decimal/default time: [53.676 ms 53.979 ms 54.306 ms] thrpt: [1.0790 GiB/s 1.0855 GiB/s 1.0916 GiB/s] change: time: [−44.611% −44.247% −43.885%] (p = 0.00 < 0.05) thrpt: [+78.204% +79.362% +80.542%] Performance has improved. Found 5 outliers among 100 measurements (5.00%) 4 (4.00%) high mild 1 (1.00%) high severe Benchmarking decimal/bloom_filter: Warming up for 3.0000 s Warning: Unable to complete 100 samples in 5.0s. You may wish to increase target time to 6.8s, or reduce sample count to 70. decimal/bloom_filter time: [66.597 ms 66.838 ms 67.095 ms] thrpt: [894.26 MiB/s 897.70 MiB/s 900.94 MiB/s] change: time: [−40.229% −39.907% −39.555%] (p = 0.00 < 0.05) thrpt: [+65.440% +66.408% +67.306%] Performance has improved. Found 10 outliers among 100 measurements (10.00%) 8 (8.00%) high mild 2 (2.00%) high severe Benchmarking decimal/parquet_2: Warming up for 3.0000 s Warning: Unable to complete 100 samples in 5.0s. You may wish to increase target time to 10.8s, or reduce sample count to 40. decimal/parquet_2 time: [105.82 ms 106.23 ms 106.66 ms] thrpt: [562.52 MiB/s 564.84 MiB/s 567.02 MiB/s] change: time: [−48.804% −48.523% −48.229%] (p = 0.00 < 0.05) thrpt: [+93.160% +94.263% +95.329%] Performance has improved. Found 3 outliers among 100 measurements (3.00%) 2 (2.00%) high mild 1 (1.00%) high severe Benchmarking decimal/zstd: Warming up for 3.0000 s Warning: Unable to complete 100 samples in 5.0s. You may wish to increase target time to 6.4s, or reduce sample count to 70. decimal/zstd time: [61.806 ms 62.013 ms 62.282 ms] thrpt: [963.36 MiB/s 967.55 MiB/s 970.78 MiB/s] change: time: [−60.257% −59.907% −59.556%] (p = 0.00 < 0.05) thrpt: [+147.25% +149.42% +151.62%] Performance has improved. Found 11 outliers among 100 measurements (11.00%) 1 (1.00%) low severe 2 (2.00%) low mild 5 (5.00%) high mild 3 (3.00%) high severe Benchmarking decimal/zstd_parquet_2: Warming up for 3.0000 s Warning: Unable to complete 100 samples in 5.0s. You may wish to increase target time to 11.9s, or reduce sample count to 40. decimal/zstd_parquet_2 time: [119.51 ms 119.87 ms 120.29 ms] thrpt: [498.82 MiB/s 500.54 MiB/s 502.07 MiB/s] change: time: [−55.716% −55.352% −54.992%] (p = 0.00 < 0.05) thrpt: [+122.18% +123.98% +125.82%] Performance has improved. Found 11 outliers among 100 measurements (11.00%) 2 (2.00%) low mild 2 (2.00%) high mild 7 (7.00%) high severe Benchmarking decimal/cdc: Warming up for 3.0000 s Warning: Unable to complete 100 samples in 5.0s. You may wish to increase target time to 7.8s, or reduce sample count to 60. decimal/cdc time: [77.810 ms 78.092 ms 78.389 ms] thrpt: [765.42 MiB/s 768.33 MiB/s 771.11 MiB/s] change: time: [−35.689% −35.360% −35.011%] (p = 0.00 < 0.05) thrpt: [+53.873% +54.703% +55.494%] Performance has improved. Found 10 outliers among 100 measurements (10.00%) 10 (10.00%) high mild ``` </details> <details> <summary>fsb bench</summary> ``` fsb/default time: [5.4152 ms 5.4276 ms 5.4407 ms] thrpt: [2.8943 GiB/s 2.9013 GiB/s 2.9079 GiB/s] change: time: [−27.824% −27.512% −27.215%] (p = 0.00 < 0.05) thrpt: [+37.391% +37.953% +38.550%] Performance has improved. Found 7 outliers among 100 measurements (7.00%) 1 (1.00%) low mild 5 (5.00%) high mild 1 (1.00%) high severe fsb/bloom_filter time: [5.8681 ms 5.8908 ms 5.9154 ms] thrpt: [2.6621 GiB/s 2.6732 GiB/s 2.6835 GiB/s] change: time: [−25.701% −25.298% −24.896%] (p = 0.00 < 0.05) thrpt: [+33.148% +33.866% +34.592%] Performance has improved. Found 6 outliers among 100 measurements (6.00%) 3 (3.00%) high mild 3 (3.00%) high severe fsb/parquet_2 time: [8.6936 ms 8.7692 ms 8.8503 ms] thrpt: [1.7793 GiB/s 1.7957 GiB/s 1.8113 GiB/s] change: time: [−31.621% −31.005% −30.286%] (p = 0.00 < 0.05) thrpt: [+43.444% +44.939% +46.243%] Performance has improved. Found 10 outliers among 100 measurements (10.00%) 9 (9.00%) high mild 1 (1.00%) high severe fsb/zstd time: [7.1760 ms 7.2043 ms 7.2360 ms] thrpt: [2.1762 GiB/s 2.1858 GiB/s 2.1944 GiB/s] change: time: [−21.480% −21.068% −20.631%] (p = 0.00 < 0.05) thrpt: [+25.994% +26.692% +27.356%] Performance has improved. Found 2 outliers among 100 measurements (2.00%) 1 (1.00%) high mild 1 (1.00%) high severe fsb/zstd_parquet_2 time: [9.2977 ms 9.4085 ms 9.5422 ms] thrpt: [1.6503 GiB/s 1.6737 GiB/s 1.6937 GiB/s] change: time: [−30.266% −29.333% −28.383%] (p = 0.00 < 0.05) thrpt: [+39.632% +41.508% +43.401%] Performance has improved. Found 7 outliers among 100 measurements (7.00%) 4 (4.00%) high mild 3 (3.00%) high severe fsb/cdc time: [7.9524 ms 7.9700 ms 7.9878 ms] thrpt: [1.9714 GiB/s 1.9758 GiB/s 1.9802 GiB/s] change: time: [−21.280% −20.793% −20.375%] (p = 0.00 < 0.05) thrpt: [+25.589% +26.251% +27.033%] Performance has improved. Found 2 outliers among 100 measurements (2.00%) 2 (2.00%) high mildaa ``` </details> # Are there any user-facing changes?  No.
diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
@@ -26,8 +26,8 @@ use std::sync::{Arc, Mutex};
 use std::vec::IntoIter;
 
 use arrow_array::cast::AsArray;
-use arrow_array::types::*;
 use arrow_array::{ArrayRef, Int32Array, RecordBatch, RecordBatchWriter};
+use arrow_array::{PrimitiveArray, types::*};
 use arrow_schema::{
     ArrowError, DataType as ArrowDataType, Field, IntervalUnit, SchemaRef, TimeUnit,
 };
@@ -1751,19 +1751,19 @@ fn write_leaf(
                 }
                 ArrowDataType::Decimal32(_, _) => {
                     let array = column.as_primitive::<Decimal32Type>();
-                    get_decimal_32_array_slice(array, indices.iter().copied())
+                    get_decimal_array_slice(array, indices.iter().copied())
                 }
                 ArrowDataType::Decimal64(_, _) => {
                     let array = column.as_primitive::<Decimal64Type>();
-                    get_decimal_64_array_slice(array, indices.iter().copied())
+                    get_decimal_array_slice(array, indices.iter().copied())
                 }
                 ArrowDataType::Decimal128(_, _) => {
                     let array = column.as_primitive::<Decimal128Type>();
-                    get_decimal_128_array_slice(array, indices.iter().copied())
+                    get_decimal_array_slice(array, indices.iter().copied())
                 }
                 ArrowDataType::Decimal256(_, _) => {
                     let array = column.as_primitive::<Decimal256Type>();
-                    get_decimal_256_array_slice(array, indices.iter().copied())
+                    get_decimal_array_slice(array, indices.iter().copied())
                 }
                 ArrowDataType::Float16 => {
                     let array = column.as_primitive::<Float16Type>();
@@ -1821,14 +1821,10 @@ fn get_interval_ym_array_slice(
     array: &arrow_array::IntervalYearMonthArray,
     indices: impl ExactSizeIterator<Item = usize>,
 ) -> Vec<FixedLenByteArray> {
-    let mut values = Vec::with_capacity(indices.len());
-    for i in indices {
-        let mut value = array.value(i).to_le_bytes().to_vec();
-        let mut suffix = vec![0; 8];
-        value.append(&mut suffix);
-        values.push(FixedLenByteArray::from(ByteArray::from(value)))
-    }
-    values
+    chunk_array_slice(12, indices, move |i, chunk| {
+        let value = array.value(i);
+        chunk[0..4].copy_from_slice(&value.to_le_bytes());
+    })
 }
 
 /// Returns 12-byte values representing 3 values of months, days and milliseconds (4-bytes each).
@@ -1837,93 +1833,111 @@ fn get_interval_dt_array_slice(
     array: &arrow_array::IntervalDayTimeArray,
     indices: impl ExactSizeIterator<Item = usize>,
 ) -> Vec<FixedLenByteArray> {
-    let mut values = Vec::with_capacity(indices.len());
-    for i in indices {
-        let mut out = [0; 12];
+    chunk_array_slice(12, indices, move |i, chunk| {
         let value = array.value(i);
-        out[4..8].copy_from_slice(&value.days.to_le_bytes());
-        out[8..12].copy_from_slice(&value.milliseconds.to_le_bytes());
-        values.push(FixedLenByteArray::from(ByteArray::from(out.to_vec())));
+        chunk[4..8].copy_from_slice(&value.days.to_le_bytes());
+        chunk[8..12].copy_from_slice(&value.milliseconds.to_le_bytes());
+    })
+}
+
+trait NativeDecimalType: DecimalType {
+    type NativeBytes: AsRef<[u8]>;
+
+    fn to_be_bytes(value: Self::Native) -> Self::NativeBytes;
+}
+impl NativeDecimalType for Decimal32Type {
+    type NativeBytes = [u8; Self::BYTE_LENGTH];
+
+    fn to_be_bytes(value: Self::Native) -> Self::NativeBytes {
+        value.to_be_bytes()
     }
-    values
 }
+impl NativeDecimalType for Decimal64Type {
+    type NativeBytes = [u8; Self::BYTE_LENGTH];
 
-fn get_decimal_32_array_slice(
-    array: &arrow_array::Decimal32Array,
-    indices: impl ExactSizeIterator<Item = usize>,
-) -> Vec<FixedLenByteArray> {
-    let mut values = Vec::with_capacity(indices.len());
-    let size = decimal_length_from_precision(array.precision());
-    for i in indices {
-        let as_be_bytes = array.value(i).to_be_bytes();
-        let resized_value = as_be_bytes[(4 - size)..].to_vec();
-        values.push(FixedLenByteArray::from(ByteArray::from(resized_value)));
+    fn to_be_bytes(value: Self::Native) -> Self::NativeBytes {
+        value.to_be_bytes()
     }
-    values
 }
+impl NativeDecimalType for Decimal128Type {
+    type NativeBytes = [u8; Self::BYTE_LENGTH];
 
-fn get_decimal_64_array_slice(
-    array: &arrow_array::Decimal64Array,
-    indices: impl ExactSizeIterator<Item = usize>,
-) -> Vec<FixedLenByteArray> {
-    let mut values = Vec::with_capacity(indices.len());
-    let size = decimal_length_from_precision(array.precision());
-    for i in indices {
-        let as_be_bytes = array.value(i).to_be_bytes();
-        let resized_value = as_be_bytes[(8 - size)..].to_vec();
-        values.push(FixedLenByteArray::from(ByteArray::from(resized_value)));
+    fn to_be_bytes(value: Self::Native) -> Self::NativeBytes {
+        value.to_be_bytes()
     }
-    values
 }
+impl NativeDecimalType for Decimal256Type {
+    type NativeBytes = [u8; Self::BYTE_LENGTH];
 
-fn get_decimal_128_array_slice(
-    array: &arrow_array::Decimal128Array,
-    indices: impl ExactSizeIterator<Item = usize>,
-) -> Vec<FixedLenByteArray> {
-    let mut values = Vec::with_capacity(indices.len());
-    let size = decimal_length_from_precision(array.precision());
-    for i in indices {
-        let as_be_bytes = array.value(i).to_be_bytes();
-        let resized_value = as_be_bytes[(16 - size)..].to_vec();
-        values.push(FixedLenByteArray::from(ByteArray::from(resized_value)));
+    fn to_be_bytes(value: Self::Native) -> Self::NativeBytes {
+        value.to_be_bytes()
     }
-    values
 }
 
-fn get_decimal_256_array_slice(
-    array: &arrow_array::Decimal256Array,
+fn get_decimal_array_slice<T: NativeDecimalType>(
+    array: &PrimitiveArray<T>,
     indices: impl ExactSizeIterator<Item = usize>,
 ) -> Vec<FixedLenByteArray> {
-    let mut values = Vec::with_capacity(indices.len());
-    let size = decimal_length_from_precision(array.precision());
-    for i in indices {
-        let as_be_bytes = array.value(i).to_be_bytes();
-        let resized_value = as_be_bytes[(32 - size)..].to_vec();
-        values.push(FixedLenByteArray::from(ByteArray::from(resized_value)));
+    let chunk_size = decimal_length_from_precision(array.precision());
+    assert!(chunk_size <= T::BYTE_LENGTH);
+
+    if chunk_size == T::BYTE_LENGTH {
+        // Special-case that allows inlining memcpy.
+        chunk_array_slice(chunk_size, indices, move |i, chunk| {
+            let as_be_bytes = T::to_be_bytes(array.value(i));
+            chunk.copy_from_slice(as_be_bytes.as_ref());
+        })
+    } else {
+        chunk_array_slice(chunk_size, indices, move |i, chunk| {
+            let as_be_bytes = T::to_be_bytes(array.value(i));
+            let resized_value = &as_be_bytes.as_ref()[(T::BYTE_LENGTH - chunk.len())..];
+            chunk.copy_from_slice(resized_value);
+        })
     }
-    values
 }
 
 fn get_float_16_array_slice(
     array: &arrow_array::Float16Array,
     indices: impl ExactSizeIterator<Item = usize>,
 ) -> Vec<FixedLenByteArray> {
-    let mut values = Vec::with_capacity(indices.len());
-    for i in indices {
-        let value = array.value(i).to_le_bytes().to_vec();
-        values.push(FixedLenByteArray::from(ByteArray::from(value)));
-    }
-    values
+    chunk_array_slice(2, indices, move |i, chunk| {
+        let value = array.value(i).to_le_bytes();
+        chunk.copy_from_slice(&value);
+    })
 }
 
 fn get_fsb_array_slice(
     array: &arrow_array::FixedSizeBinaryArray,
     indices: impl ExactSizeIterator<Item = usize>,
 ) -> Vec<FixedLenByteArray> {
-    let mut values = Vec::with_capacity(indices.len());
-    for i in indices {
-        let value = array.value(i).to_vec();
-        values.push(FixedLenByteArray::from(ByteArray::from(value)))
+    chunk_array_slice(array.value_size(), indices, move |i, chunk| {
+        let value = array.value(i);
+        chunk.copy_from_slice(value);
+    })
+}
+
+#[inline]
+fn chunk_array_slice(
+    chunk_size: usize,
+    indices: impl ExactSizeIterator<Item = usize>,
+    writer: impl Fn(usize, &mut [u8]),
+) -> Vec<FixedLenByteArray> {
+    let capacity = indices.len() * chunk_size;
+    // TODO: This could be done with Vec::spare_capacity_mut,
+    //       but [MaybeUninit]::write_copy_of_slice is gated behind MSRV 1.93
+    let mut arena = vec![0; capacity];
+    for (i, chunk) in indices.zip(arena.chunks_exact_mut(chunk_size)) {
+        writer(i, chunk);
+    }
+    chunk_contiguous_vec(arena, chunk_size)
+}
+
+fn chunk_contiguous_vec(arena: Vec<u8>, chunk_size: usize) -> Vec<FixedLenByteArray> {
+    let mut values = Vec::with_capacity(arena.len() / chunk_size);
+    let mut arena = Bytes::from(arena);
+    while arena.len() >= chunk_size {
+        let slice = arena.split_to(chunk_size);
+        values.push(FixedLenByteArray::from(ByteArray::from(slice)));
     }
     values
 }
