nat-openai opened a new pull request, #6819:
URL: https://github.com/apache/datafusion-comet/pull/6819
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases. You can
link an issue to this PR using the GitHub syntax. For example `Closes #123`
indicates that this PR will close issue #123.
-->
Partially addresses #5712: large integral-sequence performance and unbounded
output allocations. The issue’s remaining follow-ups stay open.
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in
the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your
changes and offer better suggestions for fixes.
-->
Integral sequence currently limits the number of generated elements, but has
no practical byte-based allocation limit. A single batch can allocate several
gigabytes, and concurrent tasks or retained outputs can multiply that memory
use. The native implementation also loses to Spark on the large shapes reported
in #5712.
This change bounds live sequence-buffer allocations and reduces sizing and
generation overhead.
Sequence buffers can outlive the producing native plan through Arrow clones,
slices and JVM exports. Integrating these allocations into the general memory
pool would require broader changes to propagate shared reservations across
those ownership boundaries, charge each backing allocation precisely once, and
release its reservation only after the final owner disappears. A dedicated
executor-wide allowance provides a bounded solution without requiring that more
invasive reservation-propagation work.
The 256 MiB default is configurable independently of Spark’s
execution-memory budget. Requests that exceed it fail before allocation;
applications can reduce the batch size or raise the allowance at startup. It
bounds live sequence backing buffers, rather than total native memory or
process RSS.
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is
sometimes worth providing a summary of the individual changes in this PR.
-->
- Add spark.comet.exec.sequence.maxBytesPerExecutor, read from executor
startup SparkConf, with a default of 256 MiB. Concurrent tasks share the
outstanding-byte counter; SQL session changes cannot enlarge the allowance.
- Validate and size the complete invocation before allocating. Account for
values, offsets, optional validity and alignment padding, and retain each
allocation’s charge through Arrow sharing and JVM exports.
- Reduce kernel overhead by borrowing scalar inputs, using constant-size
sizing scratch, allocating the final output capacity once, inlining small
input-access helpers, and copying previously generated rows for larger repeated
scalar outputs.
- Generate and admit the full requested batch for scalar-only calls. Prevent
constant folding into a list scalar whose later expansion could bypass
admission.
- Check the captured Spark task context during sizing and generation so
cancellation releases partially constructed output.
- Extend behavioural tests, benchmark coverage and documentation. Native
eligibility remains limited to integral literal/column arguments; temporal
sequences continue to use the dispatcher.
The allowance fails immediately when a request does not fit; it does not
spill or wait for another task to release memory. Existing Spark row-error
behaviour and the batch list-offset ceiling are preserved.
And LLM was used for for exploring potential solutions and implementing
these commits.
## How are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are
they covered by existing tests)?
-->
Native tests: 18 kernel/memory tests and the scalar-folding registration
regression pass in release mode. Coverage includes all four integral widths,
null/scalar inputs, arithmetic boundaries, error precedence,
admission/refusal/recovery, concurrent reservations, clone/slice/FFI lifetimes,
and cancellation/unwind cleanup.
JVM tests: nine targeted tests pass on Spark 4.1.3, covering executor
configuration, allocation refusal and recovery, native/fallback routing,
captured-task cancellation, and the existing sequence.sql fixture.
Build and checks: native release build, workspace all-target Clippy with
warnings denied, Rust formatting, Spotless and whitespace checks pass.
Before merge: the broader Spark SQL suite still needs the
run-spark-4.1-tests CI gate.
### Performance
`CometSequenceBenchmark` completed on Spark 4.1.3, Scala 2.13.17 and Java 17
with `local[1]` and 8,192 rows. Integral endpoints were materialised Parquet
columns. The benchmark explicitly raises the sequence allowance to **4 GiB** to
measure large outputs.
These are Spark Benchmark best/average timings, **not medians**:
| Shape | Spark best / average (ms) | Comet best / average (ms) |
|---|---:|---:|
| 2 elements | 13 / 17 | 7 / 9 |
| 5 elements | 10 / 13 | 5 / 7 |
| 365 elements | 13 / 14 | 7 / 8 |
| 1,000 elements | 17 / 18 | 9 / 11 |
| 10,000 elements | 67 / 69 | 39 / 40 |
| 50,000 elements | 270 / 272 | 167 / 169 |
| Descending, 365 elements | 12 / 13 | 6 / 8 |
| Explicit step 7 | 10 / 11 | 5 / 6 |
| Sparse nulls, 365 elements | 12 / 13 | 6 / 8 |
| Date dispatcher control | 11 / 13 | 16 / 18 |
Comet is faster on all integral shapes in this run. These are fresh
measurements from the current harness and Spark version; absolute timings are
not directly comparable with the original issue’s report. Concurrent-task peak
RSS was not measured.
The native performance gains compared to the main branch come from
simplifying the sizing and generation loops: values are written directly into
fixed-capacity buffers, scalar inputs are borrowed instead of expanded into
temporary arrays, and small value, null and length helpers are inlined to avoid
repeated calls and intermediate structures. Larger repeated scalar sequences
also reuse the first generated row through bulk copying. The largest
improvements occur with column-valued endpoints producing hundreds or thousands
of elements per row, where generation costs dominate: 365- and 10,000-element
sequences are approximately 5× faster. Ordinary two- and five-element column
sequences inly improve by about 10%.
--
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]
