Follow up on post-merge review of the native sequence kernel: perf crossover, unbounded per-batch allocation, and missing ceiling coverage

[andygrove]

Follow-up from a post-merge review of #5614 (merged as 7190df6), which added the native spark_sequence kernel for integral element types. Review comments are at #5614 (review).

Nothing here is a correctness problem. I forced all five of Spark's sequenceLength failure paths through the native kernel with column arguments and compared against Spark on both 3.5 and 4.1, and the exception class and message match byte for byte on every one. A 4000-row randomized fuzz biased toward the Long and Int boundaries found no divergences across 3628 comparable rows, and fourteen downstream consumers of the produced list all match Spark. The sequenceLength port itself is exact against the 3.5.8 and 4.1.3 sources, including the Long.MinValue / -1 special case and the ordering of the size check ahead of internalError in the BigInt fallback.

The items below are performance, memory-safety and coverage follow-ups.

1. The native kernel is slower than Spark above roughly a thousand elements per row

The benchmark table in the PR stops at 10,000 elements per row and reports it as parity, so a reader concludes the native path is never worse than Spark. It is. Measured on an M3 Max, release build, local[1], 8192 rows, best-of-7 after 3 warmups, with the routing asserted from ExtendedExplainInfo on every arm:

elems/row	Spark	native	dispatcher	native vs Spark	native vs dispatcher
5	59 ms	41 ms	46 ms	1.44X	1.12X
365	42 ms	34 ms	222 ms	1.24X	6.53X
1000	40 ms	36 ms	518 ms	1.11X	14.39X
10000	51 ms	72 ms	4624 ms	0.71X	64.22X
50000	120 ms	231 ms	23307 ms	0.52X	100.90X

The PR's own CometSequenceBenchmark reproduces the sign independently on the same machine: seq_long_10000_elems gives Spark best 62 ms / avg 73 ms against Comet best 109 ms / avg 112 ms, so Spark's average beats Comet's best and this is not measurement noise. The seq_date_spine_dispatcher control reads 0.7X here against 1.0X in the PR, suggesting this machine sits about 1.4x in Spark's favour relative to the author's, but even allowing for all of that the long shapes do not reach parity.

This looks inherent to the representation rather than a defect in the element loop. Spark allocates one long[] per row, which at 10,000 elements is 80 KB and stays resident in L2 while it is written and immediately consumed, whereas a per-batch buffer has to stream to DRAM. So the action is to document the crossover, not to rewrite the kernel:

• State the crossover in the sequence entry of docs/source/contributor-guide/expression-audits/array_funcs.md.
• Add a dispatcher arm to CometSequenceBenchmark. Against the JVM codegen dispatcher, which is what sequence did before this landed and therefore the real baseline for every existing Comet user, the native kernel is 6.5X to 101X faster. That is a much stronger result than 2X to 3X against Spark and it is currently unmeasured. Relates to [EPIC] Criterion bench coverage for all native expressions #5396.

2. Up to 3.3 GB allocated per batch outside Comet's memory pool, with the only ceiling at about 17 GB

native/spark-expr/src/array_funcs/sequence.rs guards the reservation with try_reserve_exact, which turns an allocator refusal into a query error. That is worth having, but it does not bound the allocation. The only cap is total > i32::MAX elements, which for bigint is about 17 GB, and the Vec comes from the global allocator rather than DataFusion's MemoryPool. The allocation is therefore not counted against spark.comet.memory*, cannot be spilled, and applies no back-pressure.

Measured at the default spark.comet.batchSize, 50,000 elements per row allocates 3.3 GB in a single reservation and completes. Peak process RSS for that query is 2852 MB above baseline against Spark's 1076 MB for the same query, a 2.6x higher peak. On a Linux executor with overcommit the OOM killer arrives well before try_reserve_exact gets a chance to return Err, so the graceful path is the one a user is least likely to reach. This is also the peak-memory question raised during review that was never answered with a measurement, and it was only ever measured at local[1], so behaviour under concurrent tasks is still unknown.

• Add a byte-based ceiling alongside the element count, sized from the batch memory budget rather than i32::MAX, so SequenceBatchTooLarge fires while the executor is still healthy.
• Measure peak memory with concurrent tasks rather than local[1].
• Consider whether this should go through the memory pool at all. Relates to Investigate adopting DataFusion's allocator-level memory accounting to replace manual memory tuning #4576.

3. SequenceBatchTooLarge has no test, and it is nearly free to test

This is the one behaviour introduced by the PR that fails a query Spark completes, it has a new mapping duplicated across all three version shims, and nothing exercises it. It is also cheap to cover, because the total > i32::MAX check runs in the first pass before a single byte is allocated.

Confirmed working against the merge commit: at spark.comet.batchSize=8192, sequence(0, 262143) over 8192 single-partition rows produces the intended message naming spark.comet.batchSize, while Spark returns 2147483648. Lowering spark.comet.batchSize to 4096 makes Comet return Spark's answer, so a fixture pins the documented remedy as well as the error.

• Add to spark/src/test/resources/sql-tests/expressions/array/sequence.sql:

statement
CREATE TABLE t_seq_ceiling(a INT, b INT) USING parquet

query expect_error(Lower `spark.comet.batchSize`)
SELECT sum(CAST(size(sequence(a, b)) AS BIGINT)) FROM t_seq_ceiling

with a = 0, b = 262143 over 8192 rows in a single partition.

4. The leaf-arguments-only gate excludes the shape most real queries use

CometSequence.argsAreLiteralsOrRefs is the right conservative call for the null and side-effect problem and is not in question here. It does mean the native path only engages when both endpoints already exist as columns or literals, so the idiomatic spine sequence(x, x + n) stays on the dispatcher, and so does anything behind a coercion CAST. The benchmark is the evidence: it had to be rewritten to materialise c_stop_5, c_stop_365 and c_stop_10000 as stored columns before any integral case reached the kernel at all.

The workaround a user would reach for first does not work either. FROM (SELECT c_start, c_start + 364 AS c_stop FROM p) is folded straight back by CollapseProject and the explain still reports JVM codegen dispatcher: sequence, so there is no way to opt in short of rewriting the table.

• Document which argument shapes reach the native path. "Leaf arguments only" is not something a user can map onto their own SQL.
• Investigate a safe widening: accepting an argument subtree that provably cannot throw and preserves nulls would cover x + n at least under non-ANSI.

5. Smaller cleanups

• spark/src/test/resources/sql-tests/expressions/array/sequence.sql has a PR review-thread URL (#discussion_r3910237757) in a comment. Comments should describe the code rather than how it came to be, and that link will not survive the next change to the surrounding reasoning. The #5349 reference at the top of the file is the durable kind and should stay.
• The SequenceBatchTooLarge message text is written out four times, in native/common/src/error.rs and verbatim in all three ShimSparkErrorConverter files. Unlike SequenceIllegalBoundaries it has no version-specific behaviour, so a shared constant would keep the copies from drifting. The case "Internal" arm added by the same PR is character-identical in all three shims too.
• The sequence row in docs/source/user-guide/latest/expressions.md describes the native versus dispatcher split, which a user cannot observe, and omits the per-batch ceiling, which is the one thing they can. The audit entry covers it, but that is not where somebody who has just hit the error will be looking.
• max_elements in the CollectionSizeLimitExceeded JSON payload is now dead. All three shims call createArrayWithElementsExceedLimitError, which supplies MAX_ROUNDED_ARRAY_LENGTH itself, so the field is only read by the Rust-side Display.

Not covered by this review

Spark 3.4, 4.0 and 4.2 were not executed locally, though CI is green on all of them. The error-parity probes ran on 3.5 (Scala 2.12) and 4.1 (Scala 2.13). The fuzz-testing/ module was not run. Iceberg and shuffle interactions with the produced ListArray were not examined beyond the fourteen downstream consumers listed above.