[wip] DateTimeParts expansion rule

encodings/alp/src/alp/array.rs

@@ -15,7 +15,9 @@ use vortex_array::DeserializeMetadata;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
 use vortex_array::SerializeMetadata;
-use vortex_array::execution::ExecutionCtx;
+use vortex_array::kernel::BindCtx;
+use vortex_array::kernel::KernelRef;
+use vortex_array::kernel::kernel;
 use vortex_array::patches::Patches;
 use vortex_array::patches::PatchesMetadata;
 use vortex_array::serde::ArrayChildren;
@@ -41,7 +43,6 @@ use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_error::vortex_ensure;
-use vortex_vector::Vector;
 
 use crate::ALPFloat;
 use crate::alp::Exponents;
@@ -140,17 +141,16 @@ impl VTable for ALPVTable {
         )
     }
 
-    fn batch_execute(array: &ALPArray, ctx: &mut ExecutionCtx) -> VortexResult<Vector> {
-        let encoded_vector = array.encoded().batch_execute(ctx)?;
-
-        let patches_vectors = if let Some(patches) = array.patches() {
+    fn bind_kernel(array: &ALPArray, ctx: &mut BindCtx) -> VortexResult<KernelRef> {
+        let encoded = array.encoded().bind_kernel(ctx)?;
+        let patches_kernels = if let Some(patches) = array.patches() {
             Some((
-                patches.indices().batch_execute(ctx)?,
-                patches.values().batch_execute(ctx)?,
+                patches.indices().bind_kernel(ctx)?,
+                patches.values().bind_kernel(ctx)?,
                 patches
                     .chunk_offsets()
                     .as_ref()
-                    .map(|co| co.batch_execute(ctx))
+                    .map(|co| co.bind_kernel(ctx))
                     .transpose()?,
             ))
         } else {
@@ -161,7 +161,24 @@ impl VTable for ALPVTable {
         let exponents = array.exponents();
 
         match_each_alp_float_ptype!(array.dtype().as_ptype(), |T| {
-            decompress_into_vector::<T>(encoded_vector, exponents, patches_vectors, patches_offset)
+            Ok(kernel(move || {
+                let encoded_vector = encoded.execute()?;
+                let patches_vectors = match patches_kernels {
+                    Some((idx_kernel, val_kernel, co_kernel)) => Some((
+                        idx_kernel.execute()?,
+                        val_kernel.execute()?,
+                        co_kernel.map(|k| k.execute()).transpose()?,
+                    )),
+                    None => None,
+                };
+
+                decompress_into_vector::<T>(
+                    encoded_vector,
+                    exponents,
+                    patches_vectors,
+                    patches_offset,
+                )
+            }))
         })
     }
 }
@@ -456,15 +473,18 @@ mod tests {
     use std::sync::LazyLock;
 
     use rstest::rstest;
+    use vortex_array::VectorExecutor;
     use vortex_array::arrays::PrimitiveArray;
+    use vortex_array::session::ArraySession;
     use vortex_array::vtable::ValidityHelper;
     use vortex_dtype::PTypeDowncast;
     use vortex_session::VortexSession;
     use vortex_vector::VectorOps;
 
     use super::*;
 
-    static SESSION: LazyLock<VortexSession> = LazyLock::new(VortexSession::empty);
+    static SESSION: LazyLock<VortexSession> =
+        LazyLock::new(|| VortexSession::empty().with::<ArraySession>());
 
     #[rstest]
     #[case(0)]
@@ -480,7 +500,7 @@ mod tests {
         let values = PrimitiveArray::from_iter((0..size).map(|i| i as f32));
         let encoded = alp_encode(&values, None).unwrap();
 
-        let result_vector = encoded.to_array().execute(&SESSION).unwrap();
+        let result_vector = encoded.to_array().execute_vector(&SESSION).unwrap();
         // Compare against the traditional array-based decompress path
         let expected = decompress_into_array(encoded);
 
@@ -504,7 +524,7 @@ mod tests {
         let values = PrimitiveArray::from_iter((0..size).map(|i| i as f64));
         let encoded = alp_encode(&values, None).unwrap();
 
-        let result_vector = encoded.to_array().execute(&SESSION).unwrap();
+        let result_vector = encoded.to_array().execute_vector(&SESSION).unwrap();
         // Compare against the traditional array-based decompress path
         let expected = decompress_into_array(encoded);
 
@@ -534,7 +554,7 @@ mod tests {
         let encoded = alp_encode(&array, None).unwrap();
         assert!(encoded.patches().unwrap().array_len() > 0);
 
-        let result_vector = encoded.to_array().execute(&SESSION).unwrap();
+        let result_vector = encoded.to_array().execute_vector(&SESSION).unwrap();
         // Compare against the traditional array-based decompress path
         let expected = decompress_into_array(encoded);
 
@@ -562,7 +582,7 @@ mod tests {
         let array = PrimitiveArray::from_option_iter(values);
         let encoded = alp_encode(&array, None).unwrap();
 
-        let result_vector = encoded.to_array().execute(&SESSION).unwrap();
+        let result_vector = encoded.to_array().execute_vector(&SESSION).unwrap();
         // Compare against the traditional array-based decompress path
         let expected = decompress_into_array(encoded);
 
@@ -601,7 +621,7 @@ mod tests {
         let encoded = alp_encode(&array, None).unwrap();
         assert!(encoded.patches().unwrap().array_len() > 0);
 
-        let result_vector = encoded.to_array().execute(&SESSION).unwrap();
+        let result_vector = encoded.to_array().execute_vector(&SESSION).unwrap();
         // Compare against the traditional array-based decompress path
         let expected = decompress_into_array(encoded);
 
@@ -643,7 +663,7 @@ mod tests {
         let slice_len = slice_end - slice_start;
         let sliced_encoded = encoded.slice(slice_start..slice_end);
 
-        let result_vector = sliced_encoded.execute(&SESSION).unwrap();
+        let result_vector = sliced_encoded.execute_vector_optimized(&SESSION).unwrap();
         let result_primitive = result_vector.into_primitive().into_f64();
 
         for idx in 0..slice_len {

encodings/datetime-parts/Cargo.toml

@@ -25,6 +25,7 @@ vortex-dtype = { workspace = true }
 vortex-error = { workspace = true }
 vortex-mask = { workspace = true }
 vortex-scalar = { workspace = true }
+vortex-session = { workspace = true }
 
 [dev-dependencies]
 rstest = { workspace = true }

encodings/datetime-parts/src/lib.rs

@@ -3,18 +3,31 @@
 
 pub use array::*;
 pub use compress::*;
+use vortex_array::session::ArraySessionExt;
+use vortex_array::vtable::ArrayVTableExt;
+use vortex_session::VortexSession;
 
 mod array;
 mod canonical;
 mod compress;
 mod compute;
 mod ops;
+mod rules;
 mod timestamp;
 
+/// Initialize the DateTimeParts encoding in the given session.
+pub fn initialize(session: &mut VortexSession) {
+    session.arrays().register(DateTimePartsVTable.as_vtable());
+    // session
+    //     .arrays_mut()
+    //     .optimizer_mut()
+    //     .register_reduce_rule(DateTimePartsExpandRule);
+}
+
 #[cfg(test)]
 mod test {
-    use vortex_array::ProstMetadata;
     use vortex_array::test_harness::check_metadata;
+    use vortex_array::ProstMetadata;
     use vortex_dtype::PType;
 
     use crate::DateTimePartsMetadata;

encodings/datetime-parts/src/rules.rs

@@ -0,0 +1,68 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_array::arrays::ConstantArray;
+use vortex_array::builtins::ArrayBuiltins;
+use vortex_array::optimizer::rules::ArrayReduceRule;
+use vortex_array::optimizer::rules::Exact;
+use vortex_array::ArrayRef;
+use vortex_array::IntoArray;
+use vortex_dtype::datetime::TemporalMetadata;
+use vortex_dtype::datetime::TimeUnit;
+use vortex_dtype::DType;
+use vortex_dtype::PType;
+use vortex_error::vortex_panic;
+use vortex_error::VortexResult;
+
+use crate::DateTimePartsArray;
+use crate::DateTimePartsVTable;
+
+/// Expand a date-time-parts array into an expression that evaluates to the timestamp.
+#[derive(Debug)]
+pub(crate) struct DateTimePartsExpandRule;
+
+impl ArrayReduceRule<Exact<DateTimePartsVTable>> for DateTimePartsExpandRule {
+    fn matcher(&self) -> Exact<DateTimePartsVTable> {
+        Exact::from(&DateTimePartsVTable)
+    }
+
+    fn reduce(&self, array: &DateTimePartsArray) -> VortexResult<Option<ArrayRef>> {
+        let DType::Extension(ext) = array.dtype().clone() else {
+            vortex_panic!(ComputeError: "expected dtype to be DType::Extension variant")
+        };
+
+        let Ok(temporal_metadata) = TemporalMetadata::try_from(ext.as_ref()) else {
+            vortex_panic!(ComputeError: "must decode TemporalMetadata from extension metadata");
+        };
+
+        let divisor: i64 = match temporal_metadata.time_unit() {
+            TimeUnit::Nanoseconds => 1_000_000_000,
+            TimeUnit::Microseconds => 1_000_000,
+            TimeUnit::Milliseconds => 1_000,
+            TimeUnit::Seconds => 1,
+            TimeUnit::Days => vortex_panic!(InvalidArgument: "cannot decode into TimeUnit::D"),
+        };
+
+        // Up-cast days to i64 for computation.
+        let days = array
+            .days()
+            .cast(DType::Primitive(PType::I64, array.dtype().nullability()))?;
+
+        // Multiply days by the number of seconds in a day and the unit divisor.
+        let days = days.mul(ConstantArray::new(divisor * 86_400, array.len()).into_array())?;
+
+        // Multiply the seconds by the unit divisor.
+        let seconds = array
+            .seconds()
+            .cast(DType::Primitive(PType::I64, array.dtype().nullability()))?
+            .mul(ConstantArray::new(divisor, array.len()).into_array())?;
+
+        // The subseconds are already in the correct unit, just cast to i64.
+        let subseconds = array
+            .subseconds()
+            .cast(DType::Primitive(PType::I64, array.dtype().nullability()))?;
+
+        // Sum the three components together.
+        Ok(Some(days.add(seconds)?.add(subseconds)?))
+    }
+}

encodings/fastlanes/src/bitpacking/array/bitpack_decompress.rs

@@ -204,6 +204,7 @@ mod tests {
     use std::sync::LazyLock;
 
     use vortex_array::IntoArray;
+    use vortex_array::VectorExecutor;
     use vortex_array::assert_arrays_eq;
     use vortex_array::validity::Validity;
     use vortex_buffer::Buffer;
@@ -536,7 +537,7 @@ mod tests {
             let unpacked_array = unpack_array(&bitpacked);
 
             // Method 3: Using the execute() method (this is what would be used in production).
-            let executed = bitpacked.into_array().execute(&SESSION).unwrap();
+            let executed = bitpacked.into_array().execute_vector(&SESSION).unwrap();
 
             // All three should produce the same length.
             assert_eq!(vector_result.len(), array.len(), "vector length mismatch");
@@ -556,7 +557,10 @@ mod tests {
 
             // Verify that the execute() method works correctly by comparing with unpack_array.
             // We convert unpack_array result to a vector using execute() to compare.
-            let unpacked_executed = unpacked_array.into_array().execute(&SESSION).unwrap();
+            let unpacked_executed = unpacked_array
+                .into_array()
+                .execute_vector(&SESSION)
+                .unwrap();
             match (&executed, &unpacked_executed) {
                 (Vector::Primitive(exec_pv), Vector::Primitive(unpack_pv)) => {
                     assert_eq!(
@@ -593,7 +597,7 @@ mod tests {
         let sliced_bp = sliced.as_::<BitPackedVTable>();
         let vector_result = unpack_to_primitive_vector(sliced_bp);
         let unpacked_array = unpack_array(sliced_bp);
-        let executed = sliced.execute(&SESSION).unwrap();
+        let executed = sliced.execute_vector(&SESSION).unwrap();
 
         assert_eq!(
             vector_result.len(),

encodings/fastlanes/src/bitpacking/vtable/mod.rs

@@ -4,7 +4,9 @@
 use vortex_array::DeserializeMetadata;
 use vortex_array::ProstMetadata;
 use vortex_array::SerializeMetadata;
-use vortex_array::execution::ExecutionCtx;
+use vortex_array::kernel::BindCtx;
+use vortex_array::kernel::KernelRef;
+use vortex_array::kernel::kernel;
 use vortex_array::patches::Patches;
 use vortex_array::patches::PatchesMetadata;
 use vortex_array::serde::ArrayChildren;
@@ -23,7 +25,6 @@ use vortex_error::VortexError;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_error::vortex_err;
-use vortex_vector::Vector;
 use vortex_vector::VectorMutOps;
 
 use crate::BitPackedArray;
@@ -172,8 +173,11 @@ impl VTable for BitPackedVTable {
         )
     }
 
-    fn batch_execute(array: &BitPackedArray, _ctx: &mut ExecutionCtx) -> VortexResult<Vector> {
-        Ok(unpack_to_primitive_vector(array).freeze().into())
+    fn bind_kernel(array: &BitPackedArray, _ctx: &mut BindCtx) -> VortexResult<KernelRef> {
+        let array = array.clone();
+        Ok(kernel(move || {
+            Ok(unpack_to_primitive_vector(&array).freeze().into())
+        }))
     }
 }
 

encodings/runend/Cargo.toml

@@ -24,6 +24,7 @@ vortex-dtype = { workspace = true }
 vortex-error = { workspace = true }
 vortex-mask = { workspace = true }
 vortex-scalar = { workspace = true }
+vortex-session = { workspace = true }
 
 [lints]
 workspace = true