feat(storage): default parquet codec to zstd for flink and spark 3.5+ - #19685
feat(storage): default parquet codec to zstd for flink and spark 3.5+#19685cshuo wants to merge 2 commits into
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #19685 +/- ##
============================================
+ Coverage 77.88% 78.00% +0.11%
- Complexity 33264 33494 +230
============================================
Files 2533 2539 +6
Lines 140342 141058 +716
Branches 16912 17175 +263
============================================
+ Hits 109310 110032 +722
+ Misses 23401 23361 -40
- Partials 7631 7665 +34
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR switches the default Parquet compression codec to ZSTD for Flink and Spark 3.5+, resolving the effective default per-engine in HoodieStorageConfig with a safe GZIP fallback, and threads engine type through the write config, Flink client, metadata write config, and binary-copy clustering fallback. The resolution logic and fallbacks look correct and are well covered by tests; one forward-looking question is noted inline. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A few small nits below, but overall the code is clean and well-documented.
|
|
||
| public static class Builder { | ||
|
|
||
| private EngineType engineType = EngineType.SPARK; |
There was a problem hiding this comment.
🤖 nit: since .parquetCompressionCodec(writeConfig.getParquetCompressionCodec()) is always called explicitly at every callsite that resets the codec (e.g. SparkBinaryCopyClusteringExecutionStrategy), the engineType field in the Builder is effectively only consulted when no explicit value is set. That's the intended fast-path, but worth a brief comment here (or on build()) explaining that the engine default only fires when the codec key is absent from props — could save a future reader some head-scratching.
| static String getDefaultParquetCompressionCodec(EngineType engineType) { | ||
| switch (engineType) { | ||
| case FLINK: | ||
| return ZSTD_COMPRESSION_CODEC; |
There was a problem hiding this comment.
🤖 nit: could you flip the name to ZSTD_DEFAULT_MIN_SPARK_VERSION (or MIN_SPARK_VERSION_FOR_ZSTD_DEFAULT)? The current MIN_SPARK_VERSION_WITH_ZSTD_DEFAULT reads as "minimum Spark version that has a ZSTD default", which is accurate, but the more common Hudi constant style puts the subject concept first — makes scanning constants easier.
There was a problem hiding this comment.
🤖 Line 543: The builder defaults engineType to SPARK, so any caller that builds a HoodieStorageConfig without withEngineType(...) resolves the codec via Spark-runtime reflection, making the effective default zstd on any classpath that happens to have Spark 3.5+, even in a Java/Flink context. All three production call sites are patched here, but would it be safer to default this to GZIP when the engine is not explicitly provided, so a future caller that forgets withEngineType fails safe rather than silently writing zstd?
5 participants
Describe the issue this Pull Request addresses
Closes #19684.
Flink and Spark 3.5+ currently default Parquet base and native log files to GZIP, which provides a good compression ratio but adds significant CPU overhead to the write path. ZSTD offers a better balance of compression throughput and storage efficiency. Spark 3.3/3.4 require a compatibility exception because Hudi's non-vectorized file-group reader can still encounter the PARQUET-2160 off-heap leak when reading ZSTD files.
This implements the engine- and runtime-specific behavior discussed in #19615.
Summary and Changelog
hoodie.parquet.compression.codectozstd.HoodieStorageConfigaccording to the engine:zstdzstdgzipgzipHoodieWriteConfig, Flink write-client construction, metadata table-service configuration, and Spark binary-copy clustering configuration.com.github.luben:zstd-jnito the Flink bundle shade includes.zstd-jniinstead of packaging another copy in Spark bundles.Impact
This changes the default compression codec for newly written Parquet base files and native Parquet log blocks when using Flink or Spark 3.5 and newer. Existing files remain readable, and tables may contain files using different compression codecs.
Risk Level
Medium.
The default codec changes the physical encoding of newly written Parquet files for Flink and Spark 3.5+ workloads. The risk is mitigated by retaining GZIP for older Spark and Java runtimes, preserving explicit user configuration, packaging the required JNI dependency in the Flink bundle, and validating the engine-specific defaults and derived configuration paths.
Documentation Update
Updated the root build documentation, Spark datasource documentation, and configuration description to document the engine/version-specific defaults and the PARQUET-2160 risk for Spark 3.3 and 3.4.
Contributor's checklist