Code of Conduct
Search before asking
Describe the feature
Kyuubi supports Arrow-based query results to reduce serialization and deserialization overhead. However, the Arrow IPC batches are currently transferred without compression, so large result sets can still incur significant network transfer costs.
This follows up on #3877 (related issue #3865), which was deferred in 2022 while the community waited for an upstream Spark direction for Arrow compression.
Spark 4.1.0 has since introduced the upstream Arrow compression configurations:
spark.sql.execution.arrow.compression.codec
spark.sql.execution.arrow.compression.zstd.level
This provides an upstream-aligned approach that Kyuubi can reuse instead of introducing a Kyuubi-specific compression configuration or wire protocol.
Motivation
In a cross-subnet cluster benchmark, enabling Zstd significantly reduced the Arrow IPC payload size and result fetch time:
- 8 numeric columns: 1,167,279,640 B -> 247,522,728 B (78.8% fewer bytes); fetch time 19.3 s -> 7.5 s.
- 34 columns: 4,674,336,136 B -> 849,920,016 B (81.8% fewer bytes); fetch time 205.8 s -> 94.1 s.
The exact timing depends on the network and workload, while the byte counts above are the Arrow IPC payload bytes returned by the server.
Describe the solution
Reuse Spark upstream Arrow compression semantics:
- Reuse
spark.sql.execution.arrow.compression.codec and
spark.sql.execution.arrow.compression.zstd.level.
- Keep
none as the default, preserving the existing behavior.
- Support Zstd compression initially.
- Use standard Arrow IPC body compression metadata instead of introducing
Kyuubi-specific wire metadata.
- Let the JDBC client decode compressed batches through Arrow's
CompressionCodec.Factory.
- Preserve compression when Arrow batches are sliced and re-serialized.
- Package the required Arrow compression implementation with the Spark engine
for runtimes that do not provide it, so users do not need to configure
additional runtime jars.
For Spark versions where the upstream configuration entries are not registered
as typed SQLConf entries, read the same configuration names through
getConfString to keep the configuration semantics consistent.
Incremental result production scheduling/prefetch and additional compression
codecs are intentionally out of scope for this change.
Additional context
The previous implementation in #3877 was deferred pending an upstream Spark approach rather than because Arrow compression itself was considered unnecessary. This proposal revisits the feature now that Spark upstream has established Arrow IPC compression configuration and semantics.
Are you willing to submit PR?
Code of Conduct
Search before asking
Describe the feature
Kyuubi supports Arrow-based query results to reduce serialization and deserialization overhead. However, the Arrow IPC batches are currently transferred without compression, so large result sets can still incur significant network transfer costs.
This follows up on #3877 (related issue #3865), which was deferred in 2022 while the community waited for an upstream Spark direction for Arrow compression.
Spark 4.1.0 has since introduced the upstream Arrow compression configurations:
spark.sql.execution.arrow.compression.codecspark.sql.execution.arrow.compression.zstd.levelThis provides an upstream-aligned approach that Kyuubi can reuse instead of introducing a Kyuubi-specific compression configuration or wire protocol.
Motivation
In a cross-subnet cluster benchmark, enabling Zstd significantly reduced the Arrow IPC payload size and result fetch time:
The exact timing depends on the network and workload, while the byte counts above are the Arrow IPC payload bytes returned by the server.
Describe the solution
Reuse Spark upstream Arrow compression semantics:
spark.sql.execution.arrow.compression.codecandspark.sql.execution.arrow.compression.zstd.level.noneas the default, preserving the existing behavior.Kyuubi-specific wire metadata.
CompressionCodec.Factory.for runtimes that do not provide it, so users do not need to configure
additional runtime jars.
For Spark versions where the upstream configuration entries are not registered
as typed SQLConf entries, read the same configuration names through
getConfStringto keep the configuration semantics consistent.Incremental result production scheduling/prefetch and additional compression
codecs are intentionally out of scope for this change.
Additional context
The previous implementation in #3877 was deferred pending an upstream Spark approach rather than because Arrow compression itself was considered unnecessary. This proposal revisits the feature now that Spark upstream has established Arrow IPC compression configuration and semantics.
Are you willing to submit PR?