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

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/filtered.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/mod.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/valid_only.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.

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

@@ -63,13 +63,13 @@ impl DeferredAdd {
 struct ValidOnlyIdentity;
 
 #[derive(Clone)]
-struct FilterAndScatterIdentity;
+struct FilteredIdentity;
 
 #[derive(Clone)]
 struct DenseRetryIncrement;
 
 #[derive(Clone)]
-struct FilterAndScatterRepeat;
+struct FilteredRepeat;
 
 #[derive(Clone)]
 struct InvalidKernelOutput;
@@ -212,7 +212,8 @@ impl OutputElement for NullProducingI64 {
 struct I64Sink(BufferMut<i64>);
 
 // SAFETY: every row is initialized by `BufferMut::zeroed`, and the sink exposes exactly that
-// initialized slice. The `()` write token therefore proves no additional invariant.
+// initialized slice. The `()` write token therefore proves no additional invariant, and the
+// default skipped-row initializer only rewrites the zeroes.
 unsafe impl OutputSink for I64Sink {
     type Params = ();
     type Rows<'a> = &'a mut [i64];
@@ -332,14 +333,14 @@ impl RowFn for ValidOnlyIdentity {
     }
 }
 
-impl RowFn for FilterAndScatterIdentity {
+impl RowFn for FilteredIdentity {
     type Options = EmptyOptions;
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = false;
 
     fn id(&self) -> ScalarFnId {
-        static ID: CachedId = CachedId::new("test.filter_and_scatter_identity");
+        static ID: CachedId = CachedId::new("test.filtered_identity");
         *ID
     }
 
@@ -383,14 +384,14 @@ impl RowFn for DenseRetryIncrement {
     }
 }
 
-impl RowFn for FilterAndScatterRepeat {
+impl RowFn for FilteredRepeat {
     type Options = usize;
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
-        static ID: CachedId = CachedId::new("test.filter_and_scatter_repeat");
+        static ID: CachedId = CachedId::new("test.filtered_repeat");
         *ID
     }
 
@@ -403,7 +404,7 @@ impl RowFn for FilterAndScatterRepeat {
         vortex_ensure!(
             u32::try_from(*width).is_ok(),
             InvalidArgument:
-            "test.filter_and_scatter_repeat width must fit in u32, got {width}",
+            "test.filtered_repeat width must fit in u32, got {width}",
         );
 
         visitor
@@ -720,14 +721,14 @@ fn test_valid_only_bool_output_skips_invalid_rows() -> VortexResult<()> {
 }
 
 #[test]
-fn test_filter_and_scatter_skips_invalid_decode_payloads() -> VortexResult<()> {
+fn test_filtered_execution_skips_invalid_decode_payloads() -> VortexResult<()> {
     let validity = Validity::from_iter([false, true, false, true]);
     let input =
         PrimitiveArray::new(vec![i64::MIN, 10, i64::MIN, 30], validity.clone()).into_array();
     let args = VecExecutionArgs::new(vec![input], 4);
     let mut ctx = array_session().create_execution_ctx();
 
-    let actual = execute_rows(&FilterAndScatterIdentity, &EmptyOptions, &args, &mut ctx)?;
+    let actual = execute_rows(&FilteredIdentity, &EmptyOptions, &args, &mut ctx)?;
     let expected = PrimitiveArray::new(vec![0_i64, 10, 0, 30], validity).into_array();
 
     assert_arrays_eq!(&actual, &expected, &mut ctx);
@@ -737,7 +738,7 @@ fn test_filter_and_scatter_skips_invalid_decode_payloads() -> VortexResult<()> {
 #[rstest]
 #[case::width_two(2, vec![0_i64, 0, 7, 7])]
 #[case::zero_width(0, vec![])]
-fn test_filter_and_scatter_preserves_runtime_sink_params(
+fn test_filtered_execution_preserves_runtime_sink_params(
     #[case] width: usize,
     #[case] expected_elements: Vec<i64>,
 ) -> VortexResult<()> {
@@ -746,7 +747,7 @@ fn test_filter_and_scatter_preserves_runtime_sink_params(
     let args = VecExecutionArgs::new(vec![input], 2);
     let mut ctx = array_session().create_execution_ctx();
 
-    let actual = execute_rows(&FilterAndScatterRepeat, &width, &args, &mut ctx)?;
+    let actual = execute_rows(&FilteredRepeat, &width, &args, &mut ctx)?;
     let expected = FixedSizeListArray::new(
         PrimitiveArray::from_iter(expected_elements).into_array(),
         u32::try_from(width)