Use ZSTD as the default compression codec for Parquet files

[cshuo]

Summary

Hudi MOR tables now use native Parquet log files by default, regardless of whether the LSM file-group layout is enabled.

Native log files currently use GZIP compression by default. GZIP provides a good compression ratio, but its relatively low compression throughput adds considerable CPU overhead to the MOR write path.

This discussion proposes using ZSTD as the default compression codec for native Parquet log files.

Motivation

For MOR tables, updates are first written to native log files. Readers merge base files and log files, and compaction eventually rewrites the file slice into a new base file.

Native log files are therefore intermediate files in the MOR write lifecycle. Compression throughput and CPU efficiency are particularly important because the data will later be rewritten during compaction.

The main codec choices are:

Codec	Compression speed	Compression ratio
GZIP	Low	High
Snappy	High	Relatively low
ZSTD	High	Higher than Snappy and often close to GZIP

ZSTD appears to provide a better balance between write throughput, storage size, and read I/O (Benchmark)

Related Hudi discussions include:

• Evaluate rebasing Hudi's default compression from Gzip to Zstd
• Should we change default compression codec for Parquet files to snappy?

Adoption in other table formats

• Apache Iceberg changed the default Parquet compression codec for new tables from GZIP to ZSTD in Iceberg 1.4.0. Existing tables retained their previous behavior to minimize compatibility impact (Iceberg PR #8593).

• Delta Lake has adopted ZSTD as the protocol-level default. However, its Spark implementation is still transitioning and has not yet fully switched from SNAPPY to ZSTD.

Known ZSTD off-heap memory issue

Older parquet-java versions have a known ZSTD off-heap memory issue:

• PARQUET-2160
• parquet-java PR #982

The old Parquet decompression path did not explicitly close ZstdDecompressorStream. Because the stream holds native resources allocated through zstd-jni, long-running jobs could experience off-heap memory growth or fragmentation and eventually run out of memory.

The issue was fixed in parquet-java 1.13.0.

Spark also introduced a workaround in SPARK-41952, released in Spark 3.2.4, 3.3.3, and 3.4.0.

However, the Spark workaround only covers the vectorized Parquet reader. Hudi's file-group reader uses the non-vectorized reader for row-based record merging, so profiles using parquet-java 1.12.x may still be affected.

This is currently relevant to the Spark 3.3 and Spark 3.4 Hudi profiles.

Proposal

Change the default compression codec for native Parquet log files from GZIP to ZSTD.

For dependency profiles using parquet-java 1.12.x, Hudi should add a scoped workaround equivalent to PARQUET-2160 in its non-vectorized file-group read path.

Profiles using parquet-java 1.13.0 or newer should continue using the upstream implementation.

The workaround would only protect Hudi's own file-group reader. It would not attempt to modify arbitrary non-vectorized Parquet reads outside Hudi.

Runtime dependency

Parquet's ZstandardCodec depends on zstd-jni.

Before making ZSTD the default, Hudi should verify that supported runtime environments provide a compatible zstd-jni dependency.

Where the runtime does not guarantee this dependency, the corresponding Hudi bundle should package it explicitly.

[vinothchandar]

As long as its stable - we can switch.. Have others switched?

We hit the same memory leak and rolled back. But, with higher spark /parquet version - it seems to work well.

[cshuo]

Iceberg changed the default Parquet compression codec for new tables from GZIP to ZSTD in Iceberg 1.4.0. Existing tables retained their previous behavior to minimize compatibility impact (apache/iceberg#8593).
Delta Lake has adopted ZSTD as the protocol-level default. However, its Spark implementation is still transitioning and has not yet fully switched from SNAPPY to ZSTD.

[vinothchandar]

Can we do it such that zstd is default, but for lower/problematic spark versions - we override snappy..

[cshuo]

The memory leak occurs on the reader side, so overriding the write codec to Snappy for older Spark versions would not fully address it. Older Spark versions may still read ZSTD Parquet files written by newer engines.

A more consistent approach would be to use ZSTD as the default across versions and apply a reader-side workaround for the affected older Spark/Parquet versions, similar to Iceberg’s fix before upgrading to Parquet 1.13.0: apache/iceberg#5681.

[rangareddy]

+1 to this proposal. Switching native Parquet log files from GZIP to ZSTD makes complete sense for MOR workloads.

Since log files are intermediate representations that undergo frequent write and compaction cycles, prioritizing compression throughput and CPU efficiency without a significant penalty on compression ratio is a huge win for MOR write latency.

A few thoughts on the implementation & safeguard details:

• Off-Heap Safeguard (PARQUET-2160): Porting a scoped fix for Hudi's non-vectorized file-group reader on older parquet-java 1.12.x profiles (Spark 3.3/3.4) is critical to avoid off-heap memory leaks in long-running streaming/compaction jobs.

• zstd-jni Bundle Verification: Ensuring zstd-jni is shaded/packaged cleanly in hudi-spark-bundle across supported Spark versions will prevent missing dependency runtime errors.

• Benchmarking: It would be great to run a quick benchmark on standard MOR workloads (e.g., 100GB update-heavy ingestion) comparing GZIP vs ZSTD to measure:

   1. MOR write throughput / CPU cycles saved.
   2. Log file size delta.
   3. Compaction read/write performance.
  

Thanks for putting together a detailed proposal!

[cshuo]

Thanks for the feedback. I agree that these safeguards should be addressed before changing the default.
Before upgrading parquet-java to 1.13.0, Iceberg introduced a scoped workaround for PARQUET-2160 to ensure ZSTD decompression streams are closed properly. We can follow a similar approach in Hudi for the affected Parquet 1.12.x versions: apache/iceberg#5681.

[wombatu-kun]

On "have others switched?" - Delta and Spark default to snappy, Iceberg and Paimon to zstd; nobody stayed on gzip. Iceberg's switch is worth a look though: since 1.4.0 the zstd default is only stamped into the properties of newly created tables, so existing tables keep writing gzip on a library upgrade (iceberg#15236). So one thing to decide explicitly: does the new default apply to every table on the next commit after a jar upgrade, or only to new ones? Hudi persists nothing about the codec in HoodieTableConfig today, so "new tables only" is not free.

On overriding to snappy for the lower Spark profiles - that does not mitigate the leak. PARQUET-2160 is on the decompression path (CodecFactory.HeapBytesDecompressor), so what matters is which parquet-java version reads a zstd file, not which one wrote it. A Spark 3.3 job reading a table that a Spark 3.5 or Flink writer filled with zstd hits the same leak, and a file stays zstd once written. Only the read-side workaround protects anyone - and the profiles that would carry it are the least covered ones: bot.yml runs only spark3.5 and spark4.2 today.

On scope: hoodie.parquet.compression.codec is a single knob shared by base files and native Parquet log blocks (HoodieParquetDataBlock, HoodieFlinkParquetDataBlock, RowDataInlineLogWriteHandle and HoodieInlineLogAppendHandle all pass writeConfig.getParquetCompressionCodec()). So this means either a new log-file-specific config or flipping the shared default for base files too - different changes, and the base-file case is the one #13568 and #14938 already argued with numbers.

Two smaller notes: zstd-jni reaches us only transitively via parquet-hadoop and is not in the Flink bundle's shade list; and hoodie.logfile.to.parquet.compression.ratio (0.35) is calibrated against today's codec, so file sizing drifts if log files go zstd while base files stay gzip.

[cshuo]

On "have others switched?" - Delta and Spark default to snappy, Iceberg and Paimon to zstd; nobody stayed on gzip. Iceberg's switch is worth a look though: since 1.4.0 the zstd default is only stamped into the properties of newly created tables, so existing tables keep writing gzip on a library upgrade (apache/iceberg#15236). So one thing to decide explicitly: does the new default apply to every table on the next commit after a jar upgrade, or only to new ones? Hudi persists nothing about the codec in HoodieTableConfig today, so "new tables only" is not free.

I would lean toward applying the new default to all tables after upgrading the Hudi version, here are some thoughts:

• the compression codec is a Parquet file-level property recorded in each file’s metadata, so mixing files written with different codecs does not introduce read/write compatibility issues.
• most production deployments already use ZSTD or explicitly configure another codec instead of relying on the GZIP default, so changing the default would not actually affect either of these scenarios.

cc @vinothchandar @danny0405 @yihua WDYT?

On overriding to snappy for the lower Spark profiles - that does not mitigate the leak. PARQUET-2160 is on the decompression path (CodecFactory.HeapBytesDecompressor), so what matters is which parquet-java version reads a zstd file, not which one wrote it. A Spark 3.3 job reading a table that a Spark 3.5 or Flink writer filled with zstd hits the same leak, and a file stays zstd once written. Only the read-side workaround protects anyone - and the profiles that would carry it are the least covered ones: bot.yml runs only spark3.5 and spark4.2 today.

Agree, For affected older Spark/Parquet profiles, we should apply a reader-side workaround similar to Iceberg’s PARQUET-2160 fix, rather than changing their writer codec.

[cshuo]

On scope: hoodie.parquet.compression.codec is a single knob shared by base files and native Parquet log blocks (HoodieParquetDataBlock, HoodieFlinkParquetDataBlock, RowDataInlineLogWriteHandle and HoodieInlineLogAppendHandle all pass writeConfig.getParquetCompressionCodec()). So this means either a new log-file-specific config or flipping the shared default for base files too - different changes, and the base-file case is the one #13568 and #14938 already argued with numbers.

I would prefer introducing a separate compression configuration for native log files. Log files are short-lived intermediate files and will eventually be compacted into base files, so write throughput and CPU efficiency are more important than maximizing compression ratio. Even if both log files and base files ultimately use ZSTD, we may want to use a lower ZSTD level for log files to favor faster compression, while keeping a different level for long-lived base files.

[danny0405]

I would prefer introducing a separate compression configuration for native log files

I would rather not, let's leep the codec consistent to avoid unnecessary size drifts and complexities. See if Delta or Iceberg already got a workaround for the memory leak bug, they must got one right? either a risk declaration on the size or a actual fix in the code.

[cshuo]

@danny0405 I think these are two separate concerns.

• To switch the default to ZSTD, the memory leak in older parquet-java versions must be addressed. As mentioned earlier, we can follow Iceberg’s workaround: apache/iceberg#5681.
• We should also consider MOR log write throughput. In a preliminary benchmark on Alibaba Cloud EMR, for a MOR table with 20 string fields, with ZSTD enabled (default level 3), native log write throughput was only about 70% of inline log throughput. To improve throughput, we need to tune compression parameters specifically for native log files rather than keeping them fully aligned with base files. Actually, different compression policies across levels or file lifecycles are common in LSM storage. RocksDB, TiKV and Paimon support per-level compression, while Cassandra and HBase differentiate compression between flush and compaction.

[vinothchandar]

apache/iceberg#5681 is kind of hacky. Can we just control this cleanly based on parquet write version alone. I expect folks to move to recent Spark/Flink version (or let this be a catalyst)

[vinothchandar]

I would lean toward applying the new default to all tables after upgrading the Hudi version, here are some thoughts:

+1. we can do it in the next major version release.

I would prefer introducing a separate compression configuration for native log files.

sg. but zstd is as light as snappy, anyway?