Tracking Issue: Execute `RowFn` over Vortex arrays

[connortsui20]

This issue tracks the private machinery that executes a RowFn over Vortex arrays.

Parent Epic: #9128

Related API tracking issue: #9129

Design

A RowFn provides a typed row kernel. The batch executor adds the columnar behavior that every strict row function needs:

1. Plan the output dtype and nullable-row policy from the concrete dispatch.
2. Short-circuit null constants and evaluate all-constant calls once.
3. Preserve batch constants as one decoded row.
4. Conjoin the input validities.
5. Select dense, skip-invalid, or filter-and-scatter execution.
6. Try an encoding-aware reduce_encoded result when the selected path permits it.
7. Validate the output dtype and row count, then apply strict validity.

This machinery is private. A function author selects typed elements and an output capability. The executor owns the batch strategy.

Nullable-row policies

Planning selects one of three policies for each concrete dispatch:

• Dense evaluates every row and masks the output. It requires null-safe decoding and an infallible row computation.
• DenseWithRetry also evaluates every row. If it sees deferred failure evidence, it retries only the valid rows.
• ValidOnly never evaluates the row closure for an invalid row. It selects skip-invalid or filter-and-scatter execution for a mixed validity mask.

The retry is needed because dense execution can inspect payloads behind nulls. An error from such a payload is not observable. The filtered retry keeps an error from a valid row and suppresses an error that came only from null rows.

RowExecution

The row loop returns VortexResult<RowExecution>. Together, these types represent three outcomes:

• Err(error) is an immediate or structural error. Retrying cannot help.
• RowExecution::Output(output) is a successful row loop.
• RowExecution::DeferredError(error) is failure evidence from a completed row loop.

The third state is why the row loop cannot return only VortexResult<ArrayRef>. Batch execution must keep a deferred error separate until it knows whether the failing row was valid.

Once execution contains only valid rows, From<RowExecution> for VortexResult<ArrayRef> converts the deferred error into an ordinary error.

Deferred failures

visit_deferred returns an owned output and a small failure value for each row. The executor OR-reduces those values in a loop-local, then constructs one rich error after the loop.

The failure value must not be wider than the output value. A wider reduction lowers the vector width and can make checked arithmetic much slower. Keeping the accumulator out of the output sink also avoids a loop-carried memory dependency.

Sink visits can report three result forms:

• () for an infallible write.
• VortexResult<()> for an immediate failure.
• deferred evidence for one error check after the loop.

Integer division uses the immediate form with UninitElementSink. Division is already scalar and expensive. An immediate check can stop at the first failure, while uninitialized dense output avoids filling every slot before the row loop.

Skip-invalid and filter-and-scatter execution

For a mixed mask under ValidOnly, the executor can compute only the valid row indices in the original inputs. This path requires two contracts:

• every InputElement must provide a null-tolerant decode for the concrete array.
• the OutputSink must support skipped rows and initialize legal placeholders.

The executor masks those placeholders before returning the output. If either contract declines, the executor filters every input to the valid rows, runs the dense kernel, and scatters the result back into a full-length nullable array.

Owned output visits do not have a sink that can initialize skipped slots. Their valid-only visitor therefore declines and uses filter-and-scatter. UninitElementSink supports skipped slots, which keeps nullable integer division on the original inputs.

The current heuristic uses InputElement::FILTERED_DECODE_COST:

• cost 0 always prefers skip-invalid execution.
• cost 1 prefers it when at least 50% of rows survive.
• cost 2 or more prefers it when at least 85% of rows survive.

These thresholds are private executor policy. They need more calibration as new element types are added.

Constants and encodings

Constant decoding and prepared computation are separate. The tuple adapter stores a batch constant as one decoded row. A prepared visitor can then derive shared state from that value once per batch.

reduce_encoded can bypass the row loop. The unfiltered path probes the original arrays before it tries a skip-invalid sink. Filter-and-scatter can also probe the compacted arrays. The returned array still goes through the common dtype, length, and validity checks.

Steps

• Derive arity, strictness, fallibility, output dtype, and validity from RowFn.
• Implement null-constant and all-constant handling.
• Implement dense execution, filtering, scattering, and output validation.
• Implement skip-invalid execution for supporting sinks.
• Implement deferred failure reduction and valid-row retry.
• Preserve encoding-aware reductions in the batch executor.
• Add focused executor and validity benchmarks.
• Migrate primitive numeric arithmetic as the first production user.
• Calibrate the skip-invalid thresholds with more element types and mask shapes.
• Validate downstream element and sink implementations.
• Add migration benchmarks before replacing more hand-written kernels.

Unresolved questions

• What benchmark set and regression threshold should gate a migration?
• Should the skip-invalid cost model include arity, mask shape, and scatter cost?
• Is a skipped-row contract useful for owned output visits, or is filtering sufficient?

Follow-ups

• Reduce the allocations and passes in filter-and-scatter.
• Avoid duplicate reduce_encoded probes when an unfiltered sink declines.
• Revisit the thresholds after adding an element with substantial per-row decode work.
• Keep generated code checks for deferred arithmetic alongside wall-clock benchmarks.

Implementation history

The current branch separates the framework, primitive numeric integration, and focused executor benchmarks into reviewable commits.