Skip to content

[FEATURE] Support Zstd compression for Arrow IPC query results #7635

Description

@byronwang123

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?

  • Yes. I would be willing to submit a PR with guidance from the Kyuubi community to improve.
  • No. I cannot submit a PR at this time.

Metadata

Metadata

Assignees

No one assigned

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions