GH-3530: Bypass Hadoop codec abstraction to optimize compression performance#3570
Open
iemejia wants to merge 1 commit into
Open
GH-3530: Bypass Hadoop codec abstraction to optimize compression performance#3570iemejia wants to merge 1 commit into
iemejia wants to merge 1 commit into
Conversation
fe836a6 to
0571b9e
Compare
3627fcc to
a661474
Compare
…traction
Compress and decompress Parquet pages by calling the underlying compression
libraries directly, instead of routing every page through Hadoop's
CompressionCodec / CodecPool / CompressionStream machinery.
CodecFactory now provides per-codec byte-oriented implementations:
- SNAPPY -> snappy-java (heap arrays)
- ZSTD -> zstd-jni reusable ZstdCompressCtx/ZstdDecompressCtx (heap arrays)
- LZ4_RAW -> aircompressor LZ4 via reusable direct ByteBuffers
- LZ4 -> aircompressor Hadoop-framed LZ4 streams (legacy codec, distinct from
LZ4_RAW; retained so old files stay readable/writable)
- GZIP -> JDK GZIPOutputStream/GZIPInputStream with 64 KB I/O buffers (the JDK
default of 512 B causes many tiny inflate/deflate calls per page)
- LZO -> aircompressor Hadoop-framed LZO streams
- BROTLI -> brotli4j (runtime-optional; replaces the jitpack brotli-codec and
its non-aarch64 profile)
DirectCodecFactory keeps direct-memory SNAPPY/ZSTD/LZ4_RAW/BROTLI paths. The
legacy Hadoop stream wrappers and the Hadoop fallback in createCompressor/
createDecompressor are gone; unsupported codecs now throw instead of silently
falling back.
Per-column compression (PARQUET-3459) is integrated into the direct-codec
architecture: getCompressor(codecName, level) / createCompressorAtLevel build the
direct compressors at the requested level (with codec-specific level validation).
Backwards compatibility (verified with japicmp against 1.17.0):
- Retain the previously public/protected API as deprecated shims:
CodecFactory.getCodec(CompressionCodecName), CodecFactory.CODEC_BY_NAME, and
DirectCodecFactory.IndirectDecompressor / FullDirectDecompressor /
DirectCodecPool (incl. CodecPool and ParquetCompressionCodecException).
- Keep the commons-pool dependency, used only by those shims.
The default read/write path no longer touches these shims or Hadoop codecs.
Buffer strategy: SNAPPY and ZSTD use heap arrays. LZ4_RAW stays on reusable
direct ByteBuffers: isolated per-case benchmarks show this is a trade-off (direct
is faster for compressible/dictionary-encoded pages -- the common case -- while
heap is faster for high-entropy pages), and direct is the better default. Codec
comments document the measured, data-dependent behaviour rather than fixed
speedup multipliers.
Tests:
- TestCompressionInterop: end-to-end round trips proving files written by the
Hadoop codec path read correctly via the direct path and vice versa, for every
codec (including the legacy LZ4), including multi-row-group cases.
- Expanded TestDirectCodecFactory coverage (incl. per-column leveled paths).
Benchmarks (focused, covering all codecs, avoiding over-measurement):
- CompressionBenchmark isolates the compress/decompress hot path across all
supported codecs, page sizes and heap/direct factories. Its data shapes are
real Parquet pages: realistic typed values are run through Parquet's own
ValuesWriter encoders, so the compressor sees exactly the encoded bytes it would
in production. Shapes are chosen by distinct byte distribution (not an
exhaustive type x encoding matrix) and cover every non-deprecated encoding --
PLAIN, dictionary+RLE, DELTA_BINARY_PACKED, DELTA_BYTE_ARRAY,
DELTA_LENGTH_BYTE_ARRAY, BYTE_STREAM_SPLIT and RLE -- over int32/int64/float/
double/FLBA/binary/boolean, including the PLAIN dictionary-page layout. The class
documents the byte-distribution selection criterion and the combinations
deliberately skipped as redundant, plus JMH per-fork isolation (warning against
single-JVM timing loops that cross-contaminate JIT).
- Modernized the end-to-end WriteBenchmarks/ReadBenchmarks: replaced the
per-(codec,block,page) hand-written methods (which only covered UNCOMPRESSED,
SNAPPY, GZIP, LZO) with a single @Param(codec) method covering every codec, and
removed the orphaned BenchmarkFiles constants. WriteBenchmarks also reports the
resulting compressed file size via @AuxCounters. ReadBenchmarks and the
DataGenerator "generate" CLI share one per-codec corpus, so a pre-built corpus
is reused across read runs (cleanup remains available via the CLI and run.sh).
Results (isolated CompressionBenchmark; JMH 1.37, JDK 17, Linux x86_64, HEAP
factory, 1 MB pages; data-dependent, so figures are representative). Throughput
vs the Hadoop path, geomean over the 16 encoding shapes (>1 = faster):
codec compress decompress
SNAPPY 1.38x 1.25x
ZSTD 1.00x 1.18x
GZIP 1.00x 1.15x
LZ4_RAW 0.96x 1.06x
Snappy peaks near 2.9x compress / 1.8x decompress on small, highly compressible
pages where wrapper/pool overhead dominated. Zstd and Gzip decompress are
uniformly faster (the Gzip 64 KB buffer turns a prior large-page decompress
regression into a win). LZ4_RAW carries a small compress trade-off on
high-throughput pages. LZ4, Brotli and LZO additionally work without native
Hadoop codecs or GPL dependencies.
End-to-end write/read is roughly on par because compression is only a slice of
the pipeline: on the WriteBenchmarks/ReadBenchmarks dataset it is ~1% of write /
~0% of read for Snappy, ~3%/~3% for LZ4_RAW, ~10%/~2% for Zstd, and ~46%/~25% for
Gzip -- so the codec-level gains translate into a small end-to-end change for the
fast codecs and leave more headroom for the CPU-heavy ones.
a661474 to
dd69372
Compare
Member
Author
|
Hi @wgtmac, Sorry to bother you. I noticed you merged PARQUET-3459, so I thought you might be a good person to review this PR given your familiarity with Parquet's compression paths. It is a bit of a big PR but I tried to do my best to explain it clearly and document it in detail. The benchmarks look promising. Besides the performance improvements, this change also reduces direct Hadoop dependencies, which has been discussed as an important maintenance goal by @julienledem in Parquet maintainer meetings. Of course I am not removing those because of backwards compatibility but it will be easy to do if we ever decide to. If you have a chance to take a look, I'd really appreciate it. It's been waiting for review for quite some time, and I'd value your feedback. Thanks! |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Part of #3530 — Apache Parquet Java Performance Improvements
Summary
Bypass the Hadoop
CompressionCodec/CodecPool/ stream-wrapper machinery for all supported codecs, compressing/decompressing each page by calling the underlying library directly, in bothCodecFactory(heap) andDirectCodecFactory(direct memory).SnappyCodecstream wrappersZstdOutputStream/ZstdInputStreamZstdCompressCtx/ZstdDecompressCtxone-shotLz4Codec(native)GzipCodec+ codec-poolGZIPOutputStream/GZIPInputStream, 64 KB bufferscom.hadoop.compression.lzo.LzoCodecLzoHadoopStreams(Apache 2.0, wire-compatible)brotli-codec(x86-only)brotli4j(aarch64 + x86, reflection-loaded, runtime-optional)Per-column compression levels (PARQUET-3459) are integrated into the direct path. Unsupported codecs now throw instead of silently falling back. japicmp-verified with deprecated shims retained.
Portability / licensing: LZ4, Brotli, LZO now work without native Hadoop codecs or GPL deps (previously threw
native lz4 library not available/ClassNotFoundException: BrotliCodec/LzoCodec); Brotli gains aarch64.Benchmarks
CompressionBenchmarkisolates the codec hot path per codec / page size / heap-direct factory. Data shapes are real Parquet pages — realistic typed values run through Parquet's ownValuesWriterencoders — so the compressor sees production bytes. 16 shapes span every non-deprecated encoding (PLAIN, dictionary+RLE, DELTA_BINARY_PACKED, DELTA_BYTE_ARRAY, DELTA_LENGTH_BYTE_ARRAY, BYTE_STREAM_SPLIT, RLE) across int32/int64/float/double/FLBA/binary/boolean; selection is by distinct byte distribution (documented in-class).WriteBenchmarks/ReadBenchmarks— end-to-end 1 M-row write/read per codec; write also reports file size via@AuxCounters.Results
Isolated
CompressionBenchmark(JMH 1.37, JDK 17, Linux x86_64, HEAP, 1 MB pages). Throughput vs the Hadoop path, geomean over the 16 encoding shapes (>1 = faster):How much of end-to-end write/read is compression? Measured on the
WriteBenchmarks/ReadBenchmarksdataset, the codec accounts for roughly:so the isolated codec gains translate into a small end-to-end impact for the fast codecs, with more room to move on GZIP- and ZSTD-heavy workloads.