diff --git a/docs/specs/editions.md b/docs/specs/editions.md index ee5f494861f..1aa3cfe4e11 100644 --- a/docs/specs/editions.md +++ b/docs/specs/editions.md @@ -172,9 +172,9 @@ representation gains support for wide decimals, represented by a signed most-sig represented that way, it emits `vortex.decimal_byte_parts` with `lower_part_count = 0`, even if the current in-memory array has lower-part children. - An array that cannot be collapsed into that old form losslessly uses the new - `vortex.decimal_byte_parts_v2` component, initially staged in a draft edition. + `vortex.decimal_byte_parts.v2` component, initially staged in a draft edition. - A new reader deserializes both IDs into the same in-memory representation. An older reader reports - `vortex.decimal_byte_parts_v2` as unknown instead of trying to decode a wire format it does not support. + `vortex.decimal_byte_parts.v2` as unknown instead of trying to decode a wire format it does not support. - When targeting an edition that permits only the old ID, serializing a value that can be collapsed succeeds; an irreducibly multi-part value fails because no lossless downgrade exists. diff --git a/encodings/decimal-byte-parts/src/decimal_byte_parts/array.rs b/encodings/decimal-byte-parts/src/decimal_byte_parts/array.rs index 8c10c0f5088..c98b852d716 100644 --- a/encodings/decimal-byte-parts/src/decimal_byte_parts/array.rs +++ b/encodings/decimal-byte-parts/src/decimal_byte_parts/array.rs @@ -7,14 +7,12 @@ use std::fmt::Display; use std::fmt::Formatter; use std::hash::Hasher; -use prost::Message as _; use vortex_array::Array; use vortex_array::ArrayEq; use vortex_array::ArrayHash; use vortex_array::ArrayId; use vortex_array::ArrayParts; use vortex_array::ArrayRef; -use vortex_array::ArraySlots; use vortex_array::ArrayView; use vortex_array::EqMode; use vortex_array::ExecutionCtx; @@ -24,7 +22,6 @@ use vortex_array::array_slots; use vortex_array::buffer::BufferHandle; use vortex_array::dtype::DType; use vortex_array::dtype::DecimalDType; -use vortex_array::dtype::PType; use vortex_array::scalar::DecimalValue; use vortex_array::scalar::Scalar; use vortex_array::scalar::ScalarValue; @@ -37,92 +34,18 @@ use vortex_error::VortexExpect; use vortex_error::VortexResult; use vortex_error::vortex_bail; use vortex_error::vortex_ensure; -use vortex_error::vortex_err; use vortex_error::vortex_panic; use vortex_session::VortexSession; -use vortex_session::registry::CachedId; -use super::LOWER_PART_DTYPE; use super::MAX_LOWER_PARTS; use super::assemble::assemble_decimal; use super::assemble::assemble_wide_decimal_value; +use super::decimal_byte_parts_v2_id; use super::rules::PARENT_RULES; /// A [`DecimalByteParts`]-encoded Vortex array. pub type DecimalBytePartsArray = Array; -#[derive(Clone, prost::Message)] -pub struct DecimalBytesPartsMetadata { - #[prost(enumeration = "PType", tag = "1")] - zeroth_child_ptype: i32, - #[prost(uint32, tag = "2")] - lower_part_count: u32, -} - -impl DecimalBytesPartsMetadata { - fn from_array(array: ArrayView<'_, DecimalByteParts>) -> VortexResult { - Ok(Self { - zeroth_child_ptype: PType::try_from(array.msp().dtype())? as i32, - lower_part_count: u32::try_from(array.lower_parts().len()) - .map_err(|_| vortex_err!("lower part count exceeds u32"))?, - }) - } - - fn into_array_parts( - self, - dtype: &DType, - len: usize, - children: &dyn ArrayChildren, - ) -> VortexResult> { - vortex_ensure!( - dtype.as_decimal_opt().is_some(), - "decoding decimal but given non decimal dtype {dtype}" - ); - - let encoded_dtype = DType::Primitive(self.zeroth_child_ptype(), dtype.nullability()); - - let lower_part_count = self.lower_part_count()?; - vortex_ensure!( - children.len() == DecimalBytePartsSlots::FIXED_COUNT + lower_part_count, - "expected {} children, got {}", - DecimalBytePartsSlots::FIXED_COUNT + lower_part_count, - children.len() - ); - - let msp = children.get(DecimalBytePartsSlots::MSP, &encoded_dtype, len)?; - - let mut slots = ArraySlots::with_capacity(children.len()); - slots.push(Some(msp)); - for idx in 0..lower_part_count { - slots.push(Some(children.get( - DecimalBytePartsSlots::LOWER_PARTS_OFFSET + idx, - &LOWER_PART_DTYPE, - len, - )?)); - } - - Ok( - ArrayParts::new(DecimalByteParts, dtype.clone(), len, DecimalBytePartsData) - .with_slots(slots), - ) - } - - /// The number of lower parts encoded in this array. - /// - /// # Errors - /// - /// Returns an error if the count exceeds [`MAX_LOWER_PARTS`]. - fn lower_part_count(&self) -> VortexResult { - let count = usize::try_from(self.lower_part_count) - .map_err(|_| vortex_err!("lower part count {} out of range", self.lower_part_count))?; - vortex_ensure!( - count <= MAX_LOWER_PARTS, - "at most {MAX_LOWER_PARTS} lower parts are supported, got {count}" - ); - Ok(count) - } -} - /// This array encodes decimals by splitting them between 1-4 columns of primitive typed children. /// /// The most significant part (MSP) stores the most significant decimal bits. It is signed and is @@ -222,7 +145,7 @@ impl DecimalByteParts { /// /// Lower parts are ordered most significant first and must each be a non-nullable unsigned integer /// array of the same length as the MSP. See [`super::split_decimal`] for producing them from a - /// canonical decimal array. + /// decimal array. /// /// # Errors /// @@ -274,8 +197,7 @@ impl VTable for DecimalByteParts { type ValidityVTable = ValidityVTableFromChild; fn id(&self) -> ArrayId { - static ID: CachedId = CachedId::new("vortex.decimal_byte_parts"); - *ID + decimal_byte_parts_v2_id() } fn validate( @@ -331,33 +253,22 @@ impl VTable for DecimalByteParts { } fn serialize( - array: ArrayView<'_, Self>, + _array: ArrayView<'_, Self>, _session: &VortexSession, ) -> VortexResult>> { - vortex_ensure!( - array.lower_parts().is_empty(), - "serializing DecimalByteParts with lower parts is not supported" - ); - Ok(Some( - DecimalBytesPartsMetadata::from_array(array)?.encode_to_vec(), - )) + vortex_bail!("DecimalByteParts serialization requires DecimalBytePartsPlugin") } fn deserialize( &self, - dtype: &DType, - len: usize, - metadata: &[u8], + _dtype: &DType, + _len: usize, + _metadata: &[u8], _buffers: &[BufferHandle], - children: &dyn ArrayChildren, + _children: &dyn ArrayChildren, _session: &VortexSession, ) -> VortexResult> { - let metadata = DecimalBytesPartsMetadata::decode(metadata)?; - vortex_ensure!( - metadata.lower_part_count()? == 0, - "vortex.decimal_byte_parts must not carry lower parts" - ); - metadata.into_array_parts(dtype, len, children) + vortex_bail!("DecimalByteParts deserialization requires DecimalBytePartsPlugin") } fn slot_name(_array: ArrayView<'_, Self>, idx: usize) -> String { @@ -508,18 +419,15 @@ mod tests { use vortex_array::scalar::Scalar; use vortex_array::scalar::ScalarValue; use vortex_array::validity::Validity; - use vortex_array::vtable::VTable; use vortex_buffer::buffer; use vortex_error::VortexResult; use super::DecimalByteParts; - use super::DecimalBytePartsArray; use super::DecimalBytePartsArraySlotsExt; use super::DecimalBytePartsData; use crate::decimal_byte_parts::LOWER_PART_DTYPE; use crate::decimal_byte_parts::MAX_LOWER_PARTS; 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] @@ -561,62 +469,6 @@ mod tests { ); } - /// The largest unscaled value a `Decimal(38, _)` can hold: `10^38 - 1`. - const MAX_PRECISION_38: i128 = 99_999_999_999_999_999_999_999_999_999_999_999_999; - - /// The largest unscaled value a `Decimal(76, _)` can hold: `10^76 - 1`. - fn max_precision_76() -> i256 { - i256::from_i128(10).wrapping_pow(76) - i256::ONE - } - - /// Values that exercise every 64-bit window of an `i128`, both signs, and the boundaries - /// where a lower part carries into the MSP. - fn wide_i128_values() -> Vec { - vec![ - 0, - 1, - -1, - (1 << 64) - 1, - 1 << 64, - -(1 << 64), - -((1 << 64) + 1), - MAX_PRECISION_38, - -MAX_PRECISION_38, - 1 << 100, - ] - } - - /// Values that exercise every 64-bit window of an `i256`. - fn wide_i256_values() -> Vec { - vec![ - i256::ZERO, - i256::ONE, - i256::ZERO - i256::ONE, - i256_of(0, u128::MAX), - i256_of(1, 0), - i256_of(-1, 0), - i256_of(-1, u128::MAX - 1), - i256_of(1 << 64, 12345), - max_precision_76(), - i256::ZERO - max_precision_76(), - ] - } - - #[rstest] - #[case::i128_non_nullable(i128_parts(wide_i128_values(), Validity::NonNullable))] - #[case::i256_non_nullable(i256_parts(wide_i256_values(), Validity::NonNullable))] - fn test_canonical_decimal_round_trips( - #[case] array: DecimalBytePartsArray, - ) -> VortexResult<()> { - let mut ctx = array_session().create_execution_ctx(); - let canonical = array - .clone() - .into_array() - .execute::(&mut ctx)?; - assert_arrays_eq!(array, canonical, &mut ctx); - Ok(()) - } - #[test] fn test_lower_part_layout_i128() -> VortexResult<()> { let array = i128_parts(vec![(3i128 << 64) | 7], Validity::NonNullable); @@ -637,7 +489,7 @@ mod tests { #[test] fn test_lower_part_layout_i256() -> VortexResult<()> { let array = i256_parts( - vec![i256_of((5i128 << 64) | 6, (7u128 << 64) | 8)], + vec![i256::from_parts((7u128 << 64) | 8, (5i128 << 64) | 6)], Validity::NonNullable, ); assert_eq!(array.lower_parts().len(), MAX_LOWER_PARTS); @@ -652,27 +504,6 @@ mod tests { Ok(()) } - #[rstest] - #[case::i128(i128_parts(wide_i128_values(), Validity::AllValid))] - #[case::i256(i256_parts(wide_i256_values(), Validity::AllValid))] - fn test_scalar_at_matches_canonical(#[case] array: DecimalBytePartsArray) -> VortexResult<()> { - let mut ctx = array_session().create_execution_ctx(); - let canonical = array - .clone() - .into_array() - .execute::(&mut ctx)? - .into_array(); - let array = array.into_array(); - for idx in 0..array.len() { - assert_eq!( - array.execute_scalar(idx, &mut ctx)?, - canonical.execute_scalar(idx, &mut ctx)?, - "scalar mismatch at index {idx}" - ); - } - Ok(()) - } - #[rstest] fn test_scalar_at_matches_canonical_for_each_part_count( #[values(false, true)] narrow_msp: bool, @@ -858,7 +689,7 @@ mod tests { let canonical = i128_array.into_array().execute::(&mut ctx)?; assert_eq!(canonical.values_type(), DecimalType::I128); - let i256_array = i256_parts(vec![i256_of(1 << 100, 0)], Validity::NonNullable); + let i256_array = i256_parts(vec![i256::from_parts(0, 1 << 100)], Validity::NonNullable); let canonical = i256_array.into_array().execute::(&mut ctx)?; assert_eq!(canonical.values_type(), DecimalType::I256); @@ -883,7 +714,10 @@ mod tests { )?; let canonical = array.into_array().execute::(&mut ctx)?; assert_eq!(canonical.values_type(), DecimalType::I256); - assert_eq!(canonical.buffer::().as_slice(), &[i256_of(1, 9)]); + assert_eq!( + canonical.buffer::().as_slice(), + &[i256::from_parts(9, 1)] + ); Ok(()) } @@ -910,11 +744,4 @@ mod tests { assert_arrays_eq!(array, canonical.into_array(), &mut ctx); Ok(()) } - #[test] - fn test_frozen_serializer_rejects_lower_parts() -> VortexResult<()> { - let session = array_session(); - let array = i128_parts(vec![1i128 << 70], Validity::NonNullable); - assert!(VTable::serialize(array.as_view(), &session).is_err()); - Ok(()) - } } diff --git a/encodings/decimal-byte-parts/src/decimal_byte_parts/compute/filter.rs b/encodings/decimal-byte-parts/src/decimal_byte_parts/compute/filter.rs index 49c4021dd18..6921e5dfda4 100644 --- a/encodings/decimal-byte-parts/src/decimal_byte_parts/compute/filter.rs +++ b/encodings/decimal-byte-parts/src/decimal_byte_parts/compute/filter.rs @@ -27,12 +27,12 @@ mod test { use vortex_array::arrays::PrimitiveArray; use vortex_array::compute::conformance::filter::test_filter_conformance; use vortex_array::dtype::DecimalDType; + use vortex_array::dtype::i256; 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] @@ -72,11 +72,11 @@ mod test { 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), + i256::from_parts(0, 1), + i256::from_parts(5, -1), + i256::from_parts(u128::MAX, 0), + i256::from_parts(7, 1 << 64), + i256::from_parts(0, 0), ], Validity::from_iter([true, false, true, true, false]), ); diff --git a/encodings/decimal-byte-parts/src/decimal_byte_parts/compute/mod.rs b/encodings/decimal-byte-parts/src/decimal_byte_parts/compute/mod.rs index f9848e1b2e7..c8385c2d6d6 100644 --- a/encodings/decimal-byte-parts/src/decimal_byte_parts/compute/mod.rs +++ b/encodings/decimal-byte-parts/src/decimal_byte_parts/compute/mod.rs @@ -27,7 +27,6 @@ mod tests { 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; /// Values needing more than 64 bits, so the encoding carries lower parts. @@ -43,11 +42,11 @@ mod tests { fn wide_i256() -> Vec { vec![ - i256_of(1, 0), - i256_of(-1, 0), - i256_of(0, u128::MAX), - i256_of(1 << 64, 7), - i256_of(0, 0), + i256::from_parts(0, 1), + i256::from_parts(0, -1), + i256::from_parts(u128::MAX, 0), + i256::from_parts(7, 1 << 64), + i256::from_parts(0, 0), ] } diff --git a/encodings/decimal-byte-parts/src/decimal_byte_parts/compute/take.rs b/encodings/decimal-byte-parts/src/decimal_byte_parts/compute/take.rs index 5b07af47252..a74915e9f22 100644 --- a/encodings/decimal-byte-parts/src/decimal_byte_parts/compute/take.rs +++ b/encodings/decimal-byte-parts/src/decimal_byte_parts/compute/take.rs @@ -38,14 +38,13 @@ mod tests { use vortex_array::arrays::PrimitiveArray; use vortex_array::assert_arrays_eq; use vortex_array::dtype::DecimalDType; + use vortex_array::dtype::i256; use vortex_array::validity::Validity; use vortex_buffer::Buffer; use vortex_buffer::buffer; use vortex_error::VortexResult; use crate::DecimalByteParts; - use crate::decimal_byte_parts::testing::encode; - use crate::decimal_byte_parts::testing::i256_of; /// Taking pushes down into the parts during optimization, with no execution context in /// play: `ArrayRef::take` wraps the array in a `Dict` and optimizes, and the reduce rule @@ -61,7 +60,9 @@ mod tests { Validity::NonNullable, ); let indices = buffer![0u64, 2].into_array(); - let taken = encode(&decimal)?.into_array().take(indices)?; + let taken = DecimalByteParts::encode(&decimal, &mut session.create_execution_ctx())? + .into_array() + .take(indices)?; assert!( taken.is::(), @@ -80,7 +81,7 @@ mod tests { Validity::NonNullable, ))] #[case::three_lower_parts(DecimalArray::new( - Buffer::from(vec![i256_of(1, 1 << 70), i256_of(0, 2), i256_of(0, 3)]), + Buffer::from(vec![i256::from_parts(1 << 70, 1), i256::from_parts(2, 0), i256::from_parts(3, 0)]), DecimalDType::new(76, 2), Validity::NonNullable, ))] @@ -96,7 +97,9 @@ mod tests { .take(indices.clone())? .execute::(&mut ctx)?; - let taken = encode(&decimal)?.into_array().take(indices)?; + let taken = DecimalByteParts::encode(&decimal, &mut ctx)? + .into_array() + .take(indices)?; let actual = taken.execute::(&mut ctx)?; assert_arrays_eq!(expected, actual, &mut ctx); diff --git a/encodings/decimal-byte-parts/src/decimal_byte_parts/mod.rs b/encodings/decimal-byte-parts/src/decimal_byte_parts/mod.rs index 919b7bb44a3..65bb8222f7a 100644 --- a/encodings/decimal-byte-parts/src/decimal_byte_parts/mod.rs +++ b/encodings/decimal-byte-parts/src/decimal_byte_parts/mod.rs @@ -22,6 +22,7 @@ use vortex_array::dtype::PType; mod array; mod assemble; pub(crate) mod compute; +mod plugin; #[cfg(test)] mod prop_tests; mod rules; @@ -30,8 +31,11 @@ mod split; mod testing; pub use array::*; +pub use plugin::DecimalBytePartsPlugin; +pub use plugin::DecimalBytePartsV2Metadata; +pub use plugin::decimal_byte_parts_v1_id; +pub use plugin::decimal_byte_parts_v2_id; pub use split::DecimalParts; -pub use split::dbp_encode; pub use split::split_decimal; #[doc(hidden)] diff --git a/encodings/decimal-byte-parts/src/decimal_byte_parts/plugin/mod.rs b/encodings/decimal-byte-parts/src/decimal_byte_parts/plugin/mod.rs new file mode 100644 index 00000000000..3896cd51c96 --- /dev/null +++ b/encodings/decimal-byte-parts/src/decimal_byte_parts/plugin/mod.rs @@ -0,0 +1,98 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! ArrayPlugin implementation for DBP that handles different wire formats. + +use vortex_array::ArrayDeserialization; +use vortex_array::ArrayId; +use vortex_array::ArrayPlugin; +use vortex_array::ArrayRef; +use vortex_array::ArraySerialization; +use vortex_array::IntoArray; +use vortex_array::VTable; +use vortex_error::VortexResult; +use vortex_error::vortex_bail; +use vortex_error::vortex_err; +use vortex_session::VortexSession; +use vortex_session::registry::CachedId; + +use super::DecimalByteParts; +use super::DecimalBytePartsArraySlotsExt; + +#[cfg(test)] +mod tests; + +mod v1; +mod v2; + +pub use v2::DecimalBytePartsV2Metadata; + +/// The frozen single-child DBP serialized ID. +pub fn decimal_byte_parts_v1_id() -> ArrayId { + static ID: CachedId = CachedId::new("vortex.decimal_byte_parts"); + *ID +} + +/// The current in-memory DBP ID and serialized ID for arrays with lower parts. +pub fn decimal_byte_parts_v2_id() -> ArrayId { + static ID: CachedId = CachedId::new("vortex.decimal_byte_parts.v2"); + *ID +} + +/// Serde for the [`DecimalByteParts`] array using the frozen v1 and v2 wire formats. +/// +/// Each version owns its metadata schema and serde functions. The plugin writes v1 whenever an +/// array has no lower parts, so such arrays stay readable by older readers, and v2 otherwise. +/// The v2 format itself accepts any lower part count up to the maximum. +/// +/// Register this plugin, or call [`crate::initialize`], to enable both formats. Direct registration +/// of [`DecimalByteParts`] does not support serde. +#[derive(Clone, Debug)] +pub struct DecimalBytePartsPlugin; + +impl ArrayPlugin for DecimalBytePartsPlugin { + fn id(&self) -> ArrayId { + VTable::id(&DecimalByteParts) + } + + fn serialized_ids(&self) -> Vec { + vec![decimal_byte_parts_v1_id(), decimal_byte_parts_v2_id()] + } + + fn serialize( + &self, + array: &ArrayRef, + _session: &VortexSession, + ) -> VortexResult> { + let view = array.as_opt::().ok_or_else(|| { + vortex_err!( + "DecimalByteParts plugin cannot serialize {}", + array.encoding_id() + ) + })?; + let serialized = if view.lower_parts().is_empty() { + v1::serialize(view)? + } else { + v2::serialize(view)? + }; + Ok(Some(serialized)) + } + + fn deserialize( + &self, + parts: ArrayDeserialization<'_>, + _session: &VortexSession, + ) -> VortexResult { + let array = if parts.serialized_id == decimal_byte_parts_v1_id() { + v1::deserialize(parts)? + } else if parts.serialized_id == decimal_byte_parts_v2_id() { + v2::deserialize(parts)? + } else { + vortex_bail!( + "DecimalByteParts plugin does not recognize serialized ID {}", + parts.serialized_id + ) + }; + Ok(array.into_array()) + } +} diff --git a/encodings/decimal-byte-parts/src/decimal_byte_parts/plugin/tests.rs b/encodings/decimal-byte-parts/src/decimal_byte_parts/plugin/tests.rs new file mode 100644 index 00000000000..92c9d1edbb7 --- /dev/null +++ b/encodings/decimal-byte-parts/src/decimal_byte_parts/plugin/tests.rs @@ -0,0 +1,324 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +use prost::Message as _; +use rstest::rstest; +use vortex_array::ArrayContext; +use vortex_array::ArrayDeserialization; +use vortex_array::ArrayVTable; +use vortex_array::VortexSessionExecute; +use vortex_array::array_session; +use vortex_array::arrays::ConstantArray; +use vortex_array::arrays::DecimalArray; +use vortex_array::arrays::Primitive; +use vortex_array::arrays::PrimitiveArray; +use vortex_array::assert_arrays_eq; +use vortex_array::dtype::DType; +use vortex_array::dtype::DecimalDType; +use vortex_array::dtype::Nullability; +use vortex_array::dtype::PType; +use vortex_array::dtype::i256; +use vortex_array::serde::SerializeOptions; +use vortex_array::serde::SerializedArray; +use vortex_array::session::ArraySessionExt; +use vortex_array::validity::Validity; +use vortex_buffer::ByteBufferMut; +use vortex_buffer::buffer; +use vortex_error::VortexExpect; +use vortex_session::registry::ReadContext; + +use super::*; +use crate::DecimalBytePartsArray; +use crate::DecimalBytePartsArraySlotsExt; +use crate::decimal_byte_parts::MAX_LOWER_PARTS; + +#[rstest] +#[case::no_lower_parts(DecimalByteParts::try_new( + buffer![1i32, 2, 3].into_array(), DecimalDType::new(9, 2), +))] +#[case::one_lower_part(DecimalByteParts::try_new_with_lower_parts( + msp(), vec![lower_part()], DecimalDType::new(38, 2), +))] +#[case::wider_i64_storage(DecimalByteParts::encode( + &DecimalArray::new(buffer![-99i64, 0, 99], DecimalDType::new(2, 0), Validity::NonNullable), + &mut array_session().create_execution_ctx(), +))] +#[case::wider_i128_storage(DecimalByteParts::encode( + &DecimalArray::new(buffer![-99i128, 0, 99], DecimalDType::new(2, 0), Validity::NonNullable), + &mut array_session().create_execution_ctx(), +))] +#[case::wider_i256_storage(DecimalByteParts::encode( + &DecimalArray::new( + buffer![i256::from_i128(-99), i256::ZERO, i256::from_i128(99)], + DecimalDType::new(2, 0), Validity::NonNullable, + ), + &mut array_session().create_execution_ctx(), +))] +#[case::redundant_lower_parts(DecimalByteParts::try_new_with_lower_parts( + buffer![0i64; 3].into_array(), + vec![buffer![0u64; 3].into_array(), buffer![0u64; 3].into_array(), lower_part()], + DecimalDType::new(38, 2), +))] +#[case::narrowed_lower_parts(DecimalByteParts::try_new_with_lower_parts( + buffer![-1i16, 0, 1].into_array(), + vec![ + buffer![u8::MAX, 128, 0].into_array(), + ConstantArray::new(u16::MAX, 3).into_array(), + buffer![0u32, 1 << 31, u32::MAX].into_array(), + ], + DecimalDType::new(76, 2), +))] +#[case::nullable_mixed_lower_parts(DecimalByteParts::try_new_with_lower_parts( + PrimitiveArray::new( + buffer![-1i8, 0, 1], Validity::from_iter([true, false, true]), + ).into_array(), + vec![ + buffer![u64::MAX, 1 << 63, 0].into_array(), + buffer![0u8, 128, u8::MAX].into_array(), + buffer![u32::MAX, 1 << 31, 0].into_array(), + ], + DecimalDType::new(76, 2), +))] +fn serde_round_trip(#[case] array: VortexResult) -> VortexResult<()> { + let session = session(); + let array = array?; + let lower_part_count = array.lower_parts().len(); + let lower_part_dtypes: Vec<_> = array + .lower_parts() + .iter() + .map(|part| part.dtype().clone()) + .collect(); + let array = array.into_array(); + let dtype = array.dtype().clone(); + let len = array.len(); + + let expected_id = if lower_part_count == 0 { + decimal_byte_parts_v1_id() + } else { + decimal_byte_parts_v2_id() + }; + assert_eq!( + session + .array_serialize(&array)? + .vortex_expect("byte parts arrays are serializable") + .serialized_id, + expected_id + ); + + let array_ctx = ArrayContext::empty(); + let serialized = array.serialize(&array_ctx, &session, &SerializeOptions::default())?; + let mut concat = ByteBufferMut::empty(); + for buf in serialized { + concat.extend_from_slice(buf.as_ref()); + } + let parts = SerializedArray::try_from(concat.freeze())?; + let decoded = parts.decode(&dtype, len, &ReadContext::new(array_ctx.to_ids()), &session)?; + + assert_eq!( + decoded + .as_opt::() + .vortex_expect("byte parts array") + .lower_parts() + .iter() + .map(|part| part.dtype().clone()) + .collect::>(), + lower_part_dtypes, + "lower-part dtypes and order must survive serde" + ); + + let mut ctx = session.create_execution_ctx(); + assert_arrays_eq!(array, decoded, &mut ctx); + Ok(()) +} + +#[test] +fn v1_metadata_is_unchanged() -> VortexResult<()> { + let session = session(); + let array = DecimalByteParts::try_new(msp(), DecimalDType::new(19, 2))?.into_array(); + let serialized = session + .array_serialize(&array)? + .vortex_expect("byte parts arrays are serializable"); + assert_eq!(serialized.serialized_id, decimal_byte_parts_v1_id()); + // v1 metadata for an i64 MSP: field 1 = 7, with no lower-part fields emitted. + assert_eq!(serialized.metadata, [8, 7]); + Ok(()) +} + +#[rstest] +#[case::v1_lower_part_count( + decimal_byte_parts_v1_id(), + v1_metadata(1), + vec![msp(), lower_part()], + "must not carry lower parts" +)] +#[case::v1_extra_child( + decimal_byte_parts_v1_id(), + v1_metadata(0), + vec![msp(), lower_part()], + "exactly one child" +)] +#[case::v2_missing_child( + decimal_byte_parts_v2_id(), + v2_metadata(vec![PType::U64 as i32]), + vec![msp()], + "expected 2 children, got 1" +)] +#[case::v2_extra_child( + decimal_byte_parts_v2_id(), + v2_metadata(vec![PType::U64 as i32]), + vec![msp(), lower_part(), lower_part()], + "expected 2 children, got 3" +)] +#[case::v2_too_many_lower_parts( + decimal_byte_parts_v2_id(), + v2_metadata(vec![PType::U64 as i32; MAX_LOWER_PARTS + 1]), + vec![msp(), lower_part(), lower_part(), lower_part(), lower_part()], + "lower parts, got 4" +)] +#[case::v2_signed_lower_part( + decimal_byte_parts_v2_id(), + v2_metadata(vec![PType::I64 as i32]), + vec![msp(), lower_part()], + "unsigned integer dtype" +)] +#[case::v2_unknown_ptype( + decimal_byte_parts_v2_id(), + v2_metadata(vec![i32::MAX]), + vec![msp(), lower_part()], + "invalid PType" +)] +fn decoder_rejects_malformed_payloads( + #[case] serialized_id: ArrayId, + #[case] metadata: Vec, + #[case] children: Vec, + #[case] expected_error: &str, +) { + let result = deserialize_with(serialized_id, &metadata, children); + assert!( + result + .as_ref() + .is_err_and(|err| err.to_string().contains(expected_error)), + "expected {expected_error}, got {result:?}" + ); +} + +#[test] +fn serialization_requires_v2_permission() -> VortexResult<()> { + let session = session(); + let array = DecimalByteParts::try_new_with_lower_parts( + msp(), + vec![lower_part()], + DecimalDType::new(38, 2), + )? + .into_array(); + + let restricted = ArrayContext::empty().with_allowed_ids( + [decimal_byte_parts_v1_id(), ArrayVTable::id(&Primitive)] + .into_iter() + .collect(), + ); + let err = array + .serialize(&restricted, &session, &SerializeOptions::default()) + .expect_err("expected the permitted-encoding check to refuse the v2 format"); + assert!( + err.to_string().contains("not permitted"), + "error should name the permitted-encoding check, got: {err}" + ); + + // Permitting the v2 format id is exactly what allows the same array through. + let permissive = ArrayContext::empty().with_allowed_ids( + [ + decimal_byte_parts_v1_id(), + decimal_byte_parts_v2_id(), + ArrayVTable::id(&Primitive), + ] + .into_iter() + .collect(), + ); + array.serialize(&permissive, &session, &SerializeOptions::default())?; + assert!( + permissive.to_ids().contains(&decimal_byte_parts_v2_id()), + "the file's encoding table must carry the v2 format id" + ); + + Ok(()) +} + +#[test] +fn bare_vtable_refuses_serde() -> VortexResult<()> { + let session = array_session(); + session.arrays().register(DecimalByteParts); + let msp = msp(); + let array = DecimalByteParts::try_new(msp.clone(), DecimalDType::new(19, 2))?.into_array(); + let result = session.array_serialize(&array); + assert!( + result.as_ref().is_err_and(|err| err + .to_string() + .contains("DecimalByteParts serialization requires DecimalBytePartsPlugin")), + "expected unsupported VTable serialization, got {result:?}" + ); + + let id = VTable::id(&DecimalByteParts); + let plugin = session + .arrays() + .registry() + .get(&id) + .vortex_expect("registered"); + let children = vec![msp]; + let result = plugin.deserialize( + ArrayDeserialization::new(id, array.dtype(), array.len(), &[8, 7], &[], &children), + &session, + ); + assert!( + result.as_ref().is_err_and(|err| err + .to_string() + .contains("DecimalByteParts deserialization requires DecimalBytePartsPlugin")), + "expected unsupported VTable deserialization, got {result:?}" + ); + Ok(()) +} + +fn msp() -> ArrayRef { + buffer![1i64, 2, 3].into_array() +} + +fn lower_part() -> ArrayRef { + buffer![1u64, 2, 3].into_array() +} + +/// v1 metadata for an i64 MSP: field 1 = 7, then field 2 only when the count is non-zero, as +/// proto3 omits default values. +fn v1_metadata(lower_part_count: u8) -> Vec { + let mut metadata = vec![8, 7]; + if lower_part_count > 0 { + metadata.extend([16, lower_part_count]); + } + metadata +} + +fn v2_metadata(lower_part_ptypes: Vec) -> Vec { + DecimalBytePartsV2Metadata { + msp_ptype: PType::I64 as i32, + lower_part_ptypes, + } + .encode_to_vec() +} + +fn session() -> VortexSession { + let session = array_session(); + crate::initialize(&session); + session +} + +/// Decode a hand-built payload of three rows through the plugin. +fn deserialize_with( + serialized_id: ArrayId, + metadata: &[u8], + children: Vec, +) -> VortexResult { + let dtype = DType::Decimal(DecimalDType::new(38, 2), Nullability::NonNullable); + DecimalBytePartsPlugin.deserialize( + ArrayDeserialization::new(serialized_id, &dtype, 3, metadata, &[], &children), + &array_session(), + ) +} diff --git a/encodings/decimal-byte-parts/src/decimal_byte_parts/plugin/v1.rs b/encodings/decimal-byte-parts/src/decimal_byte_parts/plugin/v1.rs new file mode 100644 index 00000000000..ec816c62736 --- /dev/null +++ b/encodings/decimal-byte-parts/src/decimal_byte_parts/plugin/v1.rs @@ -0,0 +1,84 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Serde for the frozen single-child DBP wire format. + +use prost::Message as _; +use vortex_array::Array; +use vortex_array::ArrayDeserialization; +use vortex_array::ArrayParts; +use vortex_array::ArraySerialization; +use vortex_array::ArrayView; +use vortex_array::dtype::DType; +use vortex_array::dtype::PType; +use vortex_array::smallvec::smallvec; +use vortex_error::VortexResult; +use vortex_error::vortex_ensure; + +use super::decimal_byte_parts_v1_id; +use crate::DecimalByteParts; +use crate::DecimalBytePartsArray; +use crate::DecimalBytePartsArraySlotsExt; +use crate::DecimalBytePartsData; + +#[derive(Clone, prost::Message)] +struct DecimalBytePartsMetadata { + #[prost(enumeration = "PType", tag = "1")] + zeroth_child_ptype: i32, + #[prost(uint32, tag = "2")] + lower_part_count: u32, +} + +pub(super) fn serialize( + array: ArrayView<'_, DecimalByteParts>, +) -> VortexResult { + vortex_ensure!( + array.lower_parts().is_empty(), + "v1 must not carry lower parts" + ); + let msp = array.msp(); + let metadata = DecimalBytePartsMetadata { + zeroth_child_ptype: PType::try_from(msp.dtype())? as i32, + lower_part_count: 0, + } + .encode_to_vec(); + Ok(ArraySerialization::new( + decimal_byte_parts_v1_id(), + metadata, + vec![], + vec![msp.clone()], + )) +} + +pub(super) fn deserialize(parts: ArrayDeserialization<'_>) -> VortexResult { + vortex_ensure!( + parts.serialized_id == decimal_byte_parts_v1_id(), + "expected the v1 format" + ); + let metadata = DecimalBytePartsMetadata::decode(parts.metadata)?; + vortex_ensure!( + parts.dtype.as_decimal_opt().is_some(), + "expected a decimal dtype" + ); + vortex_ensure!( + metadata.lower_part_count == 0, + "v1 must not carry lower parts" + ); + vortex_ensure!(parts.children.len() == 1, "v1 must carry exactly one child"); + let ptype = PType::try_from(metadata.zeroth_child_ptype)?; + vortex_ensure!( + ptype.is_signed_int(), + "MSP must have a signed integer dtype" + ); + let encoded_dtype = DType::Primitive(ptype, parts.dtype.nullability()); + let msp = parts.children.get(0, &encoded_dtype, parts.len)?; + Array::try_from_parts( + ArrayParts::new( + DecimalByteParts, + parts.dtype.clone(), + parts.len, + DecimalBytePartsData, + ) + .with_slots(smallvec![Some(msp)]), + ) +} diff --git a/encodings/decimal-byte-parts/src/decimal_byte_parts/plugin/v2.rs b/encodings/decimal-byte-parts/src/decimal_byte_parts/plugin/v2.rs new file mode 100644 index 00000000000..c82db1cbce3 --- /dev/null +++ b/encodings/decimal-byte-parts/src/decimal_byte_parts/plugin/v2.rs @@ -0,0 +1,114 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Serde for DBP values with unsigned lower parts. + +use prost::Message as _; +use vortex_array::Array; +use vortex_array::ArrayDeserialization; +use vortex_array::ArrayParts; +use vortex_array::ArraySerialization; +use vortex_array::ArraySlots; +use vortex_array::ArrayView; +use vortex_array::dtype::DType; +use vortex_array::dtype::Nullability; +use vortex_array::dtype::PType; +use vortex_error::VortexResult; +use vortex_error::vortex_ensure; +use vortex_error::vortex_err; + +use super::decimal_byte_parts_v2_id; +use crate::DecimalByteParts; +use crate::DecimalBytePartsArray; +use crate::DecimalBytePartsArraySlotsExt; +use crate::DecimalBytePartsData; +use crate::decimal_byte_parts::MAX_LOWER_PARTS; + +/// Metadata for decimal byte parts with lower parts. +#[derive(Clone, prost::Message)] +pub struct DecimalBytePartsV2Metadata { + /// Ptype of the most significant part. + #[prost(enumeration = "PType", tag = "1")] + pub(super) msp_ptype: i32, + /// Ptypes of the lower parts, ordered most significant first. + #[prost(enumeration = "PType", repeated, tag = "2")] + pub(super) lower_part_ptypes: Vec, +} + +pub(super) fn serialize( + array: ArrayView<'_, DecimalByteParts>, +) -> VortexResult { + let lower_parts = array.lower_parts(); + + let metadata = DecimalBytePartsV2Metadata { + msp_ptype: PType::try_from(array.msp().dtype())? as i32, + lower_part_ptypes: lower_parts + .iter() + .map(|part| PType::try_from(part.dtype()).map(|ptype| ptype as i32)) + .collect::>()?, + } + .encode_to_vec(); + + let mut children = Vec::with_capacity(1 + lower_parts.len()); + children.push(array.msp().clone()); + children.extend(lower_parts.iter().cloned()); + + Ok(ArraySerialization::new( + decimal_byte_parts_v2_id(), + metadata, + vec![], + children, + )) +} + +pub(super) fn deserialize(parts: ArrayDeserialization<'_>) -> VortexResult { + let metadata = DecimalBytePartsV2Metadata::decode(parts.metadata)?; + vortex_ensure!( + parts.dtype.as_decimal_opt().is_some(), + "expected a decimal dtype" + ); + + let lower_part_count = metadata.lower_part_ptypes.len(); + vortex_ensure!( + lower_part_count <= MAX_LOWER_PARTS, + "v2 carries at most {MAX_LOWER_PARTS} lower parts, got {lower_part_count}" + ); + vortex_ensure!( + parts.children.len() == 1 + lower_part_count, + "expected {} children, got {}", + 1 + lower_part_count, + parts.children.len() + ); + + let msp_ptype = PType::try_from(metadata.msp_ptype)?; + vortex_ensure!( + msp_ptype.is_signed_int(), + "MSP must have a signed integer dtype, got {msp_ptype}" + ); + let msp_dtype = DType::Primitive(msp_ptype, parts.dtype.nullability()); + + let mut slots = ArraySlots::with_capacity(parts.children.len()); + slots.push(Some(parts.children.get(0, &msp_dtype, parts.len)?)); + for (idx, raw_ptype) in metadata.lower_part_ptypes.into_iter().enumerate() { + let ptype = PType::try_from(raw_ptype) + .map_err(|_| vortex_err!("invalid PType {raw_ptype} for lower part {idx}"))?; + vortex_ensure!( + ptype.is_unsigned_int(), + "lower part {idx} must have an unsigned integer dtype, got {ptype}" + ); + slots.push(Some(parts.children.get( + 1 + idx, + &DType::Primitive(ptype, Nullability::NonNullable), + parts.len, + )?)); + } + Array::try_from_parts( + ArrayParts::new( + DecimalByteParts, + parts.dtype.clone(), + parts.len, + DecimalBytePartsData, + ) + .with_slots(slots), + ) +} diff --git a/encodings/decimal-byte-parts/src/decimal_byte_parts/prop_tests.rs b/encodings/decimal-byte-parts/src/decimal_byte_parts/prop_tests.rs index 5630d1706a8..b1625485ee7 100644 --- a/encodings/decimal-byte-parts/src/decimal_byte_parts/prop_tests.rs +++ b/encodings/decimal-byte-parts/src/decimal_byte_parts/prop_tests.rs @@ -21,7 +21,6 @@ use vortex_error::VortexExpect; use super::DecimalByteParts; use super::DecimalBytePartsArray; -use super::testing::encode; /// Largest magnitude a `Decimal(38, _)` can hold: 38 nines. const MAX_I128: i128 = 10i128.pow(38) - 1; @@ -139,10 +138,8 @@ fn decoded_survives_encode_then_decode(tc: TestCase) { let decimal = draw_decimal(&tc); let mut ctx = ctx(); - let round_tripped = canonicalize( - encode(&decimal).vortex_expect("encode").into_array(), - &mut ctx, - ); + let encoded = DecimalByteParts::encode(&decimal, &mut ctx).vortex_expect("encode"); + let round_tripped = canonicalize(encoded.into_array(), &mut ctx); assert_eq!(round_tripped.values_type(), decimal.values_type()); assert_arrays_eq!(decimal, round_tripped, &mut ctx); @@ -160,10 +157,8 @@ fn encoded_survives_decode_then_encode(tc: TestCase) { let mut ctx = ctx(); let decoded = canonicalize(array.into_array(), &mut ctx); - let re_decoded = canonicalize( - encode(&decoded).vortex_expect("encode").into_array(), - &mut ctx, - ); + let re_encoded = DecimalByteParts::encode(&decoded, &mut ctx).vortex_expect("encode"); + let re_decoded = canonicalize(re_encoded.into_array(), &mut ctx); assert_arrays_eq!(decoded, re_decoded, &mut ctx); } diff --git a/encodings/decimal-byte-parts/src/decimal_byte_parts/split.rs b/encodings/decimal-byte-parts/src/decimal_byte_parts/split.rs index a3aa1b4afdc..8d8818924e9 100644 --- a/encodings/decimal-byte-parts/src/decimal_byte_parts/split.rs +++ b/encodings/decimal-byte-parts/src/decimal_byte_parts/split.rs @@ -27,22 +27,23 @@ use super::LOWER_PART_BITS; use super::MAX_I128_LOWER_PARTS; use super::MAX_I256_LOWER_PARTS; -/// Create a [`DecimalBytePartsArray`] from a [`DecimalArray`] by splitting it into parts. -/// -/// # Errors -/// -/// Returns an error if the decimal cannot be split. -pub fn dbp_encode( - decimal: &DecimalArray, - exec_ctx: &mut ExecutionCtx, -) -> VortexResult { - let parts = split_decimal(decimal, exec_ctx)?; - // SAFETY: splitting produces a signed MSP and zero, one, or three non-nullable u64 lower - // parts, all with the decimal's length and in most-significant-first order. This also holds - // for the constant parts used for empty and all-null inputs. The decimal dtype is preserved. - Ok(unsafe { - DecimalByteParts::new_unchecked(parts.msp, parts.lower_parts, decimal.decimal_dtype()) - }) +impl DecimalByteParts { + /// Encode a [`DecimalArray`] as byte parts, splitting wide values into lower parts. + /// + /// # Errors + /// + /// Returns an error if the decimal cannot be split. + pub fn encode( + decimal: &DecimalArray, + exec_ctx: &mut ExecutionCtx, + ) -> VortexResult { + let parts = split_decimal(decimal, exec_ctx)?; + // SAFETY: splitting produces a signed MSP and zero, one, or three non-nullable u64 lower + // parts, all with the decimal's length and in most-significant-first order. This also + // holds for the constant parts used for empty and all-null inputs. The decimal dtype is + // preserved. + Ok(unsafe { Self::new_unchecked(parts.msp, parts.lower_parts, decimal.decimal_dtype()) }) + } } /// A decimal array decomposed into byte parts. diff --git a/encodings/decimal-byte-parts/src/decimal_byte_parts/testing.rs b/encodings/decimal-byte-parts/src/decimal_byte_parts/testing.rs index 2dfe2a55b3c..af270bad407 100644 --- a/encodings/decimal-byte-parts/src/decimal_byte_parts/testing.rs +++ b/encodings/decimal-byte-parts/src/decimal_byte_parts/testing.rs @@ -11,37 +11,24 @@ use vortex_array::dtype::i256; use vortex_array::validity::Validity; use vortex_buffer::Buffer; use vortex_error::VortexExpect; -use vortex_error::VortexResult; +use super::DecimalByteParts; use super::DecimalBytePartsArray; -use super::dbp_encode; - -/// Encode a canonical decimal array as byte parts, splitting wide values into lower parts. -pub(crate) fn encode(decimal: &DecimalArray) -> VortexResult { - dbp_encode(decimal, &mut array_session().create_execution_ctx()) -} /// An `i128`-backed decimal array, encoded as byte parts with one lower part. pub(crate) fn i128_parts(values: Vec, validity: Validity) -> DecimalBytePartsArray { - encode(&DecimalArray::new( - Buffer::from(values), - DecimalDType::new(38, 2), - validity, - )) + DecimalByteParts::encode( + &DecimalArray::new(Buffer::from(values), DecimalDType::new(38, 2), validity), + &mut array_session().create_execution_ctx(), + ) .vortex_expect("valid decimal byte parts") } /// An `i256`-backed decimal array, encoded as byte parts with three lower parts. pub(crate) fn i256_parts(values: Vec, validity: Validity) -> DecimalBytePartsArray { - encode(&DecimalArray::new( - Buffer::from(values), - DecimalDType::new(76, 2), - validity, - )) + DecimalByteParts::encode( + &DecimalArray::new(Buffer::from(values), DecimalDType::new(76, 2), validity), + &mut array_session().create_execution_ctx(), + ) .vortex_expect("valid decimal byte parts") } - -/// Build an `i256` from a signed high `i128` and unsigned low `u128`. -pub(super) fn i256_of(high: i128, low: u128) -> i256 { - i256::from_parts(low, high) -} diff --git a/encodings/decimal-byte-parts/src/lib.rs b/encodings/decimal-byte-parts/src/lib.rs index 36a53c3a614..2557555eac8 100644 --- a/encodings/decimal-byte-parts/src/lib.rs +++ b/encodings/decimal-byte-parts/src/lib.rs @@ -22,7 +22,9 @@ use vortex_session::VortexSession; /// Initialize decimal-byte-parts encoding in the given session. pub fn initialize(session: &VortexSession) { - session.arrays().register(DecimalByteParts); + // One plugin owns both serialized formats: registering it reads either ID and writes the + // one that fits the array. Which of them a writer may emit is decided by its editions. + session.arrays().register(DecimalBytePartsPlugin); compute::kernel::initialize(session); session.aggregate_fns().register_aggregate_kernel( diff --git a/vortex-btrblocks/src/trace_tests.rs b/vortex-btrblocks/src/trace_tests.rs index f173440f26d..4f7161b7bba 100644 --- a/vortex-btrblocks/src/trace_tests.rs +++ b/vortex-btrblocks/src/trace_tests.rs @@ -253,7 +253,7 @@ fn trace_scan_compare_on_compressed_quantity() -> VortexResult<()> { optimize root=vortex.binary(bool, len=4096) session=false reduce_parent static:DictionaryScalarFnValuesPushDownRule slot=0 parent=vortex.binary(bool, len=4096) child=vortex.dict(i16, len=4096) -> vortex.dict(bool, len=4096) done output=vortex.dict(bool, len=4096) - child_execute_parent session[0]:execute_parent_fn slot=0 parent=vortex.binary(bool, len=4096) child=vortex.decimal_byte_parts(decimal(15,2), len=4096) -> vortex.dict(bool, len=4096) + child_execute_parent session[0]:execute_parent_fn slot=0 parent=vortex.binary(bool, len=4096) child=vortex.decimal_byte_parts.v2(decimal(15,2), len=4096) -> vortex.dict(bool, len=4096) iter 1 current=vortex.dict(bool, len=4096) builder_active=false ExecuteSlot slot=0 parent=vortex.dict(bool, len=4096) child=fastlanes.bitpacked(u8, len=4096) iter 2 current=fastlanes.bitpacked(u8, len=4096) stack_parent=vortex.dict(bool, len=4096) slot=0 builder_active=false @@ -418,8 +418,8 @@ fn trace_scan_filter_on_compressed_table() -> VortexResult<()> { optimize root=vortex.filter(i16, len=43) session=false reduce_parent static:FilterReduceAdaptor(Dict) slot=0 parent=vortex.filter(i16, len=43) child=vortex.dict(i16, len=4096) -> vortex.dict(i16, len=43) done output=vortex.dict(i16, len=43) - reduce_parent static:FilterReduceAdaptor(DecimalByteParts) slot=0 parent=vortex.filter(decimal(15,2), len=43) child=vortex.decimal_byte_parts(decimal(15,2), len=4096) -> vortex.decimal_byte_parts(decimal(15,2), len=43) - done output=vortex.decimal_byte_parts(decimal(15,2), len=43) + reduce_parent static:FilterReduceAdaptor(DecimalByteParts) slot=0 parent=vortex.filter(decimal(15,2), len=43) child=vortex.decimal_byte_parts.v2(decimal(15,2), len=4096) -> vortex.decimal_byte_parts.v2(decimal(15,2), len=43) + done output=vortex.decimal_byte_parts.v2(decimal(15,2), len=43) optimize root=vortex.filter(vortex.date[days](i32), len=43) session=false optimize root=vortex.filter(i32, len=43) session=false reduce_parent static:FoRFilterPushDownRule slot=0 parent=vortex.filter(i32, len=43) child=fastlanes.for(i32, len=4096) -> fastlanes.for(i32, len=43) @@ -455,8 +455,8 @@ fn trace_scan_take_on_compressed_table() -> VortexResult<()> { insta::assert_snapshot!(optimized.trace.to_string(), @" optimize root=vortex.dict({l_quantity=decimal(15,2), l_shipdate=vortex.date[days](i32), l_shipmode=utf8}, len=64) session=false optimize root=vortex.dict(decimal(15,2), len=64) session=false - reduce_parent static:TakeReduceAdaptor(DecimalByteParts) slot=1 parent=vortex.dict(decimal(15,2), len=64) child=vortex.decimal_byte_parts(decimal(15,2), len=4096) -> vortex.decimal_byte_parts(decimal(15,2), len=64) - done output=vortex.decimal_byte_parts(decimal(15,2), len=64) + reduce_parent static:TakeReduceAdaptor(DecimalByteParts) slot=1 parent=vortex.dict(decimal(15,2), len=64) child=vortex.decimal_byte_parts.v2(decimal(15,2), len=4096) -> vortex.decimal_byte_parts.v2(decimal(15,2), len=64) + done output=vortex.decimal_byte_parts.v2(decimal(15,2), len=64) optimize root=vortex.dict(vortex.date[days](i32), len=64) session=false reduce_parent static:TakeReduceAdaptor(Extension) slot=1 parent=vortex.dict(vortex.date[days](i32), len=64) child=vortex.ext(vortex.date[days](i32), len=4096) -> vortex.ext(vortex.date[days](i32), len=64) done output=vortex.ext(vortex.date[days](i32), len=64) diff --git a/vortex-btrblocks/tests/snapshots/golden__compact__decimal_prices.snap b/vortex-btrblocks/tests/snapshots/golden__compact__decimal_prices.snap index 780ef30a6b0..fad1b4d2b4e 100644 --- a/vortex-btrblocks/tests/snapshots/golden__compact__decimal_prices.snap +++ b/vortex-btrblocks/tests/snapshots/golden__compact__decimal_prices.snap @@ -3,7 +3,7 @@ source: vortex-btrblocks/tests/golden.rs expression: rendered --- input: decimal(12,2), len=16384, nbytes=131072 -root: vortex.decimal_byte_parts(decimal(12,2), len=16384) nbytes=47666 +root: vortex.decimal_byte_parts.v2(decimal(12,2), len=16384) nbytes=47666 metadata: msp: vortex.pco(i32, len=16384) nbytes=47666 metadata: ptype: i32, nrows: 16384, slice: 0..16384 diff --git a/vortex-btrblocks/tests/snapshots/golden__default__decimal_prices.snap b/vortex-btrblocks/tests/snapshots/golden__default__decimal_prices.snap index f669755e4b1..6eb4ccfed8d 100644 --- a/vortex-btrblocks/tests/snapshots/golden__default__decimal_prices.snap +++ b/vortex-btrblocks/tests/snapshots/golden__default__decimal_prices.snap @@ -3,7 +3,7 @@ source: vortex-btrblocks/tests/golden.rs expression: rendered --- input: decimal(12,2), len=16384, nbytes=131072 -root: vortex.decimal_byte_parts(decimal(12,2), len=16384) nbytes=49152 +root: vortex.decimal_byte_parts.v2(decimal(12,2), len=16384) nbytes=49152 metadata: msp: fastlanes.bitpacked(i32, len=16384) nbytes=49152 metadata: bit_width: 24, offset: 0 diff --git a/vortex-btrblocks/tests/snapshots/golden__unstable__decimal_prices.snap b/vortex-btrblocks/tests/snapshots/golden__unstable__decimal_prices.snap index f669755e4b1..6eb4ccfed8d 100644 --- a/vortex-btrblocks/tests/snapshots/golden__unstable__decimal_prices.snap +++ b/vortex-btrblocks/tests/snapshots/golden__unstable__decimal_prices.snap @@ -3,7 +3,7 @@ source: vortex-btrblocks/tests/golden.rs expression: rendered --- input: decimal(12,2), len=16384, nbytes=131072 -root: vortex.decimal_byte_parts(decimal(12,2), len=16384) nbytes=49152 +root: vortex.decimal_byte_parts.v2(decimal(12,2), len=16384) nbytes=49152 metadata: msp: fastlanes.bitpacked(i32, len=16384) nbytes=49152 metadata: bit_width: 24, offset: 0 diff --git a/vortex-test/compat-gen/src/fixtures/arrays/synthetic/encodings/decimal_byte_parts_v2.rs b/vortex-test/compat-gen/src/fixtures/arrays/synthetic/encodings/decimal_byte_parts_v2.rs new file mode 100644 index 00000000000..166d84f5ac6 --- /dev/null +++ b/vortex-test/compat-gen/src/fixtures/arrays/synthetic/encodings/decimal_byte_parts_v2.rs @@ -0,0 +1,105 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! `DecimalByteParts` fixture for wide decimal values that need lower parts. + +use vortex::array::ArrayId; +use vortex::array::ArrayRef; +use vortex::array::ArrayVTable; +use vortex::array::IntoArray; +use vortex::array::arrays::DecimalArray; +use vortex::array::arrays::StructArray; +use vortex::array::dtype::DecimalDType; +use vortex::array::dtype::FieldNames; +use vortex::array::dtype::i256; +use vortex::array::validity::Validity; +use vortex::buffer::Buffer; +use vortex::encodings::decimal_byte_parts::DecimalByteParts; +use vortex::encodings::decimal_byte_parts::DecimalBytePartsArray; +use vortex::encodings::decimal_byte_parts::split_decimal; +use vortex::error::VortexResult; +use vortex_array::ExecutionCtx; + +use super::N; +use crate::fixtures::FlatLayoutFixture; + +/// Encode a canonical decimal as byte parts, splitting wide values into lower parts. +fn encode_byte_parts( + decimal: &DecimalArray, + ctx: &mut ExecutionCtx, +) -> VortexResult { + let parts = split_decimal(decimal, ctx)?; + DecimalByteParts::try_new_with_lower_parts( + parts.msp, + parts.lower_parts, + decimal.decimal_dtype(), + ) +} + +pub struct DecimalBytePartsV2Fixture; + +impl FlatLayoutFixture for DecimalBytePartsV2Fixture { + fn name(&self) -> &str { + "decimal_byte_parts_v2.vortex" + } + + fn description(&self) -> &str { + "Wide decimal arrays split into a most significant part plus 64-bit lower parts" + } + + fn expected_encodings(&self) -> Vec { + vec![DecimalByteParts.id()] + } + + fn build(&self, ctx: &mut ExecutionCtx) -> VortexResult { + // An `i128` magnitude above 2^64, so the encoding must carry one lower part. + let wide_128_dtype = DecimalDType::new(38, 2); + let wide_128 = DecimalArray::new( + (0..N as i128) + .map(|i| 10i128.pow(25) + i * 7) + .collect::>(), + wide_128_dtype, + Validity::NonNullable, + ); + let wide_128_arr = encode_byte_parts(&wide_128, ctx)?; + + // Negative values, so the sign extension above the MSP is exercised on read back. + let wide_128_negative = DecimalArray::new( + (0..N as i128) + .map(|i| -(10i128.pow(25)) - i * 7) + .collect::>(), + wide_128_dtype, + Validity::NonNullable, + ); + let wide_128_negative_arr = encode_byte_parts(&wide_128_negative, ctx)?; + + // An `i256` magnitude beyond 128 bits, so all three lower parts are populated, with + // nulls to pin that validity is carried by the MSP alone. + let wide_256_dtype = DecimalDType::new(76, 2); + let base = i256::from_i128(10).wrapping_pow(40); + let wide_256 = DecimalArray::new( + (0..N as i128) + .map(|i| base + i256::from_i128(i * 7)) + .collect::>(), + wide_256_dtype, + Validity::from_iter((0..N).map(|i| i % 7 != 0)), + ); + let wide_256_arr = encode_byte_parts(&wide_256, ctx)?; + + let arr = StructArray::try_new( + FieldNames::from([ + "dec_wide_128", + "dec_wide_128_negative", + "dec_wide_256_nullable", + ]), + vec![ + wide_128_arr.into_array(), + wide_128_negative_arr.into_array(), + wide_256_arr.into_array(), + ], + N, + Validity::NonNullable, + )?; + Ok(arr.into_array()) + } +} diff --git a/vortex-test/compat-gen/src/fixtures/arrays/synthetic/encodings/mod.rs b/vortex-test/compat-gen/src/fixtures/arrays/synthetic/encodings/mod.rs index 830b50450da..4d799e33e74 100644 --- a/vortex-test/compat-gen/src/fixtures/arrays/synthetic/encodings/mod.rs +++ b/vortex-test/compat-gen/src/fixtures/arrays/synthetic/encodings/mod.rs @@ -12,6 +12,7 @@ mod bytebool; mod constant; mod datetimeparts; mod decimal_byte_parts; +mod decimal_byte_parts_v2; mod delta; mod dict; mod for_; @@ -38,6 +39,7 @@ pub fn fixtures() -> Vec> { Box::new(bytebool::ByteBoolFixture), Box::new(datetimeparts::DateTimePartsFixture), Box::new(decimal_byte_parts::DecimalBytePartsFixture), + Box::new(decimal_byte_parts_v2::DecimalBytePartsV2Fixture), // Re-enable this once delta is stable // Box::new(delta::DeltaFixture), Box::new(dict::DictFixture),