Restrict bare decimal byte-parts serde to the frozen format

Reject lower parts when DecimalByteParts is registered directly. Share metadata
helpers with DecimalBytePartsPlugin so the plugin continues to support both
wire formats without parsing metadata twice. Add regression coverage for bare
registration and frozen-format compatibility.

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

Author: mhk197

encodings/decimal-byte-parts/src/decimal_byte_parts/mod.rs

@@ -84,6 +84,53 @@ pub struct DecimalBytesPartsMetadata {
 }
 
 impl DecimalBytesPartsMetadata {
+    fn from_array(array: ArrayView<'_, DecimalByteParts>) -> VortexResult<Self> {
+        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<ArrayParts<DecimalByteParts>> {
+        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
@@ -155,14 +202,12 @@ impl VTable for DecimalByteParts {
         array: ArrayView<'_, Self>,
         _session: &VortexSession,
     ) -> VortexResult<Option<Vec<u8>>> {
-        let lower_part_count = u32::try_from(array.lower_parts().len())
-            .map_err(|_| vortex_err!("lower part count exceeds u32"))?;
+        vortex_ensure!(
+            array.lower_parts().is_empty(),
+            "serializing DecimalByteParts with lower parts requires DecimalBytePartsPlugin"
+        );
         Ok(Some(
-            DecimalBytesPartsMetadata {
-                zeroth_child_ptype: PType::try_from(array.msp().dtype())? as i32,
-                lower_part_count,
-            }
-            .encode_to_vec(),
+            DecimalBytesPartsMetadata::from_array(array)?.encode_to_vec(),
         ))
     }
 
@@ -177,36 +222,10 @@ impl VTable for DecimalByteParts {
     ) -> VortexResult<ArrayParts<Self>> {
         let metadata = DecimalBytesPartsMetadata::decode(metadata)?;
         vortex_ensure!(
-            dtype.as_decimal_opt().is_some(),
-            "decoding decimal but given non decimal dtype {dtype}"
+            metadata.lower_part_count()? == 0,
+            "vortex.decimal_byte_parts must not carry lower parts"
         );
-
-        let encoded_dtype = DType::Primitive(metadata.zeroth_child_ptype(), dtype.nullability());
-
-        let lower_part_count = metadata.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(self.clone(), dtype.clone(), len, DecimalBytePartsData)
-                .with_slots(slots),
-        )
+        metadata.into_array_parts(dtype, len, children)
     }
 
     fn slot_name(_array: ArrayView<'_, Self>, idx: usize) -> String {
@@ -362,6 +381,9 @@ pub fn decimal_byte_parts_v2_id() -> ArrayId {
 /// serializes under [`decimal_byte_parts_v2_id`]. Reading holds each ID to its own contract:
 /// the frozen ID carries no lower parts and the v2 ID carries at least one, so recognizing the
 /// newer format never widens what the frozen one may mean.
+///
+/// Register this plugin, or call [`crate::initialize`], to enable both formats. Registering
+/// [`DecimalByteParts`] directly only supports the frozen format.
 #[derive(Clone, Debug)]
 pub struct DecimalBytePartsPlugin;
 
@@ -377,25 +399,33 @@ impl ArrayPlugin for DecimalBytePartsPlugin {
     fn serialize(
         &self,
         array: &ArrayRef,
-        session: &VortexSession,
+        _session: &VortexSession,
     ) -> VortexResult<Option<ArraySerialization>> {
-        let Some(mut serialization) = ArrayPlugin::serialize(&DecimalByteParts, array, session)?
-        else {
-            return Ok(None);
+        let view = array.as_opt::<DecimalByteParts>().ok_or_else(|| {
+            vortex_err!(
+                "DecimalByteParts plugin cannot serialize {}",
+                array.encoding_id()
+            )
+        })?;
+        let serialized_id = if view.lower_parts().is_empty() {
+            VTable::id(&DecimalByteParts)
+        } else {
+            decimal_byte_parts_v2_id()
         };
-        if !array.as_::<DecimalByteParts>().lower_parts().is_empty() {
-            serialization.serialized_id = decimal_byte_parts_v2_id();
-        }
-        Ok(Some(serialization))
+        Ok(Some(ArraySerialization::from_array(
+            serialized_id,
+            array,
+            DecimalBytesPartsMetadata::from_array(view)?.encode_to_vec(),
+        )))
     }
 
     fn deserialize(
         &self,
         parts: ArrayDeserialization<'_>,
-        session: &VortexSession,
+        _session: &VortexSession,
     ) -> VortexResult<ArrayRef> {
-        let lower_part_count =
-            DecimalBytesPartsMetadata::decode(parts.metadata)?.lower_part_count()?;
+        let metadata = DecimalBytesPartsMetadata::decode(parts.metadata)?;
+        let lower_part_count = metadata.lower_part_count()?;
         if parts.serialized_id == decimal_byte_parts_v2_id() {
             vortex_ensure!(
                 lower_part_count > 0,
@@ -414,14 +444,10 @@ impl ArrayPlugin for DecimalBytePartsPlugin {
                 parts.serialized_id
             );
         }
-        Ok(Array::try_from_parts(VTable::deserialize(
-            &DecimalByteParts,
+        Ok(Array::try_from_parts(metadata.into_array_parts(
             parts.dtype,
             parts.len,
-            parts.metadata,
-            parts.buffers,
             parts.children,
-            session,
         )?)?
         .into_array())
     }
@@ -897,26 +923,19 @@ mod tests {
     fn deserialize_with(
         lower_part_count: u32,
         children: Vec<ArrayRef>,
-    ) -> VortexResult<ArrayParts<DecimalByteParts>> {
-        let metadata = DecimalBytesPartsMetadata {
-            zeroth_child_ptype: PType::I64 as i32,
-            lower_part_count,
+    ) -> VortexResult<DecimalBytePartsArray> {
+        let serialized_id = if lower_part_count == 0 {
+            VTable::id(&DecimalByteParts)
+        } else {
+            decimal_byte_parts_v2_id()
         };
-        VTable::deserialize(
-            &DecimalByteParts,
-            &DType::Decimal(DecimalDType::new(38, 2), Nullability::NonNullable),
-            3,
-            &metadata.encode_to_vec(),
-            &[],
-            &children,
-            &array_session(),
-        )
+        plugin_deserialize_with(serialized_id, lower_part_count, children)
+            .map(|array| array.as_::<DecimalByteParts>().into_owned())
     }
 
     #[test]
     fn test_deserialize_reads_lower_parts() -> VortexResult<()> {
-        let parts = deserialize_with(1, vec![msp(), lower_part()])?;
-        let array = Array::try_from_parts(parts)?;
+        let array = deserialize_with(1, vec![msp(), lower_part()])?;
         assert_eq!(array.lower_parts().len(), 1);
 
         let mut ctx = array_session().create_execution_ctx();
@@ -936,9 +955,7 @@ mod tests {
         let session = array_session();
         crate::initialize(&session);
 
-        let array = deserialize_with(1, vec![msp(), lower_part()])
-            .and_then(Array::try_from_parts)?
-            .into_array();
+        let array = deserialize_with(1, vec![msp(), lower_part()])?.into_array();
 
         let serialization = session
             .array_serialize(&array)?
@@ -968,9 +985,7 @@ mod tests {
         let mut ctx = session.create_execution_ctx();
 
         // Stands in for an array materialized from a file: the parts already exist.
-        let array = deserialize_with(1, vec![msp(), lower_part()])
-            .and_then(Array::try_from_parts)?
-            .into_array();
+        let array = deserialize_with(1, vec![msp(), lower_part()])?.into_array();
 
         let sliced = array.slice(0..2)?;
         assert_eq!(sliced.execute::<DecimalArray>(&mut ctx)?.len(), 2);
@@ -984,7 +999,7 @@ mod tests {
         let mut children = vec![buffer![0i64; 3].into_array()];
         children.extend((1..lower_part_count).map(|_| buffer![0u64; 3].into_array()));
         children.push(lower_part());
-        let array = Array::try_from_parts(deserialize_with(lower_part_count, children)?)?;
+        let array = deserialize_with(lower_part_count, children)?;
         let expected = DecimalArray::new(
             buffer![1i128, 2, 3],
             DecimalDType::new(38, 2),

encodings/decimal-byte-parts/tests/format_v2.rs

@@ -14,10 +14,13 @@
 #![expect(clippy::tests_outside_test_module)]
 
 use vortex_array::ArrayContext;
+use vortex_array::ArrayDeserialization;
 use vortex_array::ArrayId;
 use vortex_array::ArrayRef;
 use vortex_array::ArrayVTable;
 use vortex_array::IntoArray;
+use vortex_array::VortexSessionExecute;
+use vortex_array::assert_arrays_eq;
 use vortex_array::dtype::DecimalDType;
 use vortex_array::serde::SerializeOptions;
 use vortex_array::session::ArraySessionExt;
@@ -163,3 +166,92 @@ fn wide_format_is_refused_where_not_permitted() -> VortexResult<()> {
 
     Ok(())
 }
+
+#[test]
+fn bare_vtable_refuses_wide_serialization() -> VortexResult<()> {
+    let session = vortex_array::array_session();
+    session.arrays().register(DecimalByteParts);
+    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(
+        [
+            ArrayVTable::id(&DecimalByteParts),
+            ArrayVTable::id(&vortex_array::arrays::Primitive),
+        ]
+        .into_iter()
+        .collect(),
+    );
+
+    assert!(
+        array
+            .serialize(&restricted, &session, &SerializeOptions::default())
+            .is_err(),
+        "bare VTable registration must not write lower parts under the frozen ID"
+    );
+    Ok(())
+}
+
+#[test]
+fn bare_vtable_refuses_lower_parts_on_frozen_id() -> VortexResult<()> {
+    let session = vortex_array::array_session();
+    session.arrays().register(DecimalByteParts);
+    let array = DecimalByteParts::try_new_with_lower_parts(
+        msp(),
+        vec![lower_part()],
+        DecimalDType::new(38, 2),
+    )?
+    .into_array();
+    let id = ArrayVTable::id(&DecimalByteParts);
+    let plugin = session
+        .arrays()
+        .registry()
+        .get(&id)
+        .ok_or_else(|| vortex_err!("missing decimal plugin"))?;
+    let children = array.children();
+
+    // i64 MSP and one lower part, mislabeled as the frozen format.
+    let parts = ArrayDeserialization::new(
+        id,
+        array.dtype(),
+        array.len(),
+        &[8, 7, 16, 1],
+        &[],
+        &children,
+    );
+    assert!(plugin.deserialize(parts, &session).is_err());
+    Ok(())
+}
+
+#[test]
+fn bare_vtable_keeps_frozen_serde() -> VortexResult<()> {
+    let session = vortex_array::array_session();
+    session.arrays().register(DecimalByteParts);
+    let array = DecimalByteParts::try_new(msp(), DecimalDType::new(2, 0))?.into_array();
+    let serialized = session
+        .array_serialize(&array)?
+        .ok_or_else(|| vortex_err!("missing decimal serialization"))?;
+    assert_eq!(serialized.serialized_id, ArrayVTable::id(&DecimalByteParts));
+    assert_eq!(serialized.metadata, [8, 7]);
+    let plugin = session
+        .arrays()
+        .registry()
+        .get(&serialized.serialized_id)
+        .ok_or_else(|| vortex_err!("missing decimal plugin"))?;
+    let decoded = plugin.deserialize(
+        ArrayDeserialization::new(
+            serialized.serialized_id,
+            array.dtype(),
+            array.len(),
+            &serialized.metadata,
+            &[],
+            &serialized.children,
+        ),
+        &session,
+    )?;
+    assert_arrays_eq!(array, decoded, &mut session.create_execution_ctx());
+    Ok(())
+}