byronwang123 opened a new pull request, #7636:
URL: https://github.com/apache/kyuubi/pull/7636
### Why are the changes needed?
Kyuubi's Arrow result format speeds up query result delivery through
efficient columnar encode/decode, but it does not reduce the number of bytes
transferred over the wire. Benchmarks on a real cluster (TPC-DS SF10,
34-column `catalog_sales`) show the Server -> Client fetch is the dominant
cost: on a cross-subnet link the throughput is only ~19-25 MB/s and the fetch
phase
accounts for 60-97% of the total query time. Compressing the Arrow IPC body
directly attacks that bottleneck.
This carries on #3877 (relevant issue #3865), which was deferred in 2022 to
wait for an upstream Arrow compression approach. Spark 4.1.0 has since
introduced `spark.sql.execution.arrow.compression.codec` /
`spark.sql.execution.arrow.compression.zstd.level` (SQLConf,
`.version("4.1.0")`), so the upstream-aligned way is now available.
Design goals that keep this small and upstream-consistent:
- Reuse the Spark upstream configuration keys verbatim. On Spark < 4.1 (where
the keys are not yet registered as typed SQLConf entries) they are read as
raw
strings via `getConfString`, so no Kyuubi-private config key is introduced.
- No Kyuubi-specific wire protocol: the batch payload stays standard Arrow
IPC.
The codec is carried by the Arrow `ArrowRecordBatch` body compression
metadata, and the JDBC client decompresses via a
`CompressionCodec.Factory` driven by that metadata -- no server-side hint on
the wire.
- The zstd codec is constructed directly with the configured level
(`new ZstdCompressionCodec(level)`) rather than through the type-enum
factory overload, which would silently drop the level. This mirrors the latest
Spark `ArrowCompressionUtils` behavior (SPARK-57383).
- First version exposes `none` (default, zero behavior change) and `zstd`
only.
LZ4 is intentionally not included: Arrow 16's built-in
`Lz4CompressionCodec`
(commons-compress pure-Java streaming) took 162 s to compress a 670 KB
batch vs 19 ms for zstd in our benchmark, so it is unusable on this runtime.
Incremental result parallel production is a separate, known bottleneck
(`RDD.toLocalIterator` is partition-serial) and is intentionally out of
scope of
this PR. Compression also significantly reduces the memory footprint of
materialized Arrow results (e.g. 4.67 GB -> 850 MB), which may enable
follow-up improvements for incremental result production in a later PR.
### How was this patch tested?
Unit / integration tests (added in this PR):
- `KyuubiArrowConvertersSuite`: the configured zstd level is honored (level
1 vs
19 produce different bytes) and compressed batches round-trip through a
compression-aware `VectorLoader`; slice() of a compressed head batch when
a LIMIT cuts inside it.
- `SparkDatasetHelperSuite`: the session conf flows into the produced IPC
batches and the emitted batch actually carries `CompressionType.ZSTD` metadata.
- `SparkArrowbasedOperationSuite`: end-to-end over JDBC -- a zstd result set
round-trips, and a `LIMIT 150` with `maxRecordsPerBatch=100` (so the tail
batch is sliced from 100 down to 50 rows) round-trips correctly, exercising the
compression-aware loader/unloader pairing on a compressed, sliced batch.
Commands:
```
build/mvn test -pl externals/kyuubi-spark-sql-engine -am -Dtest=none
-DwildcardSuites=org.apache.spark.sql.execution.arrow.KyuubiArrowConvertersSuite,org.apache.spark.sql.kyuubi.SparkDatasetHelperSuite
\
build/mvn test -pl externals/kyuubi-spark-sql-engine -am -Dtest=none
-DwildcardSuites=org.apache.kyuubi.engine.spark.operation.SparkArrowbasedOperationSuite
\
```
Performance was measured with this PR's build on a real cluster (nntest:
Kyuubi
1.13.0-SNAPSHOT + Spark 3.5.9 + 4-executor YARN, TPC-DS SF10), fetching over
a cross-subnet Thrift connection with a thin client that counts the exact Arrow
IPC bytes returned per batch. Comparing `codec=none` vs `codec=zstd` on the
same query (so the byte delta is the server-side compression, measured end to
end):
- cs_numeric (8 numeric columns, ~1.17 GB): 1,167,279,640 -> 247,522,728
bytes (-78.8%); cross-subnet fetch 19.3 s -> 7.5 s (-61%). Normal collect path.
- cs_full (34-column full table, incremental, ~4.67 GB): 4,674,336,136 ->
849,920,016 bytes (-81.8%); fetch 205.8 s -> 94.1 s (-54%). This path runs
the
codec on the executors (`mapPartitionsInternal`), so it also confirms the
compression dependency reaches the executor classpath.
- `none` codec reproduces the pre-change byte count exactly (cs_numeric none
= 1,167,279,640 bytes, matching the baseline), i.e. zero regression on the
default path. (The Arrow 16 `VectorUnloader(root)` constructor itself
delegates to `VectorUnloader(root, true, NoCompressionCodec.INSTANCE, true)`,
so the `none` path in this PR emits identical bytes.)
Wall-clock depends on the network; the byte reductions do not.
Correctness of this build was verified by the unit and embedded-engine JDBC
end-to-end tests above (42 tests passed in total: KyuubiArrowConvertersSuite,
SparkDatasetHelperSuite, and SparkArrowbasedOperationSuite), and
additionally on
the same YARN cluster: over a Kerberos JDBC/beeline connection with
`spark.sql.execution.arrow.compression.codec=zstd` and
`spark.sql.execution.arrow.maxRecordsPerBatch=100`, a `SELECT ... LIMIT 150`
(tail batch sliced from 100 down to 50 rows) returned exactly 150 rows. The
engine JVM was confirmed (via `jcmd VM.class_hierarchy`) to have loaded
`org.apache.arrow.compression.ZstdCompressionCodec`, i.e. the server actually
compressed rather than silently falling back to `none`. This exercised both
compression code paths -- the driver-side slice in `doCollectLimit` (the
LIMIT
path) and the upstream config plumbing via `getConfString` -- with no
Kyuubi-specific wire hint.
Native library shading: a standalone shaded JDBC fat jar (`java -jar`, no
external classpath) was verified to load zstd-jni native, run the Arrow zstd
IPC
compression path, and pass the allocator-leak check. This directly addresses
the
native-shading concern raised on the original #3877.
Dependency / license:
- `dev/dependencyList` regenerated via `build/dependency.sh --replace`
(adds arrow-compression, zstd-jni, commons-compress, commons-io).
- `LICENSE-binary`: zstd-jni added under the BSD section. Note zstd-jni is
dual-licensed (BSD-2-Clause OR GPL-2-with-classpath-exception); it is
redistributed here under the BSD-2-Clause terms. commons-compress /
commons-io / arrow-compression are Apache-2.0 and covered by the main
license.
--
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]
