Merge remote-tracking branch 'origin/develop' into ji/vector-eq

# Conflicts:
#	vortex-vector/src/binaryview/types.rs
#	vortex-vector/src/listview/scalar.rs
#	vortex-vector/src/struct_/scalar.rs

Author: joseph-isaacs

Cargo.lock

@@ -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,7 @@ use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_error::vortex_ensure;
-use vortex_vector::Vector;
+use vortex_error::vortex_err;
 
 use crate::ALPFloat;
 use crate::alp::Exponents;
@@ -140,17 +142,61 @@ impl VTable for ALPVTable {
         )
     }
 
-    fn batch_execute(array: &ALPArray, ctx: &mut ExecutionCtx) -> VortexResult<Vector> {
-        let encoded_vector = array.encoded().batch_execute(ctx)?;
+    fn with_children(array: &mut Self::Array, children: Vec<ArrayRef>) -> VortexResult<()> {
+        // Children: encoded, patches (if present): indices, values, chunk_offsets (optional)
+        let patches_info = array
+            .patches
+            .as_ref()
+            .map(|p| (p.array_len(), p.offset(), p.chunk_offsets().is_some()));
+
+        let expected_children = match &patches_info {
+            Some((_, _, has_chunk_offsets)) => 1 + 2 + if *has_chunk_offsets { 1 } else { 0 },
+            None => 1,
+        };
+
+        vortex_ensure!(
+            children.len() == expected_children,
+            "ALPArray expects {} children, got {}",
+            expected_children,
+            children.len()
+        );
+
+        let mut children_iter = children.into_iter();
+        array.encoded = children_iter
+            .next()
+            .ok_or_else(|| vortex_err!("Expected encoded child"))?;
+
+        if let Some((array_len, offset, _has_chunk_offsets)) = patches_info {
+            let indices = children_iter
+                .next()
+                .ok_or_else(|| vortex_err!("Expected patch indices child"))?;
+            let values = children_iter
+                .next()
+                .ok_or_else(|| vortex_err!("Expected patch values child"))?;
+            let chunk_offsets = children_iter.next();
+
+            array.patches = Some(Patches::new(
+                array_len,
+                offset,
+                indices,
+                values,
+                chunk_offsets,
+            ));
+        }
+
+        Ok(())
+    }
 
-        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 +207,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 +519,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 +546,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 +570,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 +600,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 +628,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 +667,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 +709,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/alp/src/alp/array.rs

@@ -45,6 +45,7 @@ use vortex_error::VortexError;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
+use vortex_error::vortex_ensure;
 use vortex_error::vortex_err;
 
 use crate::alp_rd::alp_rd_decode;
@@ -185,6 +186,47 @@ impl VTable for ALPRDVTable {
             left_parts_patches,
         )
     }
+
+    fn with_children(array: &mut Self::Array, children: Vec<ArrayRef>) -> VortexResult<()> {
+        // Children: left_parts, right_parts, patches (if present): indices, values
+        let patches_info = array
+            .left_parts_patches
+            .as_ref()
+            .map(|p| (p.array_len(), p.offset()));
+
+        let expected_children = if patches_info.is_some() { 4 } else { 2 };
+
+        vortex_ensure!(
+            children.len() == expected_children,
+            "ALPRDArray expects {} children, got {}",
+            expected_children,
+            children.len()
+        );
+
+        let mut children_iter = children.into_iter();
+        array.left_parts = children_iter
+            .next()
+            .ok_or_else(|| vortex_err!("Expected left_parts child"))?;
+        array.right_parts = children_iter
+            .next()
+            .ok_or_else(|| vortex_err!("Expected right_parts child"))?;
+
+        if let Some((array_len, offset)) = patches_info {
+            let indices = children_iter
+                .next()
+                .ok_or_else(|| vortex_err!("Expected patch indices child"))?;
+            let values = children_iter
+                .next()
+                .ok_or_else(|| vortex_err!("Expected patch values child"))?;
+
+            array.left_parts_patches = Some(Patches::new(
+                array_len, offset, indices, values,
+                None, // chunk_offsets not currently supported for ALPRD
+            ));
+        }
+
+        Ok(())
+    }
 }
 
 #[derive(Clone, Debug)]

encodings/alp/src/alp_rd/array.rs

@@ -35,8 +35,10 @@ use vortex_buffer::BitBuffer;
 use vortex_buffer::BufferHandle;
 use vortex_buffer::ByteBuffer;
 use vortex_dtype::DType;
+use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
+use vortex_error::vortex_ensure;
 use vortex_error::vortex_panic;
 use vortex_scalar::Scalar;
 
@@ -99,6 +101,22 @@ impl VTable for ByteBoolVTable {
 
         Ok(ByteBoolArray::new(buffer, validity))
     }
+
+    fn with_children(array: &mut Self::Array, children: Vec<ArrayRef>) -> VortexResult<()> {
+        vortex_ensure!(
+            children.len() <= 1,
+            "ByteBoolArray expects at most 1 child (validity), got {}",
+            children.len()
+        );
+
+        array.validity = if children.is_empty() {
+            Validity::from(array.dtype.nullability())
+        } else {
+            Validity::Array(children.into_iter().next().vortex_expect("checked"))
+        };
+
+        Ok(())
+    }
 }
 
 #[derive(Clone, Debug)]

encodings/bytebool/src/array.rs

@@ -34,8 +34,10 @@ use vortex_buffer::BufferHandle;
 use vortex_dtype::DType;
 use vortex_dtype::Nullability;
 use vortex_dtype::PType;
+use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
+use vortex_error::vortex_ensure;
 use vortex_error::vortex_err;
 
 vtable!(DateTimeParts);
@@ -142,6 +144,21 @@ impl VTable for DateTimePartsVTable {
 
         DateTimePartsArray::try_new(dtype.clone(), days, seconds, subseconds)
     }
+
+    fn with_children(array: &mut Self::Array, children: Vec<ArrayRef>) -> VortexResult<()> {
+        vortex_ensure!(
+            children.len() == 3,
+            "DateTimePartsArray expects exactly 3 children (days, seconds, subseconds), got {}",
+            children.len()
+        );
+
+        let mut children_iter = children.into_iter();
+        array.days = children_iter.next().vortex_expect("checked");
+        array.seconds = children_iter.next().vortex_expect("checked");
+        array.subseconds = children_iter.next().vortex_expect("checked");
+
+        Ok(())
+    }
 }
 
 #[derive(Clone, Debug)]

encodings/datetime-parts/src/array.rs

@@ -44,6 +44,7 @@ use vortex_dtype::match_each_signed_integer_ptype;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
+use vortex_error::vortex_ensure;
 use vortex_scalar::DecimalValue;
 use vortex_scalar::Scalar;
 
@@ -116,6 +117,16 @@ impl VTable for DecimalBytePartsVTable {
 
         DecimalBytePartsArray::try_new(msp, *decimal_dtype)
     }
+
+    fn with_children(array: &mut Self::Array, children: Vec<ArrayRef>) -> VortexResult<()> {
+        vortex_ensure!(
+            children.len() == 1,
+            "DecimalBytePartsArray expects exactly 1 child (msp), got {}",
+            children.len()
+        );
+        array.msp = children.into_iter().next().vortex_expect("checked");
+        Ok(())
+    }
 }
 
 /// This array encodes decimals as between 1-4 columns of primitive typed children.

encodings/decimal-byte-parts/src/decimal_byte_parts/mod.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(),