You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "pingzh (via GitHub)" <gi...@apache.org> on 2026/09/27 17:40:15 UTC
[PR] perf: copy off-heap strings directly into Arrow [datafusion-comet]
pingzh opened a new pull request, #6281:
URL: https://github.com/apache/datafusion-comet/pull/6281
## Which issue does this PR close?
Closes #5910.
## Rationale for this change
An off-heap Spark `UTF8String` currently passes through `getByteBuffer()`, which allocates a JVM byte array and copies the payload before Arrow copies it into its own buffer. Copying directly into reserved Arrow storage removes that intermediate allocation while retaining independent ownership of the output bytes.
## What changes are included in this PR?
- In ordinary `StringWriter`, reserve the value range with `setValueLengthSafe`, obtain the current buffer address after any reallocation, copy with `writeToMemory`, and mark the value defined. Check offset arithmetic and destination bounds before the raw copy. Heap-backed inputs retain the existing setter path.
- Add byte-level regression coverage for source reuse, null/empty/multibyte/invalid UTF-8 values, heap slices, separate data and metadata growth, retained C Data exports, allocation-failure cleanup, and offset overflow. Register the suite in Linux and macOS CI.
- Add `CometStringWriterBenchmark`, comparing the old setter against the actual writer with explicit off-heap inputs and heap-backed controls, preallocated and growing destinations, allocation counters, and output validation.
## How are these changes tested?
- `make core` and full Maven compilation succeeded. Spotless, Scalastyle, and `dev/ci/check-suites.py` passed.
- All 35 tests in `CometStringWriterSuite` and `CometArrowStreamSuite` passed in each configuration below. Spark profiles were built with JDK 17; Spark 4.x test classes were additionally executed with JDK 21.
| Spark | JDK | Tests |
|---|---|---:|
| 3.4.3 | 17 | 35/35 |
| 3.5.9 | 17 | 35/35 |
| 4.0.4 | 17, 21 | 35/35 each |
| 4.1.3 | 17, 21 | 35/35 each |
Focused Maven command (select the corresponding Spark profile):
```sh
./mvnw test -Pspark-4.1 -Dtest=none -Dsuites=org.apache.spark.sql.comet.execution.arrow.CometStringWriterSuite,org.apache.spark.sql.comet.execution.arrow.CometArrowStreamSuite
```
The full benchmark ran on Spark 4.1.3 / Temurin JDK 17.0.20.1, Linux x86-64, AMD EPYC 9V74. Each case uses a 1-second warmup and at least 1 second of measured writing. Source construction and initial destination allocation are outside the measured region; growth during writing is included.
Preallocated off-heap destinations:
| Value size | Baseline JVM bytes/row | Direct copy JVM bytes/row | Conversion throughput vs baseline |
|---|---:|---:|---:|
| 16 B | 32 | 0 | 1.09x |
| 1 KiB | 1,096 | 0 | 1.82x |
| 64 KiB | 65,608 | 0 | 3.16x |
| 1 MiB | 1,048,648 | 0 | 1.58x |
| Varied (0 B–1 MiB) | 209,163.5 | 0 | 3.47x |
Growing off-heap destinations retain 0.8–81 JVM bytes/row of Arrow growth bookkeeping, while still removing the payload-sized array. Heap-array and heap-slice controls have identical allocation counts and throughput within approximately 2% of baseline across the measured cases. These are conversion microbenchmark results, not end-to-end query speedups, and the JVM allocation counter excludes Arrow's native storage.
A separate JDK 21 JFR allocation run (`--smoke`) attributes sampled baseline byte arrays to `UTF8String.getBytes → getByteBuffer → ByteBufferStringWriter.setValue`; the direct writer has no sampled payload-array allocation stack. The thread-allocation counters above provide the per-row measurement.
Reproduce with:
```sh
make benchmark-org.apache.spark.sql.comet.execution.arrow.CometStringWriterBenchmark
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] perf: copy off-heap strings directly into Arrow [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao merged PR #6281:
URL: https://github.com/apache/datafusion-comet/pull/6281
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] perf: copy off-heap strings directly into Arrow [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6281:
URL: https://github.com/apache/datafusion-comet/pull/6281#discussion_r4116821100
##########
spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowWriters.scala:
##########
@@ -485,9 +485,28 @@ private[arrow] class StringWriter(val valueVector: VarCharVector) extends ArrowF
override def setValue(input: SpecializedGetters, ordinal: Int): Unit = {
val utf8 = input.getUTF8String(ordinal)
- val utf8ByteBuffer = utf8.getByteBuffer
- // todo: for off-heap UTF8String, how to pass in to arrow without copy?
- valueVector.setSafe(count, utf8ByteBuffer, utf8ByteBuffer.position(), utf8.numBytes())
+ if (utf8.getBaseObject == null) {
+ val length = utf8.numBytes()
+ // Null entries may not have offsets yet. Match Arrow's append position before reserving.
+ val start =
+ if (valueVector.getLastSet < 0) 0L
+ else valueVector.getStartOffset(valueVector.getLastSet + 1).toLong
+ require(
+ length >= 0 && start >= 0 && start + length <= Int.MaxValue,
+ "String column exceeds the 32-bit Arrow offset range")
Review Comment:
This pre-check re-derives Arrow's append position to guard the 32-bit offset range, but `setValueLengthSafe` already enforces that bound. `handleSafe` asks `reallocDataBuffer` for `start + length` bytes, and anything past `Int.MaxValue` throws `OversizedAllocationException` before it allocates or changes any state. I tried both versions against Arrow 18.3.0, in this suite's overflow state and with two real 1.1 GiB strings in one vector. Without the pre-check, the off-heap path fails with the same `OversizedAllocationException` as the heap path and the old writer. With it, only off-heap batches fail with `IllegalArgumentException`. So the same oversized batch reports a different error depending on where its strings live. The `length >= 0` half is worth keeping, because Arrow never checks it and a negative length silently writes an end offset below the start. Could this become just `require(length >= 0, ...)`, with the overflow test asserting `OversizedAllocationException` for all t
hree backings? That would keep heap and off-heap on the same failure, and the writer wouldn't need to mirror `handleSafe`'s internals.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] perf: copy off-heap strings directly into Arrow [datafusion-comet]
Posted by "pingzh (via GitHub)" <gi...@apache.org>.
pingzh commented on code in PR #6281:
URL: https://github.com/apache/datafusion-comet/pull/6281#discussion_r4116917860
##########
spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowWriters.scala:
##########
@@ -485,9 +485,28 @@ private[arrow] class StringWriter(val valueVector: VarCharVector) extends ArrowF
override def setValue(input: SpecializedGetters, ordinal: Int): Unit = {
val utf8 = input.getUTF8String(ordinal)
- val utf8ByteBuffer = utf8.getByteBuffer
- // todo: for off-heap UTF8String, how to pass in to arrow without copy?
- valueVector.setSafe(count, utf8ByteBuffer, utf8ByteBuffer.position(), utf8.numBytes())
+ if (utf8.getBaseObject == null) {
+ val length = utf8.numBytes()
+ // Null entries may not have offsets yet. Match Arrow's append position before reserving.
+ val start =
+ if (valueVector.getLastSet < 0) 0L
+ else valueVector.getStartOffset(valueVector.getLastSet + 1).toLong
+ require(
+ length >= 0 && start >= 0 && start + length <= Int.MaxValue,
+ "String column exceeds the 32-bit Arrow offset range")
Review Comment:
@andygrove good call, updated
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] perf: copy off-heap strings directly into Arrow [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6281:
URL: https://github.com/apache/datafusion-comet/pull/6281#issuecomment-5876700817
This is a light fully automated review since there are so many PRs open.
`benchmarks/micro/run.py` discovers every benchmark object in `spark/src/test/scala/org/apache/spark/sql/benchmark/` (`discover_suites`, `run.py:469`) and launches each one as `-Dexec.mainClass=org.apache.spark.sql.benchmark.<name>` (`suite_command`, `run.py:545`). `CometStringWriterBenchmark` sits in that directory but declares `package org.apache.spark.sql.comet.execution.arrow` (`spark/src/test/scala/org/apache/spark/sql/benchmark/CometStringWriterBenchmark.scala:20`). So a default `run.py run` will pick it up and fail it with a `ClassNotFoundException` for `org.apache.spark.sql.benchmark.CometStringWriterBenchmark`. Naming the real class in a `--suites` file doesn't get around it either, because `load_suites` keeps only the simple name and `suite_command` puts the fixed package back on. `CometArrowWriterBenchmark` in the same directory already has this problem. Could this file move to `spark/src/test/scala/org/apache/spark/sql/comet/execution/arrow/` next to `CometStringWriter
Suite`? The `make benchmark-...` command in the docstring would still work, and the runner would stop picking it up. If you'd rather have it in the runner's results, would it make sense to have `discover_suites` read each file's `package` line and pass the full class name to `suite_command`? That would fix `CometArrowWriterBenchmark` at the same time.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org