mod array;
mod canonical;
mod operations;
mod validity;
use kernel::PARENT_KERNELS;
use vortex_dtype::DType;
use vortex_error::VortexExpect;
use vortex_error::VortexResult;
use vortex_error::vortex_bail;
use vortex_error::vortex_ensure;
use vortex_session::VortexSession;
use crate::ArrayBufferVisitor;
use crate::ArrayChildVisitor;
use crate::ArrayRef;
use crate::Canonical;
use crate::EmptyMetadata;
use crate::IntoArray;
use crate::arrays::ConstantArray;
use crate::arrays::masked::MaskedArray;
use crate::arrays::masked::compute::rules::PARENT_RULES;
use crate::arrays::masked::mask_validity_canonical;
use crate::buffer::BufferHandle;
use crate::executor::ExecutionCtx;
use crate::scalar::Scalar;
use crate::serde::ArrayChildren;
use crate::validity::Validity;
use crate::vtable;
use crate::vtable::ArrayId;
use crate::vtable::VTable;
use crate::vtable::ValidityVTableFromValidityHelper;
use crate::vtable::VisitorVTable;
use crate::vtable::validity_nchildren;
use crate::vtable::validity_to_child;
mod kernel;
vtable!(Masked);
#[derive(Debug)]
pub struct MaskedVTable;
impl MaskedVTable {
pub const ID: ArrayId = ArrayId::new_ref("vortex.masked");
}
impl VisitorVTable<MaskedVTable> for MaskedVTable {
fn visit_buffers(_array: &MaskedArray, _visitor: &mut dyn ArrayBufferVisitor) {}
fn visit_children(array: &MaskedArray, visitor: &mut dyn ArrayChildVisitor) {
visitor.visit_child("child", &array.child);
visitor.visit_validity(&array.validity, array.child.len());
}
fn nchildren(array: &MaskedArray) -> usize {
1 + validity_nchildren(&array.validity)
}
fn nth_child(array: &MaskedArray, idx: usize) -> Option<ArrayRef> {
match idx {
0 => Some(array.child.clone()),
1 => validity_to_child(&array.validity, array.child.len()),
_ => None,
}
}
}
impl VTable for MaskedVTable {
type Array = MaskedArray;
type Metadata = EmptyMetadata;
type ArrayVTable = Self;
type OperationsVTable = Self;
type ValidityVTable = ValidityVTableFromValidityHelper;
type VisitorVTable = Self;
fn id(_array: &Self::Array) -> ArrayId {
Self::ID
}
fn metadata(_array: &MaskedArray) -> VortexResult<Self::Metadata> {
Ok(EmptyMetadata)
}
fn serialize(_metadata: Self::Metadata) -> VortexResult<Option<Vec<u8>>> {
Ok(Some(vec![]))
}
fn deserialize(
_bytes: &[u8],
_dtype: &DType,
_len: usize,
_buffers: &[BufferHandle],
_session: &VortexSession,
) -> VortexResult<Self::Metadata> {
Ok(EmptyMetadata)
}
fn build(
dtype: &DType,
len: usize,
_metadata: &Self::Metadata,
buffers: &[BufferHandle],
children: &dyn ArrayChildren,
) -> VortexResult<MaskedArray> {
if !buffers.is_empty() {
vortex_bail!("Expected 0 buffer, got {}", buffers.len());
}
let child = children.get(0, &dtype.as_nonnullable(), len)?;
let validity = if children.len() == 1 {
Validity::from(dtype.nullability())
} else if children.len() == 2 {
let validity = children.get(1, &Validity::DTYPE, len)?;
Validity::Array(validity)
} else {
vortex_bail!(
"`MaskedArray::build` expects 1 or 2 children, got {}",
children.len()
);
};
MaskedArray::try_new(child, validity)
}
fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
let validity_mask = array.validity_mask()?;
if validity_mask.all_false() {
return Ok(
ConstantArray::new(Scalar::null(array.dtype().as_nullable()), array.len())
.into_array(),
);
}
let child = array.child().clone().execute::<Canonical>(ctx)?;
Ok(mask_validity_canonical(child, &validity_mask, ctx)?.into_array())
}
fn reduce_parent(
array: &Self::Array,
parent: &ArrayRef,
child_idx: usize,
) -> VortexResult<Option<ArrayRef>> {
PARENT_RULES.evaluate(array, parent, child_idx)
}
fn execute_parent(
array: &Self::Array,
parent: &ArrayRef,
child_idx: usize,
ctx: &mut ExecutionCtx,
) -> VortexResult<Option<ArrayRef>> {
PARENT_KERNELS.execute(array, parent, child_idx, ctx)
}
fn with_children(array: &mut Self::Array, children: Vec<ArrayRef>) -> VortexResult<()> {
vortex_ensure!(
children.len() == 1 || children.len() == 2,
"MaskedArray expects 1 or 2 children, got {}",
children.len()
);
let mut iter = children.into_iter();
let child = iter
.next()
.vortex_expect("children length already validated");
let validity = if let Some(validity_array) = iter.next() {
Validity::Array(validity_array)
} else {
Validity::from(array.dtype.nullability())
};
let new_array = MaskedArray::try_new(child, validity)?;
*array = new_array;
Ok(())
}
}
#[cfg(test)]
mod tests {
use rstest::rstest;
use vortex_buffer::ByteBufferMut;
use vortex_dtype::Nullability;
use vortex_error::VortexError;
use crate::ArrayContext;
use crate::Canonical;
use crate::IntoArray;
use crate::LEGACY_SESSION;
use crate::VortexSessionExecute;
use crate::arrays::MaskedArray;
use crate::arrays::MaskedVTable;
use crate::arrays::PrimitiveArray;
use crate::serde::ArrayParts;
use crate::serde::SerializeOptions;
use crate::validity::Validity;
#[rstest]
#[case(
MaskedArray::try_new(
PrimitiveArray::from_iter([1i32, 2, 3]).into_array(),
Validity::AllValid
).unwrap()
)]
#[case(
MaskedArray::try_new(
PrimitiveArray::from_iter([1i32, 2, 3, 4, 5]).into_array(),
Validity::from_iter([true, true, false, true, false])
).unwrap()
)]
#[case(
MaskedArray::try_new(
PrimitiveArray::from_iter(0..100).into_array(),
Validity::from_iter((0..100).map(|i| i % 3 != 0))
).unwrap()
)]
fn test_serde_roundtrip(#[case] array: MaskedArray) {
let dtype = array.dtype().clone();
let len = array.len();
let ctx = ArrayContext::empty();
let serialized = array
.to_array()
.serialize(&ctx, &SerializeOptions::default())
.unwrap();
let mut concat = ByteBufferMut::empty();
for buf in serialized {
concat.extend_from_slice(buf.as_ref());
}
let concat = concat.freeze();
let parts = ArrayParts::try_from(concat).unwrap();
let decoded = parts.decode(&dtype, len, &ctx, &LEGACY_SESSION).unwrap();
assert!(decoded.is::<MaskedVTable>());
assert_eq!(
array.as_ref().display_values().to_string(),
decoded.display_values().to_string()
);
}
#[test]
fn test_execute_with_all_valid_preserves_nullable_dtype() -> Result<(), VortexError> {
let child = PrimitiveArray::from_iter([1i32, 2, 3]).into_array();
assert_eq!(child.dtype().nullability(), Nullability::NonNullable);
let array = MaskedArray::try_new(child, Validity::AllValid)?;
assert_eq!(array.dtype().nullability(), Nullability::Nullable);
let mut ctx = LEGACY_SESSION.create_execution_ctx();
let result: Canonical = array.into_array().execute(&mut ctx)?;
assert_eq!(
result.as_ref().dtype().nullability(),
Nullability::Nullable,
"MaskedArray execute should produce Nullable dtype"
);
Ok(())
}
}