feat[arary]: add a slice array to replace the slice op.

encodings/alp/src/alp/array.rs

@@ -3,6 +3,7 @@
 
 use std::fmt::Debug;
 use std::hash::Hash;
+use std::ops::Range;
 
 use vortex_array::Array;
 use vortex_array::ArrayBufferVisitor;
@@ -13,6 +14,7 @@ use vortex_array::ArrayRef;
 use vortex_array::Canonical;
 use vortex_array::DeserializeMetadata;
 use vortex_array::ExecutionCtx;
+use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
 use vortex_array::SerializeMetadata;
@@ -191,6 +193,17 @@ impl VTable for ALPVTable {
             ctx,
         )?))
     }
+
+    fn slice(array: &Self::Array, range: Range<usize>) -> VortexResult<Option<ArrayRef>> {
+        Ok(Some(
+            ALPArray::new(
+                array.encoded().slice(range.clone()),
+                array.exponents(),
+                array.patches().and_then(|p| p.slice(range)),
+            )
+            .into_array(),
+        ))
+    }
 }
 
 #[derive(Clone, Debug)]

encodings/alp/src/alp/ops.rs

@@ -1,11 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::ops::Range;
-
-use vortex_array::Array;
-use vortex_array::ArrayRef;
-use vortex_array::IntoArray;
 use vortex_array::vtable::OperationsVTable;
 use vortex_error::VortexExpect;
 use vortex_scalar::Scalar;
@@ -16,15 +11,6 @@ use crate::ALPVTable;
 use crate::match_each_alp_float_ptype;
 
 impl OperationsVTable<ALPVTable> for ALPVTable {
-    fn slice(array: &ALPArray, range: Range<usize>) -> ArrayRef {
-        ALPArray::new(
-            array.encoded().slice(range.clone()),
-            array.exponents(),
-            array.patches().and_then(|p| p.slice(range)),
-        )
-        .into_array()
-    }
-
     fn scalar_at(array: &ALPArray, index: usize) -> Scalar {
         if let Some(patches) = array.patches()
             && let Some(patch) = patches.get_patched(index)

encodings/alp/src/alp_rd/array.rs

@@ -13,6 +13,7 @@ use vortex_array::ArrayHash;
 use vortex_array::ArrayRef;
 use vortex_array::Canonical;
 use vortex_array::DeserializeMetadata;
+use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
 use vortex_array::SerializeMetadata;
@@ -83,6 +84,25 @@ impl VTable for ALPRDVTable {
         ArrayId::new_ref("vortex.alprd")
     }
 
+    fn slice(array: &Self::Array, range: std::ops::Range<usize>) -> VortexResult<Option<ArrayRef>> {
+        let left_parts_exceptions = array
+            .left_parts_patches()
+            .and_then(|patches| patches.slice(range.clone()));
+
+        // SAFETY: slicing components does not change the encoded values
+        Ok(Some(unsafe {
+            ALPRDArray::new_unchecked(
+                array.dtype().clone(),
+                array.left_parts().slice(range.clone()),
+                array.left_parts_dictionary().clone(),
+                array.right_parts().slice(range),
+                array.right_bit_width(),
+                left_parts_exceptions,
+            )
+            .into_array()
+        }))
+    }
+
     fn encoding(_array: &Self::Array) -> ArrayVTable {
         ALPRDVTable.as_vtable()
     }

encodings/alp/src/alp_rd/ops.rs

@@ -1,11 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::ops::Range;
-
 use vortex_array::Array;
-use vortex_array::ArrayRef;
-use vortex_array::IntoArray;
 use vortex_array::vtable::OperationsVTable;
 use vortex_error::VortexExpect;
 use vortex_scalar::Scalar;
@@ -14,25 +10,6 @@ use crate::ALPRDArray;
 use crate::ALPRDVTable;
 
 impl OperationsVTable<ALPRDVTable> for ALPRDVTable {
-    fn slice(array: &ALPRDArray, range: Range<usize>) -> ArrayRef {
-        let left_parts_exceptions = array
-            .left_parts_patches()
-            .and_then(|patches| patches.slice(range.clone()));
-
-        // SAFETY: slicing components does not change the encoded values
-        unsafe {
-            ALPRDArray::new_unchecked(
-                array.dtype().clone(),
-                array.left_parts().slice(range.clone()),
-                array.left_parts_dictionary().clone(),
-                array.right_parts().slice(range),
-                array.right_bit_width(),
-                left_parts_exceptions,
-            )
-            .into_array()
-        }
-    }
-
     fn scalar_at(array: &ALPRDArray, index: usize) -> Scalar {
         // The left value can either be a direct value, or an exception.
         // The exceptions array represents exception positions with non-null values.

encodings/bytebool/src/array.rs

@@ -117,6 +117,16 @@ impl VTable for ByteBoolVTable {
 
         Ok(())
     }
+
+    fn slice(array: &ByteBoolArray, range: Range<usize>) -> VortexResult<Option<ArrayRef>> {
+        Ok(Some(
+            ByteBoolArray::new(
+                array.buffer().slice(range.clone()),
+                array.validity().slice(range),
+            )
+            .into_array(),
+        ))
+    }
 }
 
 #[derive(Clone, Debug)]
@@ -213,14 +223,6 @@ impl CanonicalVTable<ByteBoolVTable> for ByteBoolVTable {
 }
 
 impl OperationsVTable<ByteBoolVTable> for ByteBoolVTable {
-    fn slice(array: &ByteBoolArray, range: Range<usize>) -> ArrayRef {
-        ByteBoolArray::new(
-            array.buffer().slice(range.clone()),
-            array.validity().slice(range),
-        )
-        .into_array()
-    }
-
     fn scalar_at(array: &ByteBoolArray, index: usize) -> Scalar {
         Scalar::bool(array.buffer()[index] == 1, array.dtype().nullability())
     }

encodings/datetime-parts/src/array.rs

@@ -3,6 +3,7 @@
 
 use std::fmt::Debug;
 use std::hash::Hash;
+use std::ops::Range;
 
 use vortex_array::Array;
 use vortex_array::ArrayBufferVisitor;
@@ -12,6 +13,7 @@ use vortex_array::ArrayHash;
 use vortex_array::ArrayRef;
 use vortex_array::Canonical;
 use vortex_array::DeserializeMetadata;
+use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
 use vortex_array::SerializeMetadata;
@@ -169,6 +171,19 @@ impl VTable for DateTimePartsVTable {
     ) -> VortexResult<Option<ArrayRef>> {
         PARENT_RULES.evaluate(array, parent, child_idx)
     }
+
+    fn slice(array: &Self::Array, range: Range<usize>) -> VortexResult<Option<ArrayRef>> {
+        // SAFETY: slicing all components preserves values
+        Ok(Some(unsafe {
+            DateTimePartsArray::new_unchecked(
+                array.dtype().clone(),
+                array.days().slice(range.clone()),
+                array.seconds().slice(range.clone()),
+                array.subseconds().slice(range),
+            )
+            .into_array()
+        }))
+    }
 }
 
 #[derive(Clone, Debug)]

encodings/datetime-parts/src/ops.rs

@@ -1,11 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::ops::Range;
-
 use vortex_array::Array;
-use vortex_array::ArrayRef;
-use vortex_array::IntoArray;
 use vortex_array::vtable::OperationsVTable;
 use vortex_dtype::DType;
 use vortex_dtype::datetime::TemporalMetadata;
@@ -19,19 +15,6 @@ use crate::timestamp;
 use crate::timestamp::TimestampParts;
 
 impl OperationsVTable<DateTimePartsVTable> for DateTimePartsVTable {
-    fn slice(array: &DateTimePartsArray, range: Range<usize>) -> ArrayRef {
-        // SAFETY: slicing all components preserves values
-        unsafe {
-            DateTimePartsArray::new_unchecked(
-                array.dtype().clone(),
-                array.days().slice(range.clone()),
-                array.seconds().slice(range.clone()),
-                array.subseconds().slice(range),
-            )
-            .into_array()
-        }
-    }
-
     fn scalar_at(array: &DateTimePartsArray, index: usize) -> Scalar {
         let DType::Extension(ext) = array.dtype().clone() else {
             vortex_panic!(

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

@@ -138,6 +138,14 @@ impl VTable for DecimalBytePartsVTable {
     ) -> VortexResult<Option<ArrayRef>> {
         PARENT_RULES.evaluate(array, parent, child_idx)
     }
+
+    fn slice(array: &Self::Array, range: Range<usize>) -> VortexResult<Option<ArrayRef>> {
+        // SAFETY: slicing encoded MSP does not change the encoded values
+        Ok(Some(unsafe {
+            DecimalBytePartsArray::new_unchecked(array.msp.slice(range), *array.decimal_dtype())
+                .into_array()
+        }))
+    }
 }
 
 /// This array encodes decimals as between 1-4 columns of primitive typed children.
@@ -248,14 +256,6 @@ impl CanonicalVTable<DecimalBytePartsVTable> for DecimalBytePartsVTable {
 }
 
 impl OperationsVTable<DecimalBytePartsVTable> for DecimalBytePartsVTable {
-    fn slice(array: &DecimalBytePartsArray, range: Range<usize>) -> ArrayRef {
-        // SAFETY: slicing encoded MSP does not change the encoded values
-        unsafe {
-            DecimalBytePartsArray::new_unchecked(array.msp.slice(range), *array.decimal_dtype())
-                .into_array()
-        }
-    }
-
     fn scalar_at(array: &DecimalBytePartsArray, index: usize) -> Scalar {
         // TODO(joe): support parts len != 1
         let scalar = array.msp.scalar_at(index);

encodings/fastlanes/src/bitpacking/mod.rs

@@ -8,6 +8,7 @@ pub use array::bitpack_decompress;
 pub use array::unpack_iter;
 
 mod compute;
+mod rules;
 
 mod vtable;
 pub use vtable::BitPackedVTable;

encodings/fastlanes/src/bitpacking/rules.rs

@@ -0,0 +1,8 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_array::optimizer::rules::ParentRuleSet;
+
+use crate::BitPackedVTable;
+
+pub(super) const RULES: ParentRuleSet<BitPackedVTable> = ParentRuleSet::new(&[]);

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

@@ -1,10 +1,14 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+use std::cmp::max;
+use std::ops::Range;
+
 use vortex_array::ArrayRef;
 use vortex_array::Canonical;
 use vortex_array::DeserializeMetadata;
 use vortex_array::ExecutionCtx;
+use vortex_array::IntoArray;
 use vortex_array::ProstMetadata;
 use vortex_array::SerializeMetadata;
 use vortex_array::buffer::BufferHandle;
@@ -18,6 +22,7 @@ use vortex_array::vtable::ArrayVTable;
 use vortex_array::vtable::ArrayVTableExt;
 use vortex_array::vtable::NotSupported;
 use vortex_array::vtable::VTable;
+use vortex_array::vtable::ValidityHelper;
 use vortex_array::vtable::ValidityVTableFromValidityHelper;
 use vortex_dtype::DType;
 use vortex_dtype::PType;
@@ -28,6 +33,7 @@ use vortex_error::vortex_ensure;
 use vortex_error::vortex_err;
 
 use crate::BitPackedArray;
+use crate::bitpacking::rules::RULES;
 use crate::bitpacking::vtable::kernels::filter::PARENT_KERNELS;
 
 mod array;
@@ -251,6 +257,40 @@ impl VTable for BitPackedVTable {
     ) -> VortexResult<Option<Canonical>> {
         PARENT_KERNELS.execute(array, parent, child_idx, ctx)
     }
+
+    fn reduce_parent(
+        array: &Self::Array,
+        parent: &ArrayRef,
+        child_idx: usize,
+    ) -> VortexResult<Option<ArrayRef>> {
+        RULES.evaluate(array, parent, child_idx)
+    }
+
+    fn slice(array: &Self::Array, range: Range<usize>) -> VortexResult<Option<ArrayRef>> {
+        let offset_start = range.start + array.offset() as usize;
+        let offset_stop = range.end + array.offset() as usize;
+        let offset = offset_start % 1024;
+        let block_start = max(0, offset_start - offset);
+        let block_stop = offset_stop.div_ceil(1024) * 1024;
+
+        let encoded_start = (block_start / 8) * array.bit_width() as usize;
+        let encoded_stop = (block_stop / 8) * array.bit_width() as usize;
+
+        // slice the buffer using the encoded start/stop values
+        // SAFETY: slicing packed values without decoding preserves invariants
+        Ok(Some(unsafe {
+            BitPackedArray::new_unchecked(
+                array.packed().slice(encoded_start..encoded_stop),
+                array.dtype.clone(),
+                array.validity().slice(range.clone()),
+                array.patches().and_then(|p| p.slice(range.clone())),
+                array.bit_width(),
+                range.len(),
+                offset as u16,
+            )
+            .into_array()
+        }))
+    }
 }
 
 #[derive(Debug)]

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

@@ -1,46 +1,14 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::cmp::max;
-use std::ops::Range;
-
-use vortex_array::ArrayRef;
-use vortex_array::IntoArray;
 use vortex_array::vtable::OperationsVTable;
-use vortex_array::vtable::ValidityHelper;
 use vortex_scalar::Scalar;
 
 use crate::BitPackedArray;
 use crate::BitPackedVTable;
 use crate::bitpack_decompress;
 
 impl OperationsVTable<BitPackedVTable> for BitPackedVTable {
-    fn slice(array: &BitPackedArray, range: Range<usize>) -> ArrayRef {
-        let offset_start = range.start + array.offset() as usize;
-        let offset_stop = range.end + array.offset() as usize;
-        let offset = offset_start % 1024;
-        let block_start = max(0, offset_start - offset);
-        let block_stop = offset_stop.div_ceil(1024) * 1024;
-
-        let encoded_start = (block_start / 8) * array.bit_width() as usize;
-        let encoded_stop = (block_stop / 8) * array.bit_width() as usize;
-
-        // slice the buffer using the encoded start/stop values
-        // SAFETY: slicing packed values without decoding preserves invariants
-        unsafe {
-            BitPackedArray::new_unchecked(
-                array.packed().slice(encoded_start..encoded_stop),
-                array.dtype.clone(),
-                array.validity().slice(range.clone()),
-                array.patches().and_then(|p| p.slice(range.clone())),
-                array.bit_width(),
-                range.len(),
-                offset as u16,
-            )
-            .into_array()
-        }
-    }
-
     fn scalar_at(array: &BitPackedArray, index: usize) -> Scalar {
         if let Some(patches) = array.patches()
             && let Some(patch) = patches.get_patched(index)