Support wide decimals in DecimalBytePartsArray and kernels (#9809)

`DecimalBytePartsArray` previously stored the entire unscaled decimal
value in one signed integer child, limiting it to values that fit in 64
bits. It now supports wide decimals by representing each value as
integer parts that can be compressed independently, while preserving the
decimal's logical precision, scale, and nullability.

The array has a signed most significant part (MSP) and up to three
unsigned 64-bit lower parts, ordered most significant first. Splitting
canonical decimal storage produces:

| Decimal storage | Children |
| --- | --- |
| `i8` / `i16` / `i32` / `i64` | Signed MSP only; shares the original
value buffer |
| `i128` | `i64` MSP + one `u64` lower part |
| `i256` | `i64` MSP + three `u64` lower parts |

Only the MSP carries validity. Every lower part must be a non-nullable
`u64` array with the same length, and splitting wide decimals zeroes the
parts at null positions. All children remain `ArrayRef`s, so their
individual encodings are independent of the decimal representation.

`execute::<DecimalArray>` reassembles a `DecimalArray` from the MSP and
lower parts children.

`take` with nullable indices is not yet supported by the
DecimalByteParts kernel for arrays with lower parts; it falls back to
canonical execution. Taking each part directly would make the lower
parts nullable, violating the representation's invariant. The frozen
serializer also continues to reject arrays with lower parts.

This PR also refines the splitting and assembly modules.
* Assembly takes `ArrayRef`s instead of `PrimitiveArray`s so that in the
future, we can add special fast paths for constant arrays
* Assembly casts lower parts to `u64`s, allowing for assembly of
narrowed lower parts.
* Assembly loop is optimized such that it vectorizes for `i256` assembly
on local runs.

---------

Signed-off-by: Matt Katz <mhkatz97@gmail.com>

Author: mhk197

.gitignore

@@ -52,6 +52,8 @@ coverage.xml
 *.cover
 *.py,cover
 .hypothesis/
+# hegeltest's example database, the Rust equivalent of .hypothesis/
+.hegel/
 .pytest_cache/
 cover/
 

Cargo.lock

@@ -176,6 +176,7 @@ glob = "0.3.2"
 goldenfile = "1"
 half = { version = "2.7.1", features = ["std", "num-traits"] }
 hashbrown = "0.17.1"
+hegeltest = "0.28.7"
 http = "1.5.0"
 humansize = "2.1.3"
 indicatif = "0.18.0"

Cargo.toml

@@ -27,9 +27,11 @@ vortex-session = { workspace = true }
 
 [dev-dependencies]
 divan = { workspace = true }
+hegeltest = { workspace = true }
 rand = { workspace = true }
 rstest = { workspace = true }
 vortex-array = { path = "../../vortex-array", features = ["_test-harness"] }
+vortex-bench-support = { workspace = true }
 
 [[bench]]
 name = "dbp_assemble"

encodings/decimal-byte-parts/Cargo.toml

@@ -1,51 +1,77 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-//! Shared decimal inputs for splitting and assembly benchmarks.
+//! Shared inputs for splitting and assembly benchmarks.
 
 use rand::RngExt;
 use rand::SeedableRng;
 use rand::rngs::StdRng;
-use vortex_array::arrays::DecimalArray;
-use vortex_array::dtype::DecimalDType;
 use vortex_array::dtype::DecimalType;
 use vortex_array::dtype::i256;
-use vortex_array::validity::Validity;
 use vortex_buffer::Buffer;
-use vortex_error::vortex_panic;
 
 pub(super) fn cases() -> Vec<(DecimalType, usize)> {
-    [DecimalType::I64, DecimalType::I128, DecimalType::I256]
+    [DecimalType::I128, DecimalType::I256]
         .into_iter()
         .flat_map(|values_type| [1_024, 8_192].map(|len| (values_type, len)))
         .collect()
 }
 
-pub(super) fn decimal_array(
-    values_type: DecimalType,
-    len: usize,
-    validity: Validity,
-) -> DecimalArray {
+pub(super) fn i128_values(len: usize) -> Buffer<i128> {
     let mut rng = StdRng::seed_from_u64(42);
+    let max = 10i128.pow(38) - 1;
+    (0..len).map(|_| rng.random_range(-max..=max)).collect()
+}
+
+pub(super) fn i256_values(len: usize) -> Buffer<i256> {
+    let mut rng = StdRng::seed_from_u64(42);
+    // Keep the magnitude below 10^76 while exercising all four signed/unsigned words.
+    (0..len)
+        .map(|_| i256::from_parts(rng.random(), rng.random::<i128>() >> 4))
+        .collect()
+}
+
+#[cfg(not(codspeed))]
+pub(super) mod arrays {
+    use rand::RngExt;
+    use rand::SeedableRng;
+    use rand::rngs::StdRng;
+    use vortex_array::arrays::DecimalArray;
+    use vortex_array::dtype::DecimalDType;
+    use vortex_array::dtype::DecimalType;
+    use vortex_array::validity::Validity;
+    use vortex_buffer::Buffer;
+    use vortex_error::vortex_panic;
+
+    use super::i128_values;
+    use super::i256_values;
 
-    macro_rules! decimal {
-        ($T:ty, $precision:literal) => {{
-            let max = <$T>::pow(10, $precision) - 1;
-            let values: Buffer<$T> = (0..len).map(|_| rng.random_range(-max..=max)).collect();
-            DecimalArray::new(values, DecimalDType::new($precision, 2), validity)
-        }};
+    pub(crate) fn cases() -> Vec<(DecimalType, usize)> {
+        [DecimalType::I64, DecimalType::I128, DecimalType::I256]
+            .into_iter()
+            .flat_map(|values_type| [1_024, 8_192].map(|len| (values_type, len)))
+            .collect()
     }
 
-    match values_type {
-        DecimalType::I64 => decimal!(i64, 18),
-        DecimalType::I128 => decimal!(i128, 38),
-        DecimalType::I256 => {
-            // Keep the magnitude below 10^76 while exercising all four signed/unsigned words.
-            let values: Buffer<i256> = (0..len)
-                .map(|_| i256::from_parts(rng.random(), rng.random::<i128>() >> 4))
-                .collect();
-            DecimalArray::new(values, DecimalDType::new(76, 2), validity)
+    pub(crate) fn decimal_array(
+        values_type: DecimalType,
+        len: usize,
+        validity: Validity,
+    ) -> DecimalArray {
+        match values_type {
+            DecimalType::I64 => {
+                let mut rng = StdRng::seed_from_u64(42);
+                let max = 10i64.pow(18) - 1;
+                let values: Buffer<i64> = (0..len).map(|_| rng.random_range(-max..=max)).collect();
+                DecimalArray::new(values, DecimalDType::new(18, 2), validity)
+            }
+            DecimalType::I128 => {
+                DecimalArray::new(i128_values(len), DecimalDType::new(38, 2), validity)
+            }
+            DecimalType::I256 => {
+                DecimalArray::new(i256_values(len), DecimalDType::new(76, 2), validity)
+            }
+            _ => vortex_panic!("unsupported benchmark storage type: {values_type}"),
         }
-        _ => vortex_panic!("unsupported benchmark storage type: {values_type}"),
     }
 }