Add filter-and-scatter RowFn execution

vortex-array/src/scalar_fn/unstable/row/batch/execute/dense.rs

@@ -3,7 +3,6 @@
 
 use vortex_error::VortexError;
 use vortex_error::VortexResult;
-use vortex_error::vortex_panic;
 use vortex_mask::Mask;
 use vortex_mask::MaskValuesRef;
 
@@ -34,6 +33,7 @@ impl RowFnExecutionArgs {
     /// recompute the error.
     pub(super) fn execute_dense_with_retry(
         &self,
+        kernel: impl Fn(BorrowedRowFnArgs<'_>, &mut ExecutionCtx) -> VortexResult<ArrayRef>,
         execute_dense_attempt: impl FnOnce(
             BorrowedRowFnArgs<'_>,
             &mut ExecutionCtx,
@@ -51,14 +51,15 @@ impl RowFnExecutionArgs {
         match attempt {
             DenseAttempt::Values(values) => self.finalize_dense_output(values, ctx),
             DenseAttempt::DeferredError(error) => {
-                self.resolve_deferred_error(error, try_valid_rows, ctx)
+                self.resolve_deferred_error(error, kernel, try_valid_rows, ctx)
             }
         }
     }
 
     fn resolve_deferred_error(
         &self,
         deferred_error: VortexError,
+        kernel: impl Fn(BorrowedRowFnArgs<'_>, &mut ExecutionCtx) -> VortexResult<ArrayRef>,
         try_valid_rows: impl FnOnce(
             BorrowedRowFnArgs<'_>,
             MaskValuesRef,
@@ -80,14 +81,11 @@ impl RowFnExecutionArgs {
         // retrying only observable rows.
         drop(deferred_error);
 
-        if let Some(result) = self.try_execute_valid_rows(try_valid_rows, valid_rows, ctx)? {
+        if let Some(result) = self.try_execute_valid_rows(try_valid_rows, &valid_rows, ctx)? {
             return Ok(result);
         }
 
-        vortex_panic!(
-            "dense retry requires direct valid-row support after a deferred error, but {} declined it",
-            self.id,
-        )
+        self.filter_and_scatter(kernel, &valid_rows, ctx)
     }
 
     fn finalize_dense_output(

vortex-array/src/scalar_fn/unstable/row/batch/execute/filter_scatter.rs

@@ -0,0 +1,95 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Fallback execution for row kernels that cannot operate on the original partially valid inputs.
+//!
+//! Direct skip-invalid execution retains the original row domain: it decodes every input column,
+//! initializes a full-length output, and visits only valid rows. Some input representations cannot
+//! decode unspecified payloads behind nulls, and some output sinks cannot initialize rows that the
+//! kernel skips. In either case, there is no safe row loop to enter over the original columns.
+//!
+//! This fallback filters every input to the valid row domain before decoding it. The ordinary
+//! kernel then produces a compact, all-valid output. A take with nullable indices scatters those
+//! values back to the original row domain and restores its nulls.
+//!
+//! Filtering and scattering add columnar work around the row loop. Filtering nested or compressed
+//! inputs can also materialize their selected representation. Batch execution therefore tries
+//! direct skip-invalid execution first and uses this path only when a required capability is
+//! unavailable.
+//!
+//! A prepared, type-specific view can provide selected-row decoding without filtering. That is an
+//! optional direct-execution capability rather than a replacement for this fallback: it requires a
+//! representation with safe selected access, and it does not initialize an output sink that cannot
+//! represent skipped rows. Calling the general scalar-extraction API once per valid row is also not
+//! equivalent because it repeats array execution and scalar construction inside the row loop.
+
+use smallvec::SmallVec;
+use vortex_error::VortexResult;
+use vortex_mask::Mask;
+use vortex_mask::MaskValuesRef;
+
+use super::super::RowFnExecutionArgs;
+use super::super::args::BorrowedRowFnArgs;
+use crate::ArrayRef;
+use crate::ExecutionCtx;
+use crate::IntoArray;
+use crate::arrays::PrimitiveArray;
+use crate::dtype::Nullability;
+use crate::validity::Validity;
+
+impl RowFnExecutionArgs {
+    /// Filter the original batch to valid rows, run the kernel, then restore its row count.
+    pub(super) fn filter_and_scatter(
+        &self,
+        kernel: impl Fn(BorrowedRowFnArgs<'_>, &mut ExecutionCtx) -> VortexResult<ArrayRef>,
+        original_validity: &MaskValuesRef,
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<ArrayRef> {
+        let original_len = original_validity.len();
+        let filtered_len = original_validity.true_count();
+        let filter_mask = Mask::Values(MaskValuesRef::clone(original_validity));
+
+        let filtered_inputs: SmallVec<[ArrayRef; 4]> = self
+            .inputs
+            .iter()
+            .map(|input| input.filter(filter_mask.clone()))
+            .collect::<VortexResult<_>>()?;
+
+        let filtered_args = self.execution_args(&filtered_inputs, filtered_len);
+        let filtered = kernel(filtered_args, ctx)?;
+        let filtered = self.validate_kernel_output(filtered, filtered_len, ctx)?;
+
+        let output = Self::scatter_to_original_rows(filtered, original_validity)?;
+
+        self.finalize_output(output, original_len)
+    }
+
+    /// Scatter `filtered` back to the rows selected by `original_validity`.
+    fn scatter_to_original_rows(
+        filtered: ArrayRef,
+        original_validity: &MaskValuesRef,
+    ) -> VortexResult<ArrayRef> {
+        let original_len = original_validity.len();
+        let mut take_indices = vec![0u64; original_len];
+
+        let valid_rows = original_validity
+            .slices()
+            .iter()
+            .flat_map(|&(start, end)| start..end);
+        for (filtered_index, original_index) in valid_rows.enumerate() {
+            take_indices[original_index] = u64::try_from(filtered_index)?;
+        }
+
+        // Null indices restore invalid rows without selecting a value from the compact output.
+        let take_indices = PrimitiveArray::new(
+            take_indices,
+            Validity::from_mask(
+                Mask::Values(MaskValuesRef::clone(original_validity)),
+                Nullability::Nullable,
+            ),
+        )
+        .into_array();
+
+        filtered.take(take_indices)
+    }
+}

vortex-array/src/scalar_fn/unstable/row/batch/execute/mod.rs

@@ -21,6 +21,7 @@ use crate::validity::Validity;
 
 mod constant;
 mod dense;
+mod filter_scatter;
 mod valid_only;
 
 mod output;
@@ -29,8 +30,9 @@ pub(crate) use output::finalize_kernel_output;
 impl RowFnExecutionArgs {
     /// Apply constant folding and null handling around `kernel`.
     ///
-    /// For a partially valid batch, `try_valid_rows` executes only valid rows over the original
-    /// inputs. Every result is checked against the planned shape and dtype.
+    /// For a partially valid batch, `try_valid_rows` can avoid filtering. `Ok(None)` filters the
+    /// valid rows and scatters the output back. Every result is checked against the planned shape
+    /// and dtype.
     pub(crate) fn execute(
         &self,
         kernel: impl Fn(BorrowedRowFnArgs<'_>, &mut ExecutionCtx) -> VortexResult<ArrayRef>,
@@ -76,7 +78,7 @@ impl RowFnExecutionArgs {
         match self.plan.policy() {
             RowPolicy::Dense => self.execute_dense(kernel, ctx),
             RowPolicy::DenseWithRetry => {
-                self.execute_dense_with_retry(execute_dense_attempt, try_valid_rows, ctx)
+                self.execute_dense_with_retry(kernel, execute_dense_attempt, try_valid_rows, ctx)
             }
             RowPolicy::ValidOnly => self.execute_valid_only(kernel, try_valid_rows, ctx),
         }

vortex-array/src/scalar_fn/unstable/row/batch/execute/valid_only.rs

@@ -2,7 +2,6 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use vortex_error::VortexResult;
-use vortex_error::vortex_panic;
 use vortex_mask::Mask;
 use vortex_mask::MaskValuesRef;
 
@@ -14,12 +13,7 @@ use crate::IntoArray;
 use crate::builtins::ArrayBuiltins;
 
 impl RowFnExecutionArgs {
-    /// Resolve validity, then execute valid rows over the original inputs.
-    ///
-    /// # Panics
-    ///
-    /// Panics if the concrete row signature cannot use direct valid-row execution. Inputs must
-    /// support null-tolerant decoding, and output sinks must initialize skipped rows.
+    /// Resolve validity and try direct valid-row execution before filter-and-scatter.
     pub(super) fn execute_valid_only(
         &self,
         kernel: impl Fn(BorrowedRowFnArgs<'_>, &mut ExecutionCtx) -> VortexResult<ArrayRef>,
@@ -44,14 +38,11 @@ impl RowFnExecutionArgs {
             Mask::Values(valid_rows) => valid_rows,
         };
 
-        if let Some(result) = self.try_execute_valid_rows(try_valid_rows, valid_rows, ctx)? {
+        if let Some(result) = self.try_execute_valid_rows(try_valid_rows, &valid_rows, ctx)? {
             return Ok(result);
         }
 
-        vortex_panic!(
-            "valid-only execution requires direct valid-row support; {} selected an unsupported signature",
-            self.id,
-        )
+        self.filter_and_scatter(kernel, &valid_rows, ctx)
     }
 
     /// Try execution against the original inputs, then mask a returned full-length result.
@@ -62,12 +53,12 @@ impl RowFnExecutionArgs {
             MaskValuesRef,
             &mut ExecutionCtx,
         ) -> VortexResult<Option<ArrayRef>>,
-        valid: MaskValuesRef,
+        valid: &MaskValuesRef,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<Option<ArrayRef>> {
         let Some(values) = try_valid_rows(
             self.execution_args(&self.inputs, self.row_count),
-            MaskValuesRef::clone(&valid),
+            MaskValuesRef::clone(valid),
             ctx,
         )?
         else {

vortex-array/src/scalar_fn/unstable/row/batch/mod.rs

@@ -39,7 +39,7 @@ pub(crate) struct RowFnExecutionArgs {
     /// The number of rows in the original execution scope.
     row_count: usize,
 
-    /// The input columns, collected once for validity, constant handling, and execution.
+    /// The input columns, collected once for validity, constant folding, filtering, and execution.
     inputs: SmallVec<[ArrayRef; 4]>,
 
     /// The input dtypes, collected with the columns and reused by both planning and execution.