Replace filter-and-scatter RowFn fallback with filtered valid-row execution

When a partially valid batch cannot execute directly over the original
inputs, batch execution previously filtered every input to the valid
rows, ran the dense kernel over the compact domain, and scattered the
compact output back with a nullable-index take.

Batch execution now dispatches a filtered valid-row execution instead:
it still filters the inputs (required when a representation cannot
decode null payloads), but the row loop reads consecutive compact rows
and writes each result directly at its original row index into a
full-length output, so the kernel output never needs a columnar
scatter. Owned outputs place default placeholders in skipped positions
and sinks run their skipped-row initializer, after which batch
execution masks the skipped rows exactly like direct skip-invalid
execution.

A sink without a skipped-row initializer can no longer execute a
partially valid batch, because every skip-invalid strategy now writes
into the original row domain. Both in-tree sinks already initialize
skipped rows.

Signed-off-by: Claude <noreply@anthropic.com>
Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0161JW9aU6W4zmzRQsnjbE3Q

Author: claude

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

@@ -33,7 +33,6 @@ 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,
@@ -43,6 +42,11 @@ impl RowFnExecutionArgs {
             MaskValuesRef,
             &mut ExecutionCtx,
         ) -> VortexResult<Option<ArrayRef>>,
+        execute_filtered_rows: impl FnOnce(
+            BorrowedRowFnArgs<'_>,
+            MaskValuesRef,
+            &mut ExecutionCtx,
+        ) -> VortexResult<ArrayRef>,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<ArrayRef> {
         let attempt =
@@ -51,20 +55,24 @@ impl RowFnExecutionArgs {
         match attempt {
             DenseAttempt::Values(values) => self.finalize_dense_output(values, ctx),
             DenseAttempt::DeferredError(error) => {
-                self.resolve_deferred_error(error, kernel, try_valid_rows, ctx)
+                self.resolve_deferred_error(error, try_valid_rows, execute_filtered_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,
             &mut ExecutionCtx,
         ) -> VortexResult<Option<ArrayRef>>,
+        execute_filtered_rows: impl FnOnce(
+            BorrowedRowFnArgs<'_>,
+            MaskValuesRef,
+            &mut ExecutionCtx,
+        ) -> VortexResult<ArrayRef>,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<ArrayRef> {
         let valid_rows = match self.validity.execute_mask(self.row_count, ctx)? {
@@ -85,7 +93,7 @@ impl RowFnExecutionArgs {
             return Ok(result);
         }
 
-        self.filter_and_scatter(kernel, &valid_rows, ctx)
+        self.execute_filtered(execute_filtered_rows, &valid_rows, ctx)
     }
 
     fn finalize_dense_output(

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

@@ -0,0 +1,62 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Filtered execution for row kernels that cannot operate on the original partially valid inputs.
+//!
+//! Direct skip-invalid execution retains the original input columns: it decodes every column,
+//! initializes a full-length output, and visits only valid rows. Some input representations cannot
+//! decode unspecified payloads behind nulls, so there is no safe row loop to enter over the
+//! original columns.
+//!
+//! This path filters every input to the valid row domain before decoding it. The kernel then
+//! iterates only the valid rows: it reads consecutive rows from the compact, all-valid inputs and
+//! writes each result directly at its original row index into a full-length output. Skipped
+//! positions hold placeholders that batch execution masks, exactly like direct skip-invalid
+//! execution, so the output never needs a columnar scatter.
+//!
+//! Filtering nested or compressed inputs can materialize their selected representation, so batch
+//! execution tries direct skip-invalid execution first and filters only when a required input
+//! capability is unavailable. Calling the general scalar-extraction API once per valid row is 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::builtins::ArrayBuiltins;
+
+impl RowFnExecutionArgs {
+    /// Filter the batch to valid rows, then execute the kernel into the original row domain.
+    pub(super) fn execute_filtered(
+        &self,
+        execute_filtered_rows: impl FnOnce(
+            BorrowedRowFnArgs<'_>,
+            MaskValuesRef,
+            &mut ExecutionCtx,
+        ) -> VortexResult<ArrayRef>,
+        valid: &MaskValuesRef,
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<ArrayRef> {
+        let filtered_len = valid.true_count();
+        let filter_mask = Mask::Values(MaskValuesRef::clone(valid));
+
+        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 values = execute_filtered_rows(filtered_args, MaskValuesRef::clone(valid), ctx)?;
+        let values = self.validate_kernel_output(values, self.row_count, ctx)?;
+
+        let mask = valid.as_ref().into_array();
+        self.finalize_output(values.mask(mask)?, self.row_count)
+    }
+}

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

@@ -21,7 +21,7 @@ use crate::validity::Validity;
 
 mod constant;
 mod dense;
-mod filter_scatter;
+mod filtered;
 mod valid_only;
 
 mod output;
@@ -31,7 +31,8 @@ impl RowFnExecutionArgs {
     /// Apply constant folding and null handling around `kernel`.
     ///
     /// 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
+    /// inputs to the valid rows and `execute_filtered_rows` runs the kernel over them while
+    /// writing into the original row domain. Every result is checked against the planned shape
     /// and dtype.
     pub(crate) fn execute(
         &self,
@@ -45,6 +46,11 @@ impl RowFnExecutionArgs {
             MaskValuesRef,
             &mut ExecutionCtx,
         ) -> VortexResult<Option<ArrayRef>>,
+        execute_filtered_rows: impl FnOnce(
+            BorrowedRowFnArgs<'_>,
+            MaskValuesRef,
+            &mut ExecutionCtx,
+        ) -> VortexResult<ArrayRef>,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<ArrayRef> {
         // Strictness: an all-null batch has no observable row work. Keep the literal-constant
@@ -77,10 +83,15 @@ impl RowFnExecutionArgs {
 
         match self.plan.policy() {
             RowPolicy::Dense => self.execute_dense(kernel, ctx),
-            RowPolicy::DenseWithRetry => {
-                self.execute_dense_with_retry(kernel, execute_dense_attempt, try_valid_rows, ctx)
+            RowPolicy::DenseWithRetry => self.execute_dense_with_retry(
+                execute_dense_attempt,
+                try_valid_rows,
+                execute_filtered_rows,
+                ctx,
+            ),
+            RowPolicy::ValidOnly => {
+                self.execute_valid_only(kernel, try_valid_rows, execute_filtered_rows, ctx)
             }
-            RowPolicy::ValidOnly => self.execute_valid_only(kernel, try_valid_rows, ctx),
         }
     }
 }

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

@@ -13,7 +13,7 @@ use crate::IntoArray;
 use crate::builtins::ArrayBuiltins;
 
 impl RowFnExecutionArgs {
-    /// Resolve validity and try direct valid-row execution before filter-and-scatter.
+    /// Resolve validity and try direct valid-row execution before filtered execution.
     pub(super) fn execute_valid_only(
         &self,
         kernel: impl Fn(BorrowedRowFnArgs<'_>, &mut ExecutionCtx) -> VortexResult<ArrayRef>,
@@ -22,6 +22,11 @@ impl RowFnExecutionArgs {
             MaskValuesRef,
             &mut ExecutionCtx,
         ) -> VortexResult<Option<ArrayRef>>,
+        execute_filtered_rows: impl FnOnce(
+            BorrowedRowFnArgs<'_>,
+            MaskValuesRef,
+            &mut ExecutionCtx,
+        ) -> VortexResult<ArrayRef>,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<ArrayRef> {
         let validity = self.validity.clone().execute_mask(self.row_count, ctx)?;
@@ -42,7 +47,7 @@ impl RowFnExecutionArgs {
             return Ok(result);
         }
 
-        self.filter_and_scatter(kernel, &valid_rows, ctx)
+        self.execute_filtered(execute_filtered_rows, &valid_rows, ctx)
     }
 
     /// Try execution against the original inputs, then mask a returned full-length result.