Review the RowFn prototype, merge develop, and apply the review findings

Cargo.toml

@@ -160,7 +160,13 @@ flatbuffers = "25.2.10"
 fsst-rs = "0.6.0"
 futures = { version = "0.3.31", default-features = false }
 fuzzy-matcher = "0.3"
-geo = "0.31.0"
+# `vortex-geo`'s `contains_route` transcribes geo's `impl_contains_from_relate!` dispatch table, so
+# any bump that moves a row silently changes containment verdicts — the tests stay green wherever
+# relate and the direct algorithm agree. Pinned exactly so that taking any new geo, patch releases
+# included, is a deliberate edit of this line that re-verifies the table; a caret requirement would
+# let `cargo update` (or automated lockfile maintenance) take 0.31.x with no diff to review. See
+# `vortex-geo/src/scalar_fn/contains.rs`.
+geo = "=0.31.0"
 geo-traits = "0.3.0"
 geo-types = "0.7.19"
 geoarrow = "0.8.0"

SCALAR_FN_HANDOFF.md

@@ -278,6 +278,152 @@ The checks recorded for the final API state are:
 The generated-code comparison and native timing evidence are described above and in the final
 section of `STRICT_SCALAR_FN_RESEARCH.md`.
 
+## Review pass: what changed and what was deliberately left
+
+A review of the three parts (API, execution, implementations). **The author-facing API is
+unchanged**: every proposal that would have altered it was backed out, for the reasons below, and
+what landed is cleanup, corrected documentation, and test coverage. The emitted IR of every
+`visit_prepared_into` monomorph is identical to the pre-review commit.
+
+API:
+
+- `InputElement::decode_null_tolerant` overrides that only restated the default were deleted from
+  the primitive, bool and `TensorRow` elements. `GeometryRow`'s override is the only real one. The
+  doc now says a dense-safe element should *not* override.
+- `ElementTuple` now records why it carries arities past the widest function in tree: it is sealed,
+  so a downstream crate cannot add the one it needs, and an uninstantiated arity costs only its own
+  macro expansion.
+
+Execution:
+
+- `execute_filtered` and the forced-strategy test seam now share `resolve_validity`, so the mask
+  materialization and the all-true/all-false shortcuts cannot drift apart between them.
+- The dense-retry path's comment was wrong and is corrected. It filters unconditionally because
+  `execute_dense` is not handed the `branch` closure, **not** because a deferred sink cannot skip
+  rows: `ERRORS_ARE_DEFERRED` and `SUPPORTS_SKIPPED_ROWS` are independent consts and a sink may
+  legally set both.
+
+Implementations:
+
+- `l2_norm_row` had two copies, in `l2_norm.rs` and `cosine_similarity.rs`. Cosine's prepared and
+  per-row arms must agree bit for bit, which only holds while both accumulate in the same order, so
+  the duplicate was an invitation to break exactly the property the comments defend. One copy now
+  lives in `utils.rs` beside the other shared tensor helpers.
+- `CosineSimilarity::reduce_encoded` zips its three slices instead of indexing `0..len` three times
+  per row, and documents why it materializes where `InnerProduct::reduce_encoded` stays lazy (the
+  zero-norm guard is a conditional, not an arithmetic factor).
+- `IndexedSourceExt::map_checked_into` was deleted from vortex-compute. `CheckedSink` replaced the
+  split value/evidence pass it served, and it had no caller left.
+- `contains_route` and the workspace `geo` dependency both record that the table transcribes geo's
+  `impl_contains_from_relate!` and must be re-verified on a version bump. `geo` is pinned to
+  `=0.31.0`: a caret requirement would admit 0.31.x patches, which `cargo update` (or automated
+  lockfile maintenance) takes with no diff to review, and a patch is free to reshuffle the dispatch
+  without any API change. The agreement tests stay green wherever relate and the direct algorithm
+  agree, so the pin, not the suite, is what makes the coupling break only deliberately.
+
+Split out onto `develop` instead of landing here:
+
+- **The checked-arithmetic macro collapse.** `primitive.rs` on this branch and on `develop` both
+  carry four near-identical `CheckedArithmetic` bodies that differ only in `mul_failure`, so the
+  collapse into one `impl_checked_integer!` belongs on `develop` where every caller benefits. It is
+  on `claude/collapse-checked-arith-macros`. This branch's `primitive.rs` keeps its four bodies
+  until `develop` is merged, at which point the collapse arrives with it and the merge conflict is
+  a member deletion rather than two competing macro structures.
+- **The `mul_failure` kernel tests.** The exhaustive 8-bit sweep and the 64-bit probe grid already
+  exist on `develop` from vortex-data/vortex#9210 and arrive with the same merge.
+
+Deliberately **not** done:
+
+- **No `DeferredElementSink`.** `CheckedSink` exists largely because `ElementSink` cannot name an
+  error at `finish`. A framework sink combining an element output with a type-level message would
+  remove ~100 lines per function, but there is exactly one deferred-error function. Build it when a
+  second appears, rather than copying `CheckedSink`.
+- **No change to `reduce_encoded`'s probe semantics.** Hoisting the probe out of the strategy paths
+  and masking a full-length result looks like a simplification and is not one:
+  `normalized_readthrough_survives_null_rows` pins that a filtered input is no longer `Normalized`,
+  so which arrays reach `reduce_encoded` is load-bearing and differs per strategy.
+- **No PR split.** Recommended landing order, each step individually revertible and separately
+  benchmarkable: (1) API + lifting with dense/filter only; (2) branch-and-skip + adaptive selection
+  + its benchmarks; (3) `NumericBinary`; (4) tensor; (5) geo. The seam already supports this split
+  and no API changes between steps.
+
+### Three API changes proposed, and why none of them landed
+
+All three were implemented, run against the suite, and backed out. None prevents a bug, and this
+branch's open work is *settling* the API rather than churning it, so they belong in #9129 as
+questions decided alongside the rest of the surface:
+
+- **Should `reduce_encoded` take an explicit `row_count`?** The filtered-count requirement is real
+  and easy to miss, but `args` are filtered to match, so `args[0].len()` is already both the natural
+  thing to write and correct. The parameter is documentation, and it costs every implementor a
+  signature change. What survived is the test:
+  `reduce_encoded_is_probed_before_and_after_filtering` pins that the rewrite is offered the
+  original arrays at full length and then the filtered ones at the surviving count.
+- **Should `OutputSink::row_count_matches` become `rows_len`?** A length reads cleaner and lets the
+  executor name what it found. Against that, `row_count_matches` lets a sink fold in its own
+  invariants, which `SpreadSink` uses for its width check; narrowing it turns that into a panic.
+  Neither spelling prevents a bug.
+- **Should the nullary path go?** A function with no inputs has no validity to lift, which is the
+  lifting's whole job. But `RowFn` would still give it sink allocation and dtype derivation, so
+  `random()` or `now()` is not obviously better hand-written, and the path is ~70 lines and tested.
+
+Trimming `ElementTuple` to arity four was proposed on the same reasoning and backed out for a
+stronger one: the trait is sealed, so the arities are the only ones a downstream crate can ever
+have.
+
+### Two changes this pass made and then reverted
+
+Both were proposed, implemented, reviewed, and backed out on evidence. They are recorded because
+each is an attractive idea that a later reader will have again.
+
+**Making `CheckedSink` safe with `BufferMut::zeroed` costs 1.65 to 1.71x.** Replacing the
+`MaybeUninit` storage removes an `unsafe set_len` and reads as a clear win, and `ElementSink`'s own
+comment appears to bless it by routing a zeroable placeholder to `alloc_zeroed`. Measured, it is
+not: allocate-zeroed-then-fill against allocate-then-fill, interleaved in one process over `u64`
+outputs, ran **1.221x** slower at 8 KiB, **1.71x** at 64 KiB, **1.66x** at 512 KiB and **1.71x** at
+2 MiB, stable to within 2% across two runs. `alloc_zeroed` does not avoid the write: below glibc's
+mmap threshold `calloc` recycles a dirty chunk and memsets it, and above it every fresh page faults
+on first touch. The row loop overwrites every slot regardless, so this is a duplicated pass over
+the output of the hottest kernel in the system.
+
+Note the corollary, which is a real optimization nobody has taken: `ElementSink::with_capacity`
+pays exactly this on every batch, and only branch-and-skip ever reads a placeholder back. A sink
+that allocated uninitialized on the dense and filter paths would recover it.
+
+**Hoisting `OutputSink::SUPPORTS_SKIPPED_ROWS` into the plan is not sound as an optimization.**
+#9130 records "avoid probing `reduce_encoded` twice when branch execution is unsupported" as a
+follow-up. It reads as free, and is not, because the branch path probes `reduce_encoded` against
+the _original_ arrays before it consults the sink, and that is the only probe that ever sees them
+still encoded. Skipping the path early leaves such a function with only the filtered probe, whose
+canonical arrays match no encoding fast path. For a function whose reduction is _defined_ to answer
+differently from its row loop, which is exactly what `L2Norm` over `Normalized` is, that is a wrong
+answer rather than a slow one. Nothing in tree is reachable today only because every `ValidOnly`
+dispatch happens to use `ElementSink`. **#9130's follow-up should be struck, not implemented.**
+`reduce_encoded_is_probed_before_and_after_filtering` now pins the two probes and their row
+counts.
+
+### On measurement, and what the IR gate does and does not cover
+
+Wall-clock benchmarking of the row loops was attempted first and abandoned on evidence. Two runs of
+the *same* baseline binary, pinned with `taskset -c 2`, 100 samples, disagreed by up to 4x
+(`row_wrapping_add_nullable`: 198.8 us then 52.9 us median; `specialized_checked_add`: 185.5 us then
+34.4 us). The 4-vCPU shared VM drifts more within a session than any effect being measured, which is
+the same conclusion this branch already reached on a dedicated 7950X.
+
+The gate used instead is the emitted optimized IR of every `visit_prepared_into` monomorph in
+`vortex-array`, profiled by vector width, reduction count, overflow-intrinsic survival and bounds
+checks, then compared as a multiset before and after. Reproduce with:
+
+```bash
+RUSTFLAGS="--emit=llvm-ir -C codegen-units=1" cargo rustc -p vortex-array --release --lib
+```
+
+**Its blind spot is worth stating, because it nearly landed a regression.** The IR of a row loop
+cannot show an allocator call outside it, so the `BufferMut::zeroed` substitution above passed this
+gate cleanly while costing 1.7x. An allocation-strategy change needs its own targeted A/B, which is
+cheap to write and immune to the host drift above because both arms run interleaved in one process.
+Use the IR gate for loop shape and a focused microbenchmark for anything the loop does not contain.
+
 ## Remaining boundaries
 
 - Complete the required x86 production and forced-null-strategy benchmark run above before treating

vortex-array/src/scalar_fn/fns/binary/numeric/row.rs

@@ -121,12 +121,23 @@ fn operand_ptype(args: &[DType]) -> VortexResult<PType> {
 }
 
 /// Visit at two `T` columns, applying `Op` per row into the sink that defers its overflow bit.
+///
+/// The const block enforces, at monomorphization time, the width rule stated on
+/// [`Failure`](super::primitive::Failure): evidence wider than the element would make the
+/// OR-reduction rather than the arithmetic decide how many rows fit in a vector.
 fn visit_checked<T, Op, V>(visitor: V) -> VortexResult<V::Out>
 where
     T: NativePType,
     Op: CheckedPrimitiveOp<T>,
     V: RowVisitor,
 {
+    const {
+        assert!(
+            size_of::<Op::Failure>() <= size_of::<T>(),
+            "failure evidence must be no wider than the value, or it bounds the vector width"
+        )
+    };
+
     visitor.visit_prepared_into::<(T, T), CheckedSink<T, Op>, _, _>(
         |_| (),
         |&(), (lhs, rhs), output| output.write(lhs, rhs),
@@ -145,9 +156,21 @@ where
 /// is what lets unsigned multiplication report its discarded high half instead of a comparison, and
 /// so stay vectorized.
 ///
+/// **The storage is deliberately uninitialized, not zeroed.** Substituting `BufferMut::zeroed` to
+/// make the sink safe was measured at **1.65 to 1.71x** the cost of allocate-and-fill, stable across
+/// two runs and every batch size from 8 KiB to 2 MiB, because `alloc_zeroed` does not avoid the
+/// write: below glibc's mmap threshold `calloc` recycles a dirty chunk and memsets it, and above it
+/// the first touch of each fresh page faults instead. The row loop overwrites every slot regardless,
+/// so that pass is pure duplicate work on the hottest kernel in the system. This is the case the
+/// repository's "avoid `unsafe` unless it is necessary" rule leaves room for: the safe spelling
+/// exists, and it costs a second pass over the output.
+///
 /// Rows are written into uninitialized storage, so this sink cannot finish a batch whose rows were
-/// not all visited and leaves [`OutputSink::SUPPORTS_SKIPPED_ROWS`] at `false`. Nothing is lost: a
-/// deferred-error kernel runs densely and retries valid rows only on its cold error path.
+/// not all visited, and leaves [`OutputSink::SUPPORTS_SKIPPED_ROWS`] at `false`. Nothing is lost:
+/// `SUPPORTS_SKIPPED_ROWS` is what makes branch-and-skip unavailable, which is the guard that keeps
+/// the uninitialized slots sound. Note this is _not_ implied by the dispatch policy alone: a
+/// deferred result still reaches the executor's valid-only policy whenever its arguments are not
+/// dense-safe, so the `false` here is load-bearing rather than a restatement.
 struct CheckedSink<T: NativePType, Op: CheckedPrimitiveOp<T>> {
     /// The result values, initialized one row at a time up to `row_count`.
     values: BufferMut<T>,
@@ -160,8 +183,7 @@ struct CheckedSink<T: NativePType, Op: CheckedPrimitiveOp<T>> {
     op: PhantomData<Op>,
 }
 
-/// The uninitialized output slots of a [`CheckedSink`], borrowed once for the row loop, together
-/// with the batch-wide failure reduction they contribute to.
+/// The uninitialized output slots of a [`CheckedSink`], borrowed once for the row loop.
 struct CheckedRows<'a, T: NativePType, Op: CheckedPrimitiveOp<T>> {
     values: &'a mut [MaybeUninit<T>],
     op: PhantomData<Op>,

vortex-array/src/scalar_fn/mod.rs

@@ -13,9 +13,10 @@
 //! closure. Implement `RowFn` when the function fits it, and `ScalarFnVTable` when it does not.
 //!
 //! [`RowFn`] is for a kernel whose value at a row is determined by that row alone, and which has to
-//! read every row anyway: `vortex.byte_length`, `vortex.tensor.l2_norm`,
-//! `vortex.tensor.inner_product`, `vortex.geo.distance`. Name the element types and write the row
-//! closure, and the rest is derived, including which rows get visited.
+//! read every row anyway: the arithmetic operators over primitive columns, `vortex.tensor.l2_norm`,
+//! `vortex.tensor.inner_product`, `vortex.tensor.cosine_similarity`, `vortex.geo.distance`,
+//! `vortex.geo.contains`. Name the element types and write the row closure, and the rest is
+//! derived, including which rows get visited.
 //!
 //! Its *input* side is open. [`InputElement::Elem`] is a GAT, so an element can hand the closure
 //! borrowed variable-length data (a byte-string element yielding `&[u8]`) or drill through a wrapper
@@ -63,6 +64,10 @@
 //! - **A row is not the natural unit of work.** `vortex.not` is one `!` per 64-bit word, in place
 //!   when the bit buffer is unshared, against 64 loop iterations and 64 bit writes, and its
 //!   encoding-aware fallback pushes the inversion down instead of canonicalizing.
+//! - **The row's value is cheaper to read than the row.** `vortex.byte_length` was tried as a row
+//!   function and measured 7.6x slower than its columnar implementation, because the length is a
+//!   field of the view and the row loop paid to resolve the bytes it never looked at. Being
+//!   row-determined is necessary but not sufficient.
 
 use vortex_session::registry::Id;
 

vortex-array/src/scalar_fn/row/element/bool.rs

@@ -36,13 +36,6 @@ impl InputElement for bool {
         Ok(array.execute::<BoolArray>(ctx)?.into_bit_buffer())
     }
 
-    fn decode_null_tolerant(
-        array: ArrayRef,
-        ctx: &mut ExecutionCtx,
-    ) -> VortexResult<Option<Self::Column>> {
-        Self::decode(array, ctx).map(Some)
-    }
-
     fn get(column: &Self::Column, index: usize) -> bool {
         column.value(index)
     }

vortex-array/src/scalar_fn/row/element/mod.rs

@@ -96,10 +96,12 @@ pub trait InputElement: 'static {
     /// Decode `array` _without_ assuming every row is valid, or `Ok(None)` when this element
     /// cannot for this particular array.
     ///
-    /// An element with [`DENSE_SAFE`](Self::DENSE_SAFE) set may use the default, because its ordinary
-    /// decode already tolerates null payloads. Other elements may override this by writing an
-    /// arbitrary placeholder into null slots; the caller guarantees [`get`](Self::get) is never
-    /// called for such a row. It is what the branch-and-skip null strategy decodes with.
+    /// An element with [`DENSE_SAFE`](Self::DENSE_SAFE) set **should not** override this: its
+    /// ordinary decode already tolerates null payloads, so the default is already correct and an
+    /// override just restates it. Overriding is for an element that is *not* dense-safe but can
+    /// still write an arbitrary placeholder into null slots; the caller guarantees
+    /// [`get`](Self::get) is never called for such a row. It is what the branch-and-skip null
+    /// strategy decodes with.
     ///
     /// Return `Ok(None)` rather than an error when an array has no null-tolerant decode; the lifting
     /// falls back to the filter strategy.

vortex-array/src/scalar_fn/row/element/primitive.rs

@@ -43,13 +43,6 @@ impl<T: NativePType> InputElement for T {
         Ok(array.execute::<PrimitiveArray>(ctx)?.into_buffer::<T>())
     }
 
-    fn decode_null_tolerant(
-        array: ArrayRef,
-        ctx: &mut ExecutionCtx,
-    ) -> VortexResult<Option<Self::Column>> {
-        Self::decode(array, ctx).map(Some)
-    }
-
     fn get(column: &Self::Column, index: usize) -> T {
         column[index]
     }

vortex-array/src/scalar_fn/row/element/tuple.rs

@@ -143,6 +143,10 @@ pub(in crate::scalar_fn::row) fn batch_constant(array: &ArrayRef) -> Option<Arra
 /// visits with. Implemented for `()` and tuples of one through twelve elements. This trait is
 /// framework-only; add a new decode primitive by implementing [`InputElement`], then use it inside
 /// one of those tuples.
+///
+/// The arities past the widest function in tree are deliberate. This trait is **sealed**, so a
+/// downstream crate cannot add the one it needs, and an unused arity costs only its own macro
+/// expansion: no monomorphization happens until something instantiates it.
 pub trait ElementTuple: 'static + private::Sealed {
     /// The decoded column representations.
     type Columns;
@@ -198,7 +202,7 @@ pub trait ElementTuple: 'static + private::Sealed {
 
     /// Borrow every decoded column directly, or `None` when any argument is batch-constant.
     ///
-    /// This is selected once outside the hot loop. Keeping [`ArgColumn`] out of the resulting tuple
+    /// This is selected once outside the hot loop. Keeping `ArgColumn` out of the resulting tuple
     /// gives the optimizer ordinary contiguous column access without a per-row constant check.
     fn varying(columns: &Self::Columns) -> Option<Self::VaryingColumns<'_>>;