Skip to content

Read legacy vortex.decimal nodes into the canonical array or a Widen #10995

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
connortsui20 merged 6 commits into from
Nov 5, 2026
Merged

Read legacy vortex.decimal nodes into the canonical array or a Widen #10995

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
8aeddf4
Read legacy vortex.decimal nodes into the canonical array or a Widen
connortsui20 Nov 5, 2026
797d187
Read legacy decimal fixtures as canonical arrays or Widens
connortsui20 Nov 5, 2026
5a19487
Move vortex.decimal serde from the Decimal vtable into DecimalPlugin
connortsui20 Nov 5, 2026
a865895
Build canonical decimal leaves with the recorded row count
connortsui20 Nov 5, 2026
9856e12
Reject legacy decimal nodes in device memory instead of copying them
connortsui20 Nov 5, 2026
0dcf448
Define DecimalMetadata above DecimalPlugin
connortsui20 Nov 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 14 additions & 5 deletions vortex-array/src/arrays/decimal/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,23 +1,32 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright the Vortex contributors

//! Decimal arrays whose logical precision determines their canonical integer width.
//!
//! [`DecimalPlugin`] reads and writes the `vortex.decimal` wire form in place of the [`Decimal`]
//! vtable. It reads a legacy node, stored at another width, as the canonical array or as a
//! [`Widen`].
//!
//! [`Widen`]: crate::arrays::Widen

mod array;
pub use array::DecimalArrayExt;
pub use array::DecimalArraySlotsExt;
pub use array::DecimalData;
pub use array::DecimalDataParts;
pub use array::DecimalSlots;
pub use vtable::DecimalArray;

pub(crate) mod compute;
pub use compute::rules::DecimalMaskedValidityRule;

mod vtable;
pub use compute::rules::DecimalMaskedValidityRule;
pub use vtable::Decimal;
pub use vtable::DecimalArray;
pub use vtable::DecimalPlugin;

mod utils;
pub use utils::*;

pub(crate) fn initialize(session: &vortex_session::VortexSession) {
vtable::initialize(session);
}

mod utils;
pub use utils::*;
291 changes: 33 additions & 258 deletions vortex-array/src/arrays/decimal/vtable/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,77 +1,56 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright the Vortex contributors

//! The [`Decimal`] vtable and the plugin that owns its wire form.
//!
//! The vtable does not serialize or deserialize. [`DecimalPlugin`] reads and writes the
//! `vortex.decimal` wire form, including legacy nodes.

mod kernel;
mod operations;

mod plugin;
pub use plugin::DecimalPlugin;

mod validity;

use std::hash::Hash;
use std::hash::Hasher;

use prost::Message;
use vortex_buffer::Alignment;
use vortex_error::VortexResult;
use vortex_error::vortex_bail;
use vortex_error::vortex_ensure;
use vortex_error::vortex_ensure_eq;
use vortex_error::vortex_err;
use vortex_session::VortexSession;
use vortex_session::registry::CachedId;

use crate::ArrayParts;
use crate::ArrayRef;
use crate::EqMode;
use crate::ExecutionCtx;
use crate::ExecutionResult;
use crate::array::Array;
use crate::array::ArrayId;
use crate::array::ArrayView;
use crate::array::VTable;
use crate::arrays::decimal::DecimalData;
use crate::arrays::decimal::array::DecimalSlots;
use crate::arrays::decimal::compute::rules::RULES;
use crate::arrays::fixed_width::vtable as fixed_width;
use crate::buffer::BufferHandle;
use crate::builders::ArrayBuilder;
use crate::builders::DecimalBuilder;
use crate::dtype::DType;
use crate::dtype::DecimalType;
use crate::dtype::NativeDecimalType;
use crate::match_each_decimal_value_type;
use crate::serde::ArrayChildren;
mod kernel;
mod operations;
mod validity;

use std::hash::Hash;

use vortex_session::registry::CachedId;

use crate::EqMode;
use crate::array::ArrayId;
use crate::arrays::decimal::array::DecimalSlots;
use crate::arrays::decimal::compute::rules::RULES;
use crate::hash::ArrayEq;
use crate::hash::ArrayHash;
use crate::serde::ArrayChildren;

/// A [`Decimal`]-encoded Vortex array.
pub type DecimalArray = Array<Decimal>;

pub(crate) fn initialize(session: &VortexSession) {
kernel::initialize(session);
}

/// Metadata for the existing `vortex.decimal` wire representation.
#[derive(prost::Message)]
pub struct DecimalMetadata {
/// Stored [`DecimalType`] discriminant, retained for wire compatibility.
///
/// The legacy format records the actual width separately from the logical dtype. Readers
/// need this field even when the in-memory representation uses canonical decimal leaves.
#[prost(int32, tag = "1")]
pub(super) values_type: i32,
}

impl DecimalMetadata {
fn get_values_type(&self) -> VortexResult<DecimalType> {
DecimalType::try_from(self.values_type).map_err(|_| {
vortex_err!(
"DecimalMetadata values_type must be a known DecimalType, got {}",
self.values_type
)
})
}
}

impl ArrayHash for DecimalData {
fn array_hash<H: Hasher>(&self, state: &mut H, accuracy: EqMode) {
self.values.array_hash(state, accuracy);
Expand Down Expand Up @@ -125,15 +104,10 @@ impl VTable for Decimal {
}

fn serialize(
array: ArrayView<'_, Self>,
_array: ArrayView<'_, Self>,
_session: &VortexSession,
) -> VortexResult<Option<Vec<u8>>> {
Ok(Some(
DecimalMetadata {
values_type: array.values_type() as i32,
}
.encode_to_vec(),
))
Ok(None)
}

fn validate(
Expand Down Expand Up @@ -166,40 +140,14 @@ impl VTable for Decimal {

fn deserialize(
&self,
dtype: &DType,
len: usize,
metadata: &[u8],
buffers: &[BufferHandle],
children: &dyn ArrayChildren,
_dtype: &DType,
_len: usize,
_metadata: &[u8],
_buffers: &[BufferHandle],
_children: &dyn ArrayChildren,
_session: &VortexSession,
) -> VortexResult<ArrayParts<Self>> {
let metadata = DecimalMetadata::decode(metadata)?;
let values_type = metadata.get_values_type()?;
let values = fixed_width::single_buffer(buffers)?;

let validity = fixed_width::deserialize_validity(dtype.nullability(), len, children)?;

let Some(decimal_dtype) = dtype.as_decimal_opt() else {
vortex_bail!("Expected Decimal dtype, got {:?}", dtype)
};

let slots = DecimalData::make_slots(&validity, len);
let data = match_each_decimal_value_type!(values_type, |D| {
// Check and reinterpret-cast the buffer
vortex_ensure!(
values.is_aligned_to(Alignment::of::<D>()),
"DecimalArray buffer not aligned for values type {:?}",
D::DECIMAL_TYPE
);
DecimalData::try_new_handle(values, values_type, *decimal_dtype)
})?;
Ok(ArrayParts::new(
self.clone(),
dtype.clone(),
len,
data,
slots,
))
vortex_bail!("Decimal deserialization requires DecimalPlugin")
}

fn slot_name(_array: ArrayView<'_, Self>, idx: usize) -> String {
Expand Down Expand Up @@ -230,182 +178,9 @@ impl VTable for Decimal {
}
}

/// The vtable of [`DecimalArray`].
///
/// This vtable does not serialize or deserialize. A session **must** register [`DecimalPlugin`]
/// in its place to read or write decimal arrays.
#[derive(Clone, Debug)]
pub struct Decimal;

#[cfg(test)]
mod tests {
use std::sync::LazyLock;

use prost::Message;
use rstest::rstest;
use vortex_buffer::ByteBuffer;
use vortex_buffer::ByteBufferMut;
use vortex_buffer::buffer;
use vortex_error::VortexResult;
use vortex_session::VortexSession;
use vortex_session::registry::ReadContext;

use super::DecimalMetadata;
use crate::ArrayContext;
use crate::ArrayRef;
use crate::IntoArray;
use crate::VortexSessionExecute;
use crate::array::VTable;
use crate::arrays::Decimal;
use crate::arrays::DecimalArray;
use crate::assert_arrays_eq;
use crate::buffer::BufferHandle;
use crate::dtype::DType;
use crate::dtype::DecimalDType;
use crate::dtype::DecimalType;
use crate::dtype::Nullability;
use crate::dtype::i256;
use crate::serde::SerializeOptions;
use crate::serde::SerializedArray;
use crate::validity::Validity;

static SESSION: LazyLock<VortexSession> = LazyLock::new(VortexSession::empty);

#[test]
fn test_array_serde() {
let array = DecimalArray::new(
buffer![100i128, 200i128, 300i128, 400i128, 500i128],
DecimalDType::new(10, 2),
Validity::NonNullable,
);
let dtype = array.dtype().clone();

let array_ctx = ArrayContext::empty();
let out = array
.into_array()
.serialize(&array_ctx, &SESSION, &SerializeOptions::default())
.unwrap();
// Concat into a single buffer
let mut concat = ByteBufferMut::empty();
for buf in out {
concat.extend_from_slice(buf.as_ref());
}

let concat = concat.freeze();

let parts = SerializedArray::try_from(concat).unwrap();
let decoded = parts
.decode(&dtype, 5, &ReadContext::new(array_ctx.to_ids()), &SESSION)
.unwrap();
assert!(decoded.is::<Decimal>());
assert_arrays_eq!(
decoded,
DecimalArray::from_iter([100i128, 200, 300, 400, 500], DecimalDType::new(10, 2)),
&mut SESSION.create_execution_ctx()
);
}

#[test]
fn test_nullable_decimal_serde_roundtrip() {
let mut ctx = SESSION.create_execution_ctx();
let array = DecimalArray::new(
buffer![1234567i32, 0i32, -9999999i32],
DecimalDType::new(7, 3),
Validity::from_iter([true, false, true]),
);
let dtype = array.dtype().clone();
let len = array.len();

let array_ctx = ArrayContext::empty();
let out = array
.clone()
.into_array()
.serialize(&array_ctx, &SESSION, &SerializeOptions::default())
.unwrap();
let mut concat = ByteBufferMut::empty();
for buf in out {
concat.extend_from_slice(buf.as_ref());
}

let parts = SerializedArray::try_from(concat.freeze()).unwrap();
let decoded = parts
.decode(&dtype, len, &ReadContext::new(array_ctx.to_ids()), &SESSION)
.unwrap();

assert_arrays_eq!(decoded, array, &mut ctx);
}

#[rstest]
#[case::first_unused(6)]
#[case::negative(-1)]
#[case::max(i32::MAX)]
fn test_deserialize_rejects_unknown_values_type(#[case] raw: i32) {
let metadata = DecimalMetadata { values_type: raw }.encode_to_vec();
let buffers = [BufferHandle::new_host(ByteBuffer::zeroed(4))];
let children: Vec<ArrayRef> = Vec::new();

let result = Decimal.deserialize(
&DType::Decimal(DecimalDType::new(2, 0), Nullability::NonNullable),
4,
&metadata,
&buffers,
&children,
&SESSION,
);

let expected =
format!("DecimalMetadata values_type must be a known DecimalType, got {raw}");
assert!(result.is_err_and(|err| err.to_string().contains(&expected)));
}

#[rstest]
#[case::i8(
DecimalArray::from_iter([1i8, -2, 3], DecimalDType::new(2, 0)),
DecimalType::I8
)]
#[case::i16(
DecimalArray::from_iter([1i16, -2, 3], DecimalDType::new(4, 0)),
DecimalType::I16
)]
#[case::i32(
DecimalArray::from_iter([1i32, -2, 3], DecimalDType::new(9, 2)),
DecimalType::I32
)]
#[case::i64(
DecimalArray::from_iter([1i64, -2, 3], DecimalDType::new(18, 2)),
DecimalType::I64
)]
#[case::i128(
DecimalArray::from_iter([1i128, -2, 3], DecimalDType::new(38, 2)),
DecimalType::I128
)]
#[case::i256(
DecimalArray::from_iter(
[i256::from_i128(1), i256::from_i128(-2), i256::from_i128(3)],
DecimalDType::new(76, 2),
),
DecimalType::I256
)]
fn test_serde_each_values_type(
#[case] array: DecimalArray,
#[case] values_type: DecimalType,
) -> VortexResult<()> {
let dtype = array.dtype().clone();
let len = array.len();

let array_ctx = ArrayContext::empty();
let out = array.clone().into_array().serialize(
&array_ctx,
&SESSION,
&SerializeOptions::default(),
)?;
let mut concat = ByteBufferMut::empty();
for buf in out {
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_::<Decimal>().values_type(), values_type);
assert_arrays_eq!(decoded, array, &mut SESSION.create_execution_ctx());

Ok(())
}
}
Loading
Loading