Tracking Issue: Execute `RowFn` over Vortex arrays

[connortsui20]

This is a tracking issue for the private machinery that executes a RowFn over Vortex arrays.

Parent Epic: #9128

Related API tracking issue: #9129

This issue is mostly a WIP since I still need to figure out if the mechanics can be optimize further.

Design

A RowFn supplies typed row kernels. A blanket ScalarFnVTable implementation and private lifting layer turn those kernels into columnar execution. There is no separate public strict-function vtable.

The lifting owns the parts that should not be reimplemented by every function:

1. Read arity, dense safety, decode behavior, and fallibility off the RowFn argument witness, which a compile-time check pins against every dispatch. Run dispatch to validate the arguments and derive the result dtype from the visited OutputSink, then widen it when any input is nullable.
2. Short-circuit a null constant to an all-null result, or evaluate an entirely constant call once and broadcast it.
3. Decode each input once. A partially constant argument is decoded as one row and read with stride 0.
4. Derive the strict input validity and choose how the kernel sees rows behind nulls.
5. Execute the row kernel or an encoding-aware reduction, then reconcile the output dtype and apply validity.

Null handling has two contracts. Dense may evaluate payloads behind null rows and mask the result, so it is only available when every input element is DENSE_SAFE and execution cannot exit the loop early. Filter guarantees that the row computation only sees valid rows.

A deferred-error kernel is the exception that keeps Dense available while still being fallible. It writes a memory-safe provisional value for every row and hands back evidence of failure, which the executor OR-reduces across the batch, so there is no branch or Result discriminant in the hot loop. When such a batch is nullable and the reduction is non-zero, execution is retried over only its valid rows: success means the error came exclusively from rows that will be null anyway, and a second error is real.

Two properties of that reduction are load-bearing. Getting either wrong costs the loop its vectorization rather than producing a wrong answer, which makes both invisible to tests.

• It lives in a loop-local rather than in the sink. An accumulator reached through a &mut for every row is a loop-carried memory dependence, and moving it into the sink cost the boolean kernels 2.5x to 10x.
• Its width does not exceed the element's, which is why SinkResult names the word it reduces into instead of fixing one. A 64-bit accumulator bounds how many rows a vector of the reduction covers whatever the element width, and cost the primitive Mul kernel 3.1x at i8, 1.9x at i16 and 1.2x at i32.

Naming the word is also what lets a kernel report something other than a bit. Unsigned multiplication hands back the discarded high half of its widened product, because deriving a boolean from it costs a comparison that LLVM folds into llvm.umul.with.overflow, which has no vector lowering and scalarizes the whole loop.

Adaptive execution

For a mixed validity mask under Filter, the lifting chooses between two mechanisms per batch:

• Branch-and-skip decodes the original columns, computes only set rows, fills skipped output slots with placeholders, and masks the result. This avoids filtering and scattering and preserves the input encodings.
• Filter-and-scatter filters every input to the valid rows, executes densely, and scatters the result back. This is the fallback when null-tolerant decoding is unavailable, and can win when filtering avoids substantial per-row decode work.

The choice is invisible to the RowFn. InputElement::decode_null_tolerant first determines whether branch-and-skip is sound for the concrete arrays, and OutputSink::SUPPORTS_SKIPPED_ROWS whether the sink can finish an output whose skipped rows were never visited. ElementSink supports it by pre-filling placeholders; a builder that cannot finish a skipped row declines and the batch falls back to filter-and-scatter. DECODE_SHRINKS_WHEN_FILTERED then tells the selector whether filtering can avoid substantial decode work.

The current rule always prefers branch-and-skip for bulk decodes. For a per-row decode, it filters when fewer than 75% of rows survive. This is an experimental global threshold rather than part of the RowFn contract, and it is known to pick the slower strategy in both directions on x86. Recalibrating it depends on the cost of filter-and-scatter, which is itself a live target: see the follow-up below.

Even a Dense kernel may eventually benefit from skipping all-null mask words while keeping contiguous all-valid runs dense. That needs a skipped-row output contract and benchmarks for mask shape as well as surviving fraction.

Constant decoding and constant computation are separate. The machinery detects batch constants, decodes them once, and exposes their values to the prepare step. The function still decides which work can be hoisted and computes that state itself.

Steps

• Derive the ScalarFnVTable arity, strictness, fallibility, validity, and options serialization from RowFn, and derive its result dtype through dispatch.
• Implement null-constant and all-constant lifting, dense execution, filtering, scattering, and output dtype reconciliation.
• Implement branch-and-skip with null-tolerant decoding and output placeholders, gated per sink by SUPPORTS_SKIPPED_ROWS.
• Implement deferred-error dense execution and its valid-row retry.
• Add adaptive per-batch selection, forced-strategy test controls, and bytes/geometry crossover benchmarks.
• Preserve encoding-aware reductions across dense, filtered, and branch-and-skip execution.
• Validate the machinery with primitive, bytes, tensor, sink, fallible, and geometry functions.
• Add before-and-after benchmarks for representative constant, nullable, and ordinary row workloads.

Unresolved questions

• What benchmark set and regression threshold are required before replacing an existing hand-written implementation? Wall clock on a single host has not been sufficient for this machinery: two separate interventions moved a benchmark the wrong way, and host drift between sessions exceeded the effects under measurement. The emitted IR, specifically whether an overflow intrinsic survives and how wide the reduction is, has been the more reliable gate.
• What replaces the global survivor-fraction threshold: a per-element and arity-aware threshold, or a small cost model comparing estimated full-row decode work against survivor-only decode plus filter and scatter?

Non-blocking follow-ups:

• Reduce the cost of filter-and-scatter itself. It gathers through a take over a full-length index array and then filters every input, and at 90% nulls over 65536 rows it spends more time producing 6553 surviving rows than dense execution spends on all of them (null_strategy_bytes, Apple M4 Max). Removing a redundant masking pass was already worth 1.13-1.53x, so the crossover this trades against is not stable yet.
• Avoid probing reduce_encoded twice when branch execution is unsupported.
• Revisit the survivor-fraction threshold when another element with substantial per-row decode work exists.
• Allow the fallible branch loop to stop iterating immediately after the first error.

Measured dead ends, recorded so they are not retried:

• Bounds-check elimination in the row loop is not available. Narrowing the varying view to the row count so the index is provably in bounds buys nothing measurable, and get_unchecked is not uniformly a win: it is worth about 10% on mul_u16 and mul_u32 and costs 22% on mul_u8.
• A per-argument row source that keeps the Varying view when some other argument is batch-constant is 4x slower than the ArgColumn branch it replaces, which already vectorizes.
• A batch-constant operand therefore still demotes its neighbours off the slice path. Closing that needs the row loop monomorphized over which arguments are constant, which is what the hand-written kernels do with their four match arms, and it is worth revisiting when Compare moves onto RowFn since col < literal is exactly this shape.

Implementation history

None yet.