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

[robert3005]

Instead of performing a gather/execute/scatter add a filtered execute method that avoids majority of intermediate state

[robert3005]

Need to go over this carefully but I think this is a better sparse execution mode

[codspeed-hq]

Merging this PR will degrade performance by 0.95%

⚠️ Unknown Walltime execution environment detected

Using the Walltime instrument on standard Hosted Runners will lead to inconsistent data.

For the most accurate results, we recommend using CodSpeed Macro Runners: bare-metal machines fine-tuned for performance measurement consistency.

⚡ 2 improved benchmarks
❌ 2 regressed benchmarks
✅ 342 untouched benchmarks
⏩ 1890 skipped benchmarks¹

Warning

Please fix the performance issues or acknowledge them on CodSpeed.

Performance Changes

	Mode	Benchmark	BASE	HEAD	Efficiency
❌	WallTime	mul_i8_nonnull_avx2	11.2 µs	12.8 µs	-12.66%
❌	WallTime	mul_u8_nonnull_avx2	8.8 µs	9.8 µs	-10.09%
⚡	WallTime	multiply_shapes_neon[(128, PerRowPerRow)]	2.2 µs	2 µs	+11.07%
⚡	WallTime	mul_u32_nonnull_avx512	6.3 µs	5.7 µs	+10.35%

Tip

Investigate this regression by commenting @codspeedbot fix this regression on this PR, or directly use the CodSpeed MCP with your agent.

---

_{Comparing claude/rowfn-kernel-execution-fqw0ue (813e082) with develop (7feec60)}

Footnotes

1. 1890 benchmarks were skipped, so the baseline results were used instead. If they were deleted from the codebase, click here and archive them to remove them from the performance reports. ↩

[connortsui20]

For an input like:

[1, null, 3]

the filter/scatter on develop does this:

1. Remove the null row, producing [1, 3].
2. Allocate a two-row sink.
3. Write both rows.
4. Expand the result back to [1, null, 3].

The sink never has an unwritten row, so it does not need a skipped-row initializer.

This PR removes that path. It allocates three output rows and writes only rows 0 and 2. The sink cannot initialize row 1, so execution fails.

The test is wrong because it treats that new failure as required behavior. The old compact path can still produce the correct result safely.

---

So I think we should just have both paths? This new filtered execution should be an additional fast path, not a replacement. It avoids the scatter when the sink
can initialize skipped rows, but sinks that cannot do that still need the compact filter-and-scatter fallback.

[connortsui20]

replace this test with:

#[test]
fn test_no_skip_sink_executes_partially_valid_batch() -> VortexResult<()> {
    let validity = Validity::from_iter([true, false]);
    let input = PrimitiveArray::new(vec![1_i64, 2], validity.clone()).into_array();
    let args = VecExecutionArgs::new(vec![input], 2);
    let mut ctx = array_session().create_execution_ctx();

    let actual = execute_rows(&NoSkipIdentity, &EmptyOptions, &args, &mut ctx)?;
    let expected = PrimitiveArray::new(vec![1_i64, 0], validity).into_array();

    assert_arrays_eq!(&actual, &expected, &mut ctx);
    Ok(())
}

[robert3005]

I have pushed an update, this is a feature of the sink

[connortsui20]

this is what codex had to say:

Details

The review identifies two separate problems: an immediate merge-build failure and a deeper loss of OutputSink functionality.

The execution problem

Take a partially valid input:

input: [10, null, 30]
valid: [true, false, true]

The kernel must run only for 10 and 30, producing:

output: [f(10), null, f(30)]

There are three ways to execute this.

Strategy	Input rows	Sink rows	Requirement
Direct valid-row execution	3	3	Null-tolerant input decoding and skipped-row initializer
New filtered execution	2	3	Skipped-row initializer
Compact filter-and-scatter	2	2, then scatter to 3	No skipped-row initializer

The new path filters the input down to [10, 30], but it does not make the output compact. In execute_sink_filtered, it:

1. Calculates original_len = 3.
2. Allocates a three-row sink at line 211.
3. Initializes all three positions at line 222.
4. Writes filtered row 0 into output row 0.
5. Skips output row 1.
6. Writes filtered row 1 into output row 2.
7. Finishes the sink and only afterward masks row 1 as null.

Conceptually:

filtered input: [10, 30]
physical output before masking: [f(10), placeholder, f(30)]
logical output after masking:   [f(10), null,        f(30)]

That avoids the columnar scatter, which is useful. But the placeholder is essential. Sink::finish runs before the validity mask is applied, so the middle position must already contain something that the sink can legally finish.

Why initialization cannot be universally required

The patch changes the optional method into the required method at OutputSink::initialize_skipped_rows.

Its premise is effectively:

If a sink can allocate storage, it can always construct a legal placeholder value.

That does not follow. A sink might only be able to construct a value from the row callback’s inputs or prepared state. Its freshly allocated storage can be safe to drop while still being illegal to pass to finish until every slot has been written.

The old optional contract represented that distinction:

fn skipped_rows_initializer() -> Option<for<'a> fn(&mut Self::Rows<'a>)> {
    None
}

• Some(initializer) meant the sink supports full-length execution with skipped positions.
• None meant it does not, but it could still execute densely over a compact batch.

Making the initializer required prevents the second kind of sink from implementing OutputSink, even though filter-and-scatter can execute it correctly.

How filter-and-scatter handles that sink

The deleted fallback worked entirely in the valid-row domain:

filtered input:  [10, 30]
compact sink:    [f(10), f(30)]
nullable indices: [0, null, 1]
final output:    [f(10), null, f(30)]

The sink has only two positions, and the dense kernel writes both. There is no skipped sink position and therefore no placeholder requirement. The nullable take restores the original row positions afterward.

This makes filter-and-scatter a correctness fallback, not merely a slower implementation of the new path.

Why the merge result does not compile

The current base includes #9623, which added PolygonSink. It still implements:

fn skipped_rows_initializer() -> Option<...>

The PR removes that trait method and introduces required initialize_skipped_rows. When Git combines the changes:

• E0407: PolygonSink::skipped_rows_initializer is no longer a trait member.
• E0046: PolygonSink does not implement the new required initialize_skipped_rows.

PolygonSink itself can use an empty polygon as its placeholder, so adapting it would be easy. But doing so would only fix the compilation error. It would not fix the generic contract regression for sinks that return None.

Intended routing

The fix should preserve all three paths:

Partially valid batch
        |
        +-- Inputs support original-row decoding
        |       and sink initializer is Some
        |          -> direct valid-row execution
        |
        +-- Inputs must be filtered
        |       and sink initializer is Some
        |          -> new filtered-input/full-length-output path
        |
        +-- Sink initializer is None
                   -> compact filter-and-scatter

That also matches #9130, which says filter-and-scatter is used when either direct input decoding or skipped-row initialization declines, and #9129, which treats OutputSink as a downstream extension point.

A useful regression test would define a sink that leaves skipped_rows_initializer() at its default None, execute it on [1, null], and expect [1, null]. A test expecting an initializer-related error would only codify the regression.

[connortsui20]

I don't think we can make the skipped row initializer required, and if you rebase on the latest changes from #9623 that should explain why

This only is fine if we have types that work with completely arbitrary data that has been allocated, and the spatial types are a counterexample