Execute OnPair iteratively (#10210)

`OnPair::execute` executes its compressed children inline. This change
requests its lengths, offsets, codes and dictionary offsets as primitive
arrays, and its validity as Bool, through the scheduler before decoding.

Widening the codes window reads an already-primitive child directly,
instead of building and executing a cast array over it.

Author: joseph-isaacs

encodings/onpair/src/array.rs

@@ -26,6 +26,7 @@ use vortex_array::EqMode;
 use vortex_array::ExecutionCtx;
 use vortex_array::ExecutionResult;
 use vortex_array::array_slots;
+use vortex_array::arrays::Primitive;
 use vortex_array::buffer::BufferHandle;
 use vortex_array::builders::ArrayBuilder;
 use vortex_array::builders::VarBinBuilder;
@@ -36,6 +37,8 @@ use vortex_array::dtype::OffsetBuilderPType;
 use vortex_array::dtype::PType;
 use vortex_array::match_each_integer_ptype;
 use vortex_array::match_each_varbin_builder;
+use vortex_array::require_child;
+use vortex_array::require_validity;
 use vortex_array::serde::ArrayChildren;
 use vortex_array::validity::Validity;
 use vortex_array::vtable::VTable;
@@ -601,6 +604,29 @@ impl VTable for OnPair {
     }
 
     fn execute(array: Array<Self>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        let array = require_child!(
+            array,
+            array.uncompressed_lengths(),
+            OnPairSlots::UNCOMPRESSED_LENGTHS => Primitive
+        );
+        let array = require_child!(
+            array,
+            array.codes_offsets(),
+            OnPairSlots::CODES_OFFSETS => Primitive
+        );
+        let array = require_child!(array, array.codes(), OnPairSlots::CODES => Primitive);
+        // The dictionary is built from its offsets once and memoised, so its child is only
+        // required while that has not happened.
+        let array = if array.data().dictionary.get().is_some() {
+            array
+        } else {
+            require_child!(
+                array,
+                array.dict_offsets(),
+                OnPairSlots::DICT_OFFSETS => Primitive
+            )
+        };
+        require_validity!(array, OnPairSlots::VALIDITY);
         canonicalize_onpair(array.as_view(), ctx).map(ExecutionResult::done)
     }
 

encodings/onpair/src/canonical.rs

@@ -32,7 +32,7 @@ use crate::OnPair;
 use crate::OnPairArraySlotsExt;
 use crate::array::dict_view;
 use crate::decode::code_boundary_at;
-use crate::decode::collect_widened;
+use crate::decode::collect_widened_range;
 
 pub(super) fn canonicalize_onpair(
     array: ArrayView<'_, OnPair>,
@@ -111,7 +111,7 @@ impl<'a> OnPairDecodePlan<'a> {
         // array materialises only its own codes rather than the whole column's. The
         // contiguous decoder walks `codes` in order and never reads the per-row
         // boundaries, so an empty boundary slice is sound.
-        let codes = collect_widened::<u16>(&array.codes().slice(code_start..code_end)?, ctx)?;
+        let codes = collect_widened_range::<u16>(array.codes(), code_start..code_end, ctx)?;
         let dict = dict_view(array, ctx)?;
 
         Ok(Self {

encodings/onpair/src/decode.rs

@@ -4,13 +4,17 @@
 //! Helpers for turning [`OnPair`] slot children into the inputs the upstream
 //! `onpair` decoder consumes.
 
+use std::ops::Range;
+
 use vortex_array::ArrayRef;
 use vortex_array::ArrayView;
 use vortex_array::ExecutionCtx;
+use vortex_array::arrays::Primitive;
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::builtins::ArrayBuiltins;
 use vortex_array::dtype::DType;
 use vortex_array::dtype::NativePType;
+use vortex_array::match_each_integer_ptype;
 use vortex_buffer::Buffer;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
@@ -26,8 +30,39 @@ pub(crate) fn collect_widened<T: NativePType>(
     arr: &ArrayRef,
     ctx: &mut ExecutionCtx,
 ) -> VortexResult<Buffer<T>> {
+    collect_widened_range(arr, 0..arr.len(), ctx)
+}
+
+/// Canonicalise `range` of a slot child to the decoder's native primitive width.
+///
+/// A child that `OnPair::execute` has already required to be primitive is widened in place,
+/// without a cast array or a nested execution; any other child is cast and executed.
+pub(crate) fn collect_widened_range<T: NativePType>(
+    arr: &ArrayRef,
+    range: Range<usize>,
+    ctx: &mut ExecutionCtx,
+) -> VortexResult<Buffer<T>> {
+    if let Some(primitive) = arr.as_opt::<Primitive>()
+        && !primitive.dtype().is_nullable()
+    {
+        if primitive.ptype() == T::PTYPE {
+            return Ok(primitive.into_owned().into_buffer::<T>().slice(range));
+        }
+        return match_each_integer_ptype!(primitive.ptype(), |P| {
+            primitive.as_slice::<P>()[range]
+                .iter()
+                .map(|&value| {
+                    T::from(value).ok_or_else(|| {
+                        vortex_err!("OnPair value {value} does not fit {}", T::PTYPE)
+                    })
+                })
+                .collect::<VortexResult<Vec<T>>>()
+                .map(Buffer::from)
+        });
+    }
     let dtype = DType::Primitive(T::PTYPE, arr.dtype().nullability());
     Ok(arr
+        .slice(range)?
         .cast(dtype)?
         .execute::<PrimitiveArray>(ctx)?
         .into_buffer::<T>())