Compress wide decimals when the writer permits the v2 format

.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.toml

@@ -177,6 +177,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"

encodings/decimal-byte-parts/Cargo.toml

@@ -27,6 +27,7 @@ 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"] }

encodings/decimal-byte-parts/src/decimal_byte_parts/compute/cast.rs

@@ -10,6 +10,7 @@ use vortex_array::scalar_fn::fns::cast::CastReduce;
 use vortex_error::VortexResult;
 
 use crate::DecimalByteParts;
+use crate::decimal_byte_parts::DecimalBytePartsArrayExt;
 use crate::decimal_byte_parts::DecimalBytePartsArraySlotsExt;
 
 impl CastReduce for DecimalByteParts {
@@ -19,7 +20,7 @@ impl CastReduce for DecimalByteParts {
             return Ok(None);
         }
         // DecimalBytePartsArray can only have Decimal dtype, so we only handle decimal-to-decimal casts
-        let DType::Decimal(target_decimal, target_nullability) = dtype else {
+        let DType::Decimal(_, target_nullability) = dtype else {
             // Cannot cast decimal to non-decimal types - delegate to canonical form
             return Ok(None);
         };
@@ -29,9 +30,7 @@ impl CastReduce for DecimalByteParts {
             .msp()
             .cast(array.msp().dtype().with_nullability(*target_nullability))?;
 
-        Ok(Some(
-            DecimalByteParts::try_new(new_msp, *target_decimal)?.into_array(),
-        ))
+        array.with_msp(new_msp).map(|a| Some(a.into_array()))
     }
 }
 
@@ -49,10 +48,14 @@ mod tests {
     use vortex_array::dtype::DType;
     use vortex_array::dtype::DecimalDType;
     use vortex_array::dtype::Nullability;
+    use vortex_array::validity::Validity;
     use vortex_buffer::buffer;
 
     use crate::DecimalByteParts;
     use crate::DecimalBytePartsArray;
+    use crate::decimal_byte_parts::testing::i128_parts;
+    use crate::decimal_byte_parts::testing::i256_of;
+    use crate::decimal_byte_parts::testing::i256_parts;
 
     #[test]
     fn test_cast_decimal_byte_parts_nullability() {
@@ -117,6 +120,14 @@ mod tests {
         buffer![-100i32, -200, 300, -400, 500].into_array(),
         DecimalDType::new(10, 2),
     ).unwrap())]
+    #[case::one_lower_part(i128_parts(
+        vec![1i128 << 70, -(1i128 << 70), 5, (1i128 << 64) - 1, 0],
+        Validity::NonNullable,
+    ))]
+    #[case::three_lower_parts(i256_parts(
+        vec![i256_of(1, 0), i256_of(-1, 5), i256_of(0, u128::MAX)],
+        Validity::NonNullable,
+    ))]
     fn test_cast_decimal_byte_parts_conformance(#[case] array: DecimalBytePartsArray) {
         test_cast_conformance(
             &array.into_array(),

encodings/decimal-byte-parts/src/decimal_byte_parts/compute/compare.rs

@@ -39,6 +39,12 @@ impl CompareKernel for DecimalByteParts {
             return Ok(None);
         };
 
+        // The MSP alone only determines the ordering when it holds the whole value. With
+        // lower parts present, fall back to comparing the canonical decimal.
+        if !lhs.lower_parts().is_empty() {
+            return Ok(None);
+        }
+
         let nullability = lhs.dtype().nullability() | rhs.dtype().nullability();
         let scalar_type = lhs.msp().dtype().with_nullability(nullability);
 
@@ -158,10 +164,12 @@ mod tests {
     use vortex_array::scalar_fn::fns::operators::Operator;
     use vortex_array::validity::Validity;
     use vortex_buffer::buffer;
+    use vortex_error::VortexExpect;
     use vortex_error::VortexResult;
     use vortex_session::VortexSession;
 
     use crate::DecimalByteParts;
+    use crate::decimal_byte_parts::testing::i128_parts;
 
     static SESSION: LazyLock<VortexSession> = LazyLock::new(|| {
         let session = vortex_array::array_session();
@@ -220,6 +228,45 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn compare_decimal_const_with_lower_parts() -> VortexResult<()> {
+        // The MSP-only pushdown is invalid once lower parts carry part of the value, so this
+        // must fall back to the canonical comparison rather than compare MSPs.
+        let values = vec![1i128 << 70, (1i128 << 70) + 1, 5, -(1i128 << 70)];
+        let lhs = i128_parts(values.clone(), Validity::NonNullable).into_array();
+        let decimal_dtype = *lhs
+            .dtype()
+            .as_decimal_opt()
+            .vortex_expect("decimal byte parts array");
+
+        let pivot = (1i128 << 70) + 1;
+        let rhs = ConstantArray::new(
+            Scalar::decimal(
+                DecimalValue::I128(pivot),
+                decimal_dtype,
+                Nullability::NonNullable,
+            ),
+            lhs.len(),
+        )
+        .into_array();
+
+        let mut ctx = SESSION.create_execution_ctx();
+        for (operator, predicate) in [
+            (Operator::Eq, (|v, p| v == p) as fn(i128, i128) -> bool),
+            (Operator::NotEq, |v, p| v != p),
+            (Operator::Lt, |v, p| v < p),
+            (Operator::Lte, |v, p| v <= p),
+            (Operator::Gt, |v, p| v > p),
+            (Operator::Gte, |v, p| v >= p),
+        ] {
+            let res = lhs.clone().binary(rhs.clone(), operator)?;
+            let expected =
+                BoolArray::from_iter(values.iter().map(|v| predicate(*v, pivot))).into_array();
+            assert_arrays_eq!(res, expected, &mut ctx);
+        }
+        Ok(())
+    }
+
     #[test]
     fn compare_decimal_const_unconvertible_comparison() {
         let decimal_dtype = DecimalDType::new(40, 2);

encodings/decimal-byte-parts/src/decimal_byte_parts/compute/filter.rs

@@ -5,22 +5,17 @@ use vortex_array::ArrayRef;
 use vortex_array::ArrayView;
 use vortex_array::IntoArray;
 use vortex_array::arrays::filter::FilterReduce;
-use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_mask::Mask;
 
 use crate::DecimalByteParts;
-use crate::decimal_byte_parts::DecimalBytePartsArraySlotsExt;
+use crate::decimal_byte_parts::DecimalBytePartsArrayExt;
+
 impl FilterReduce for DecimalByteParts {
     fn filter(array: ArrayView<'_, Self>, mask: &Mask) -> VortexResult<Option<ArrayRef>> {
-        DecimalByteParts::try_new(
-            array.msp().filter(mask.clone())?,
-            *array
-                .dtype()
-                .as_decimal_opt()
-                .vortex_expect("must be a decimal dtype"),
-        )
-        .map(|d| Some(d.into_array()))
+        array
+            .map_parts(|part| part.filter(mask.clone()))
+            .map(|d| Some(d.into_array()))
     }
 }
 
@@ -32,9 +27,13 @@ mod test {
     use vortex_array::arrays::PrimitiveArray;
     use vortex_array::compute::conformance::filter::test_filter_conformance;
     use vortex_array::dtype::DecimalDType;
+    use vortex_array::validity::Validity;
     use vortex_buffer::buffer;
 
     use crate::DecimalByteParts;
+    use crate::decimal_byte_parts::testing::i128_parts;
+    use crate::decimal_byte_parts::testing::i256_of;
+    use crate::decimal_byte_parts::testing::i256_parts;
 
     #[test]
     fn test_filter_decimal_byte_parts() {
@@ -59,4 +58,31 @@ mod test {
             &mut array_session().create_execution_ctx(),
         );
     }
+
+    #[test]
+    fn test_filter_decimal_byte_parts_with_lower_parts() {
+        let array = i128_parts(
+            vec![1i128 << 70, -(1i128 << 70), 5, (1i128 << 64) - 1, 0],
+            Validity::NonNullable,
+        );
+        test_filter_conformance(
+            &array.into_array(),
+            &mut array_session().create_execution_ctx(),
+        );
+
+        let array = i256_parts(
+            vec![
+                i256_of(1, 0),
+                i256_of(-1, 5),
+                i256_of(0, u128::MAX),
+                i256_of(1 << 64, 7),
+                i256_of(0, 0),
+            ],
+            Validity::from_iter([true, false, true, true, false]),
+        );
+        test_filter_conformance(
+            &array.into_array(),
+            &mut array_session().create_execution_ctx(),
+        );
+    }
 }