andygrove opened a new issue, #5712: URL: https://github.com/apache/datafusion-comet/issues/5712
Follow-up from a post-merge review of #5614 (merged as 7190df631afe3795914839203c7afe57ea23903c), which added the native `spark_sequence` kernel for integral element types. Review comments are at https://github.com/apache/datafusion-comet/pull/5614#pullrequestreview-5121907302. 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 #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 #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`: ```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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
