Implement the RowFn execution backend

vortex-array/src/arrays/masked/tests.rs

@@ -4,17 +4,22 @@
 use rstest::rstest;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
 
 use super::*;
 use crate::Canonical;
 use crate::IntoArray;
 use crate::VortexSessionExecute;
+use crate::array::Array;
 use crate::array_session;
+use crate::arrays::ConstantArray;
 use crate::arrays::ListViewArray;
 use crate::arrays::PrimitiveArray;
 use crate::assert_arrays_eq;
 use crate::dtype::DType;
+use crate::dtype::NativePType;
 use crate::dtype::Nullability;
+use crate::scalar::Scalar;
 use crate::validity::Validity;
 
 #[rstest]
@@ -55,6 +60,32 @@ fn test_canonical_dtype_matches_array_dtype() -> VortexResult<()> {
     Ok(())
 }
 
+#[test]
+fn test_try_from_parts_rejects_null_child() -> VortexResult<()> {
+    let child = PrimitiveArray::from_iter([1_i64]).into_array();
+    let masked = MaskedArray::try_new(child, Validity::AllValid)?;
+    let mut parts = match masked.try_into_parts() {
+        Ok(parts) => parts,
+        Err(_) => vortex_bail!("the uniquely owned masked array must expose its parts"),
+    };
+    let dtype = DType::Primitive(i64::PTYPE, Nullability::Nullable);
+    parts.slots[MaskedSlots::CHILD] =
+        Some(ConstantArray::new(Scalar::null(dtype), parts.len).into_array());
+
+    let error = match Array::<Masked>::try_from_parts(parts) {
+        Err(error) => error,
+        Ok(_) => vortex_bail!("rebuilding must reject a child containing nulls"),
+    };
+
+    assert!(
+        error
+            .to_string()
+            .contains("MaskedArray children must not have nulls"),
+        "unexpected error: {error}",
+    );
+    Ok(())
+}
+
 #[test]
 fn test_masked_child_with_validity() {
     // When validity has nulls, masked_child should apply inverted mask.

vortex-array/src/arrays/masked/vtable/mod.rs

@@ -73,6 +73,7 @@ impl VTable for Masked {
         *ID
     }
 
+    #[allow(clippy::disallowed_methods)]
     fn validate(
         &self,
         _data: &MaskedData,
@@ -92,6 +93,10 @@ impl VTable for Masked {
             child.dtype().as_nullable() == *dtype,
             "MaskedArray dtype does not match child and validity"
         );
+        vortex_ensure!(
+            child.all_valid(&mut legacy_session().create_execution_ctx())?,
+            "MaskedArray children must not have nulls",
+        );
         Ok(())
     }
 

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

@@ -0,0 +1,91 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Execution arguments paired with the metadata selected during planning.
+//!
+//! [`BorrowedExecutionArgs`] can point at original, sliced, or filtered arrays while retaining the
+//! dtypes, output dtype, and null policy of the original batch plan.
+
+use vortex_error::VortexResult;
+use vortex_error::vortex_err;
+
+use crate::ArrayRef;
+use crate::dtype::DType;
+use crate::scalar_fn::ExecutionArgs;
+use crate::scalar_fn::unstable::row::visitor::RowPolicy;
+
+/// A borrowed [`ExecutionArgs`] view with the planning metadata selected for its row kernel.
+///
+/// `arrays` may be filtered or sliced, while `dtypes` and `output_dtype` always describe the
+/// original planned batch. Keeping them together prevents an execution path from pairing an input
+/// view with unrelated planning metadata.
+#[derive(Clone, Copy)]
+pub(crate) struct BorrowedExecutionArgs<'a> {
+    /// The input arrays for this kernel invocation.
+    arrays: &'a [ArrayRef],
+
+    /// The number of rows in this kernel invocation.
+    row_count: usize,
+
+    /// The original input dtypes used to select the row implementation.
+    dtypes: &'a [DType],
+
+    /// The non-nullable dtype built by the selected output capability.
+    output_dtype: &'a DType,
+
+    /// The nullable execution policy selected during planning.
+    policy: RowPolicy,
+}
+
+impl<'a> BorrowedExecutionArgs<'a> {
+    /// Pair one input view with the planning metadata selected for its batch.
+    pub(crate) fn new(
+        arrays: &'a [ArrayRef],
+        row_count: usize,
+        dtypes: &'a [DType],
+        output_dtype: &'a DType,
+        policy: RowPolicy,
+    ) -> Self {
+        Self {
+            arrays,
+            row_count,
+            dtypes,
+            output_dtype,
+            policy,
+        }
+    }
+
+    /// Return the original input dtypes used to select the row implementation.
+    pub(crate) fn dtypes(&self) -> &'a [DType] {
+        self.dtypes
+    }
+
+    /// Return the non-nullable dtype built by the selected output capability.
+    pub(crate) fn output_dtype(&self) -> &'a DType {
+        self.output_dtype
+    }
+
+    /// Return the nullable execution policy selected during planning.
+    pub(crate) fn policy(&self) -> RowPolicy {
+        self.policy
+    }
+}
+
+impl ExecutionArgs for BorrowedExecutionArgs<'_> {
+    fn get(&self, index: usize) -> VortexResult<ArrayRef> {
+        self.arrays.get(index).cloned().ok_or_else(|| {
+            vortex_err!(
+                "row-function input index must be less than {}, got {index}",
+                self.arrays.len(),
+            )
+        })
+    }
+
+    fn num_inputs(&self) -> usize {
+        self.arrays.len()
+    }
+
+    fn row_count(&self) -> usize {
+        self.row_count
+    }
+}