perf[sequence]: array constant compare execution kernel

CLAUDE.md

@@ -44,6 +44,7 @@
 * If you encounter clippy errors in tests that should only pertain to production code (e.g., prohibiting panic/unwrap,
   possible numerical truncation, etc.), then consider allowing those lints at the test module level.
 * Prefer naming test modules `tests`, not `test`.
+* Prefer having test return VortexResult<_> and use ? over unwrap.
 * Prefer module-scoped imports over function-scoped imports. Only use function-scoped imports in situations where it is
   either (a) required, or (b) would be exceptionally verbose otherwise. An example where function-scoped imports is good
   is when writing an exhaustive match statement with a branch that matches many cases.

encodings/sequence/Cargo.toml

@@ -22,6 +22,7 @@ vortex-dtype = { workspace = true }
 vortex-error = { workspace = true }
 vortex-mask = { workspace = true }
 vortex-proto = { workspace = true }
+vortex-runend = { workspace = true }
 vortex-scalar = { workspace = true }
 vortex-vector = { workspace = true }
 
@@ -32,6 +33,7 @@ tokio = { workspace = true, features = ["full"] }
 vortex-array = { path = "../../vortex-array", features = ["test-harness"] }
 vortex-file = { path = "../../vortex-file", features = ["tokio"] }
 vortex-layout = { path = "../../vortex-layout" }
+vortex-session = { workspace = true }
 
 [lints]
 workspace = true

encodings/sequence/src/array.rs

@@ -56,6 +56,8 @@ use vortex_vector::VectorMut;
 use vortex_vector::VectorMutOps;
 use vortex_vector::primitive::PVector;
 
+use crate::kernel::PARENT_KERNELS;
+
 vtable!(Sequence);
 
 #[derive(Clone, prost::Message)]
@@ -296,9 +298,14 @@ impl VTable for SequenceVTable {
     fn execute_parent(
         array: &Self::Array,
         parent: &ArrayRef,
-        _child_idx: usize,
-        _ctx: &mut ExecutionCtx,
+        child_idx: usize,
+        ctx: &mut ExecutionCtx,
     ) -> VortexResult<Option<Vector>> {
+        // Try parent kernels first (e.g., comparison fusion)
+        if let Some(result) = PARENT_KERNELS.execute(array, parent, child_idx, ctx)? {
+            return Ok(Some(result));
+        }
+
         // Special-case filtered execution.
         let Some(filter) = parent.as_opt::<FilterVTable>() else {
             return Ok(None);

encodings/sequence/src/compute/compare.rs

@@ -70,6 +70,12 @@ impl CompareKernel for SequenceVTable {
     }
 }
 
+/// Find the index where `base + idx * multiplier == intercept`, if one exists.
+///
+/// Returns `None` if:
+/// - `len` is 0
+/// - `intercept` is outside the range of the sequence
+/// - `intercept` doesn't fall exactly on a sequence value
 pub(crate) fn find_intersection_scalar(
     base: PValue,
     multiplier: PValue,
@@ -78,10 +84,8 @@ pub(crate) fn find_intersection_scalar(
 ) -> Option<usize> {
     match_each_integer_ptype!(base.ptype(), |P| {
         let intercept = intercept.cast::<P>();
-
         let base = base.cast::<P>();
         let multiplier = multiplier.cast::<P>();
-
         find_intersection(base, multiplier, len, intercept)
     })
 }
@@ -92,16 +96,38 @@ fn find_intersection<P: NativePType>(
     len: usize,
     intercept: P,
 ) -> Option<usize> {
-    // Array is non-empty here.
-    let count = <P>::from_usize(len - 1).vortex_expect("idx must fit into type");
+    if len == 0 {
+        return None;
+    }
 
+    let count = P::from_usize(len - 1).vortex_expect("idx must fit into type");
     let end_element = base + (multiplier * count);
 
-    (intercept.is_ge(base)
-        && intercept.is_le(end_element)
-        && (intercept - base) % multiplier == P::zero())
-    .then(|| ((intercept - base) / multiplier).to_usize())
-    .flatten()
+    // Handle ascending vs descending sequences
+    let (min_val, max_val) = if multiplier.is_ge(P::zero()) {
+        (base, end_element)
+    } else {
+        (end_element, base)
+    };
+
+    // Check if intercept is in range
+    if !intercept.is_ge(min_val) || !intercept.is_le(max_val) {
+        return None;
+    }
+
+    // Handle zero multiplier (constant sequence)
+    if multiplier == P::zero() {
+        return (intercept == base).then_some(0);
+    }
+
+    // Check if (intercept - base) is evenly divisible by multiplier
+    let diff = intercept - base;
+    if diff % multiplier != P::zero() {
+        return None;
+    }
+
+    let idx = diff / multiplier;
+    idx.to_usize()
 }
 
 #[cfg(test)]

encodings/sequence/src/compute/mod.rs

@@ -2,7 +2,7 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 mod cast;
-mod compare;
+pub(crate) mod compare;
 mod filter;
 mod is_sorted;
 mod list_contains;