From dd6937239fa1581e7b845560bff164442548c205 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Tue, 14 Jul 2026 13:06:47 +0200 Subject: [PATCH] GH-3530: Optimize compression by bypassing the Hadoop codec abstraction 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. --- parquet-benchmarks/pom.xml | 15 + parquet-benchmarks/run.sh | 12 +- .../parquet/benchmarks/BenchmarkFiles.java | 13 - .../benchmarks/CompressionBenchmark.java | 444 +++++++ .../parquet/benchmarks/DataGenerator.java | 120 +- .../parquet/benchmarks/ReadBenchmarks.java | 129 +-- .../parquet/benchmarks/WriteBenchmarks.java | 159 +-- .../apache/parquet/bytes/TestBytesInput.java | 37 + parquet-hadoop/pom.xml | 37 +- .../apache/parquet/hadoop/CodecFactory.java | 1016 +++++++++++++---- .../parquet/hadoop/DirectCodecFactory.java | 373 ++++-- .../hadoop/TestCompressionInterop.java | 713 ++++++++++++ .../hadoop/TestDirectCodecFactory.java | 441 ++++++- .../parquet/hadoop/TestZstandardCodec.java | 2 - pom.xml | 3 +- 15 files changed, 2817 insertions(+), 697 deletions(-) create mode 100644 parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/CompressionBenchmark.java create mode 100644 parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestCompressionInterop.java diff --git a/parquet-benchmarks/pom.xml b/parquet-benchmarks/pom.xml index 3ce4dce5ce..1f5aade313 100644 --- a/parquet-benchmarks/pom.xml +++ b/parquet-benchmarks/pom.xml @@ -87,6 +87,12 @@ slf4j-api ${slf4j.version} + + com.aayushatharva.brotli4j + brotli4j + ${brotli4j.version} + runtime + @@ -102,6 +108,9 @@ ${jmh.version} + + org.openjdk.jmh.generators.BenchmarkProcessor + @@ -121,6 +130,12 @@ org.openjdk.jmh.Main + + META-INF/BenchmarkList + + + META-INF/CompilerHints + diff --git a/parquet-benchmarks/run.sh b/parquet-benchmarks/run.sh index d5e9b1656c..bde4e3af48 100755 --- a/parquet-benchmarks/run.sh +++ b/parquet-benchmarks/run.sh @@ -38,6 +38,7 @@ Information on the JMH_OPTIONS can be found by running: run.sh all -help ----------- | ---------- all | Runs all benchmarks in the module (listed here and others). build | (No benchmark run, shortcut to rebuild the JMH uber jar). +generate | (No benchmark run, shortcut to pre-generate the per-codec read corpus). clean | (No benchmark run, shortcut to clean up any temporary files). read | Reading files with different compression, page and block sizes. write | Writing files. @@ -74,9 +75,18 @@ elif [ "$BENCHMARK" == "build" ]; then elif [ "$BENCHMARK" == "clean" ]; then - # Shortcut utility to clean any state left behind from any previous run. + # Shortcut utility to clean up any state left behind from any previous run. java -cp ${SCRIPT_PATH}/target/parquet-benchmarks.jar org.apache.parquet.benchmarks.DataGenerator cleanup +elif [ "$BENCHMARK" == "generate" ]; then + + # Shortcut utility to pre-generate the per-codec 1M-row data files reused by the read benchmark. + # Build the uberjar first if it isn't there yet. + if [ ! -f ${SCRIPT_PATH}/target/parquet-benchmarks.jar ]; then + ${SCRIPT_PATH}/run.sh build + fi + java -cp ${SCRIPT_PATH}/target/parquet-benchmarks.jar org.apache.parquet.benchmarks.DataGenerator generate + else # Actually run a benchmark in the JMH harness. diff --git a/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/BenchmarkFiles.java b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/BenchmarkFiles.java index 43b907befe..5883dfdb12 100644 --- a/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/BenchmarkFiles.java +++ b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/BenchmarkFiles.java @@ -27,19 +27,6 @@ public class BenchmarkFiles { public static final String TARGET_DIR = "target/tests/ParquetBenchmarks"; public static final Path targetDir = new Path(TARGET_DIR); - public static final Path file_1M = new Path(TARGET_DIR + "/PARQUET-1M"); - - // different block and page sizes - public static final Path file_1M_BS256M_PS4M = new Path(TARGET_DIR + "/PARQUET-1M-BS256M_PS4M"); - public static final Path file_1M_BS256M_PS8M = new Path(TARGET_DIR + "/PARQUET-1M-BS256M_PS8M"); - public static final Path file_1M_BS512M_PS4M = new Path(TARGET_DIR + "/PARQUET-1M-BS512M_PS4M"); - public static final Path file_1M_BS512M_PS8M = new Path(TARGET_DIR + "/PARQUET-1M-BS512M_PS8M"); - - // different compression codecs - // public final Path parquetFile_1M_LZO = new Path("target/tests/ParquetBenchmarks/PARQUET-1M-LZO"); - public static final Path file_1M_SNAPPY = new Path(TARGET_DIR + "/PARQUET-1M-SNAPPY"); - public static final Path file_1M_GZIP = new Path(TARGET_DIR + "/PARQUET-1M-GZIP"); - // Page checksum files public static final Path file_100K_CHECKSUMS_UNCOMPRESSED = new Path(TARGET_DIR + "/PARQUET-100K-CHECKSUMS-UNCOMPRESSED"); diff --git a/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/CompressionBenchmark.java b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/CompressionBenchmark.java new file mode 100644 index 0000000000..27613d3074 --- /dev/null +++ b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/CompressionBenchmark.java @@ -0,0 +1,444 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.parquet.benchmarks; + +import static org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_1_0; +import static org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_2_0; + +import java.io.IOException; +import java.util.Random; +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.conf.Configuration; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.bytes.DirectByteBufferAllocator; +import org.apache.parquet.bytes.HeapByteBufferAllocator; +import org.apache.parquet.column.ColumnDescriptor; +import org.apache.parquet.column.ParquetProperties; +import org.apache.parquet.column.ParquetProperties.WriterVersion; +import org.apache.parquet.column.values.ValuesWriter; +import org.apache.parquet.column.values.deltalengthbytearray.DeltaLengthByteArrayValuesWriter; +import org.apache.parquet.compression.CompressionCodecFactory; +import org.apache.parquet.hadoop.CodecFactory; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Warmup; + +/** + * Isolated JMH benchmarks for raw Parquet compression and decompression throughput. + * + *

Measures the performance of {@link CompressionCodecFactory.BytesInputCompressor} + * and {@link CompressionCodecFactory.BytesInputDecompressor} for each supported codec, + * comparing the heap-based {@link CodecFactory} path (what all production users take) + * against the direct-memory {@code DirectCodecFactory} path (off-heap ByteBuffers). + * + *

This benchmark isolates the codec hot path from file I/O and other Parquet overhead, + * while still feeding the compressor exactly the bytes production does: each + * {@code dataShape} is a real Parquet data page, produced by running realistic typed + * column values through Parquet's actual {@link ValuesWriter} encoders (PLAIN, + * dictionary+RLE, DELTA_BINARY_PACKED, DELTA_BYTE_ARRAY, BYTE_STREAM_SPLIT, RLE booleans). + * This matters because compressors never see raw values — they see encoded pages, whose byte + * distribution is entirely determined by the encoding. Hand-rolled byte patterns (e.g. 4-byte + * little-endian "dictionary indices" or uniform-random doubles) mis-represent that and skew + * codec-specific results. + * + *

Shape selection — by byte distribution, not by combinatorial matrix. A block + * compressor only ever sees the encoded page bytes, so its behaviour is a function of the + * byte distribution alone. Shapes are therefore chosen to span the distinct byte layouts that real + * Parquet pages exhibit — the {@code (physical type, encoding, value distribution)} combinations + * that produce materially different bytes — rather than an exhaustive type×encoding matrix. + * Two combinations that emit the same layout compress identically, so only one is kept. Every + * non-deprecated data-page encoding is represented at least once. + * + *

Data shapes (type / encoding / semantics): + * + *

    + *
  • {@code INT32_ID_PLAIN} — random 32-bit ids, PLAIN (V1-written files).
    • + *
  • {@code INT32_DELTA} — ascending 32-bit values, DELTA_BINARY_PACKED.
    • + *
  • {@code INT32_ENUM_DICT} — low-cardinality category ids, dictionary + RLE indices.
    • + *
  • {@code INT64_TIMESTAMP_DELTA} — event timestamps, DELTA_BINARY_PACKED.
    • + *
  • {@code FLOAT_MEASURE_PLAIN} — clustered 32-bit readings, PLAIN.
    • + *
  • {@code FLOAT_MEASURE_BSS} — the same 32-bit readings, BYTE_STREAM_SPLIT.
    • + *
  • {@code DOUBLE_MEASURE_PLAIN} — clustered 64-bit readings, PLAIN.
    • + *
  • {@code DOUBLE_MEASURE_BSS} — the same 64-bit readings, BYTE_STREAM_SPLIT.
    • + *
  • {@code FLBA_DECIMAL_PLAIN} — DECIMAL as 16-byte fixed-length, PLAIN.
    • + *
  • {@code STRING_CATEGORY_DICT} — categorical strings, dictionary + RLE indices (data page).
    • + *
  • {@code STRING_DICT_PAGE_PLAIN} — distinct strings, PLAIN — the layout of a dictionary + * page (present in both V1 and V2 files).
    • + *
  • {@code STRING_TEXT_PLAIN} — free text with repetition, PLAIN (V1 data page).
    • + *
  • {@code STRING_TEXT_DELTA} — free text, DELTA_BYTE_ARRAY.
    • + *
  • {@code STRING_DELTA_LENGTH} — free text, DELTA_LENGTH_BYTE_ARRAY.
    • + *
  • {@code BOOLEAN_FLAG_RLE} — mostly-true flag with runs, RLE.
    • + *
  • {@code BINARY_UUID_RANDOM} — random 16-byte ids (hashes/uuids); compression floor.
    • + *
+ * + *

Deliberately not tested (byte-distribution-redundant, or deprecated). The following + * would not add a distinct byte layout, so testing them would only inflate the matrix: + * + *

    + *
  • Dictionary data pages for other types (INT64/FLOAT/DOUBLE/FLBA · RLE_DICTIONARY): + * a dictionary data page is RLE/bit-packed integer indices whose layout depends on the + * dictionary cardinality, not the value type — byte-identical to + * {@code INT32_ENUM_DICT} / {@code STRING_CATEGORY_DICT}.
    • + *
  • PLAIN_DICTIONARY (legacy V1 dictionary): its data page is identical to + * RLE_DICTIONARY; only the encoding id and the dictionary page differ (the latter's PLAIN + * layout is covered by {@code STRING_DICT_PAGE_PLAIN}).
    • + *
  • FIXED_LEN_BYTE_ARRAY · DELTA_BYTE_ARRAY (the V2 default for FLBA): same encoding + * and layout as {@code STRING_TEXT_DELTA}.
    • + *
  • INT64 · PLAIN: the same little-endian-integer layout as {@code INT32_ID_PLAIN}, + * only wider (extra high-order zero bytes) — no new distribution.
    • + *
  • BIT_PACKED: used for definition/repetition levels only, and deprecated.
    • + *
  • INT96 · PLAIN (deprecated legacy timestamps) and BYTE_STREAM_SPLIT for + * INT32/INT64/FLBA (spec-extended but rare): distinct but low-value today; add them only + * when specifically targeting legacy lakes or BSS-on-integer workloads.
    • + *
+ * + *

Methodology. Results are highly data-dependent, so vary {@code dataShape}. Compare + * codecs/paths only within the same {@code (codec, pageSize, dataShape)} cell. {@code pageSize} + * is the target for the volume of buffered values; the resulting encoded page (what is actually + * compressed) may be smaller, especially for dictionary/RLE shapes. JMH runs each {@code @Param} + * combination in its own forked JVM ({@code @Fork}); do not substitute ad-hoc single-JVM + * timing loops that iterate over codecs/shapes, which cross-contaminate JIT profiles and produce + * misleading deltas. The full parameter matrix is large; run targeted subsets via the JMH CLI, + * e.g. {@code -p codec=ZSTD -p dataShape=DOUBLE_MEASURE_PLAIN -p pageSize=1048576}. + */ +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.SECONDS) +@Fork(2) +@Warmup(iterations = 3, time = 1) +@Measurement(iterations = 5, time = 1) +@State(Scope.Thread) +public class CompressionBenchmark { + + @Param({"SNAPPY", "ZSTD", "LZ4_RAW", "LZ4", "GZIP", "BROTLI", "LZO"}) + public String codec; + + @Param({"65536", "131072", "262144", "1048576"}) + public int pageSize; + + @Param({"HEAP", "DIRECT"}) + public String factoryType; + + @Param({ + "INT32_ID_PLAIN", + "INT32_DELTA", + "INT32_ENUM_DICT", + "INT64_TIMESTAMP_DELTA", + "FLOAT_MEASURE_PLAIN", + "FLOAT_MEASURE_BSS", + "DOUBLE_MEASURE_PLAIN", + "DOUBLE_MEASURE_BSS", + "FLBA_DECIMAL_PLAIN", + "STRING_CATEGORY_DICT", + "STRING_DICT_PAGE_PLAIN", + "STRING_TEXT_PLAIN", + "STRING_TEXT_DELTA", + "STRING_DELTA_LENGTH", + "BOOLEAN_FLAG_RLE", + "BINARY_UUID_RANDOM" + }) + public String dataShape; + + private byte[] uncompressedData; + private byte[] compressedData; + private int decompressedSize; + + private CompressionCodecFactory.BytesInputCompressor compressor; + private CompressionCodecFactory.BytesInputDecompressor decompressor; + private CodecFactory factory; + + @Setup(Level.Trial) + public void setup() throws IOException { + uncompressedData = buildEncodedPage(dataShape, pageSize, 42L); + decompressedSize = uncompressedData.length; + + Configuration conf = new Configuration(); + if ("DIRECT".equals(factoryType)) { + factory = CodecFactory.createDirectCodecFactory(conf, DirectByteBufferAllocator.getInstance(), pageSize); + } else { + factory = new CodecFactory(conf, pageSize); + } + CompressionCodecName codecName = CompressionCodecName.valueOf(codec); + + compressor = factory.getCompressor(codecName); + decompressor = factory.getDecompressor(codecName); + + // Pre-compress for decompression benchmark; copy to a stable byte array + // since the compressor may reuse its internal buffer. + BytesInput compressed = compressor.compress(BytesInput.from(uncompressedData)); + compressedData = compressed.toByteArray(); + } + + @TearDown(Level.Trial) + public void tearDown() { + factory.release(); + } + + @Benchmark + public BytesInput compress() throws IOException { + return compressor.compress(BytesInput.from(uncompressedData)); + } + + @Benchmark + public byte[] decompress() throws IOException { + // Force materialization of the decompressed data. toByteArray() is essentially + // free for our optimized implementations (returns the existing byte[]). + return decompressor + .decompress(BytesInput.from(compressedData), decompressedSize) + .toByteArray(); + } + + // --------------------------------------------------------------------------- + // Realistic page generation: real typed values -> real Parquet encoders. + // --------------------------------------------------------------------------- + + /** + * Builds a real Parquet data page for the given {@code shape} by writing realistic typed + * values through Parquet's own {@link ValuesWriter}, then returning the encoded page bytes + * (exactly what the compressor sees in production). {@code targetSize} bounds the volume of + * buffered values. + */ + private byte[] buildEncodedPage(String shape, int targetSize, long seed) throws IOException { + Random r = new Random(seed); + switch (shape) { + case "INT32_ID_PLAIN": { + // V1 writers emit PLAIN for ints (V2 uses DELTA); both are common in the wild. + ValuesWriter w = newWriter(PARQUET_1_0, PrimitiveTypeName.INT32, 0, false, false); + return fill(w, targetSize, () -> w.writeInteger(r.nextInt())); + } + case "INT32_DELTA": { + ValuesWriter w = newWriter(PrimitiveTypeName.INT32, false, false); + int[] v = {r.nextInt(1000)}; + return fill(w, targetSize, () -> { + v[0] += r.nextInt(16); // small ascending steps (row numbers / offsets) + w.writeInteger(v[0]); + }); + } + case "INT32_ENUM_DICT": { + ValuesWriter w = newWriter(PrimitiveTypeName.INT32, true, false); + int[] cats = distinctInts(r, 40); + return fill(w, targetSize, () -> w.writeInteger(cats[skewed(r, cats.length)])); + } + case "INT64_TIMESTAMP_DELTA": { + ValuesWriter w = newWriter(PrimitiveTypeName.INT64, false, false); + long[] ts = {1_600_000_000_000L}; + return fill(w, targetSize, () -> { + ts[0] += 1000 + r.nextInt(5000); // ~event cadence in millis + w.writeLong(ts[0]); + }); + } + case "FLOAT_MEASURE_PLAIN": { + ValuesWriter w = newWriter(PrimitiveTypeName.FLOAT, false, false); + return fill(w, targetSize, () -> w.writeFloat(measurementF(r))); + } + case "FLOAT_MEASURE_BSS": { + ValuesWriter w = newWriter(PrimitiveTypeName.FLOAT, false, true); + return fill(w, targetSize, () -> w.writeFloat(measurementF(r))); + } + case "DOUBLE_MEASURE_PLAIN": { + ValuesWriter w = newWriter(PrimitiveTypeName.DOUBLE, false, false); + return fill(w, targetSize, () -> w.writeDouble(measurement(r))); + } + case "DOUBLE_MEASURE_BSS": { + ValuesWriter w = newWriter(PrimitiveTypeName.DOUBLE, false, true); + return fill(w, targetSize, () -> w.writeDouble(measurement(r))); + } + case "FLBA_DECIMAL_PLAIN": { + // V2 defaults FLBA to DELTA_BYTE_ARRAY; PLAIN fixed-length (classic decimal storage) is V1. + ValuesWriter w = newWriter(PARQUET_1_0, PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY, 16, false, false); + return fill(w, targetSize, () -> w.writeBytes(decimal16(r))); + } + case "STRING_CATEGORY_DICT": { + ValuesWriter w = newWriter(PrimitiveTypeName.BINARY, true, false); + Binary[] cats = distinctStrings(r, 50, 3, 12); + return fill(w, targetSize, () -> w.writeBytes(cats[skewed(r, cats.length)])); + } + case "STRING_DICT_PAGE_PLAIN": { + // Distinct values in PLAIN layout ([len][bytes], no value repetition) — the byte layout + // of a dictionary page, which is PLAIN-encoded in both V1 and V2 files. + ValuesWriter w = newWriter(PARQUET_1_0, PrimitiveTypeName.BINARY, 0, false, false); + return fill(w, targetSize, () -> w.writeBytes(randomToken(r, 4, 16))); + } + case "STRING_TEXT_PLAIN": { + // PLAIN binary data pages only occur under V1 (V2 binary defaults to DELTA_BYTE_ARRAY). + ValuesWriter w = newWriter(PARQUET_1_0, PrimitiveTypeName.BINARY, 0, false, false); + return fill(w, targetSize, () -> w.writeBytes(Binary.fromString(sentence(r)))); + } + case "STRING_TEXT_DELTA": { + ValuesWriter w = newWriter(PrimitiveTypeName.BINARY, false, false); + return fill(w, targetSize, () -> w.writeBytes(Binary.fromString(sentence(r)))); + } + case "STRING_DELTA_LENGTH": { + // DELTA_LENGTH_BYTE_ARRAY is not a factory default; construct it directly. + ValuesWriter w = new DeltaLengthByteArrayValuesWriter( + 64 * 1024, 64 * 1024 * 1024, new HeapByteBufferAllocator()); + return fill(w, targetSize, () -> w.writeBytes(Binary.fromString(sentence(r)))); + } + case "BOOLEAN_FLAG_RLE": { + ValuesWriter w = newWriter(PrimitiveTypeName.BOOLEAN, false, false); + boolean[] state = {true}; + return fill(w, targetSize, () -> { + if (r.nextInt(100) < 5) { + state[0] = !state[0]; // ~5% flip probability -> realistic runs + } + w.writeBoolean(state[0]); + }); + } + case "BINARY_UUID_RANDOM": { + ValuesWriter w = newWriter(PrimitiveTypeName.BINARY, false, false); + return fill(w, targetSize, () -> { + byte[] b = new byte[16]; + r.nextBytes(b); + w.writeBytes(Binary.fromConstantByteArray(b)); + }); + } + default: + throw new IllegalArgumentException("Unknown data shape: " + shape); + } + } + + /** Creates the {@link ValuesWriter} Parquet would use for {@code type} under the given options. */ + private ValuesWriter newWriter(PrimitiveTypeName type, boolean dictionary, boolean byteStreamSplit) { + return newWriter(PARQUET_2_0, type, 0, dictionary, byteStreamSplit); + } + + private ValuesWriter newWriter( + WriterVersion version, + PrimitiveTypeName type, + int typeLength, + boolean dictionary, + boolean byteStreamSplit) { + ParquetProperties props = ParquetProperties.builder() + .withWriterVersion(version) + .withDictionaryEncoding(dictionary) + .withByteStreamSplitEncoding(byteStreamSplit) + .withAllocator(new HeapByteBufferAllocator()) + .withPageSize(64 * 1024 * 1024) // large so the writer never self-limits during fill + .build(); + ColumnDescriptor col = typeLength > 0 + ? new ColumnDescriptor(new String[] {"col"}, type, typeLength, 0, 0) + : new ColumnDescriptor(new String[] {"col"}, type, 0, 0); + return props.newValuesWriter(col); + } + + /** Writes values until the buffered volume reaches {@code targetSize}, then returns the page. */ + private static byte[] fill(ValuesWriter w, int targetSize, Runnable writeOne) throws IOException { + try { + long cap = (long) targetSize * 8L + 1024L; // hard bound so we always terminate + long n = 0; + while (w.getBufferedSize() < targetSize && n < cap) { + writeOne.run(); + n++; + } + return w.getBytes().toByteArray(); + } finally { + try { + w.close(); + } catch (Exception ignored) { + // best-effort release of writer buffers during benchmark setup + } + } + } + + private static int[] distinctInts(Random r, int count) { + int[] out = new int[count]; + for (int i = 0; i < count; i++) { + out[i] = 1000 + r.nextInt(1_000_000); + } + return out; + } + + private static Binary[] distinctStrings(Random r, int count, int minLen, int maxLen) { + Binary[] out = new Binary[count]; + for (int i = 0; i < count; i++) { + out[i] = randomToken(r, minLen, maxLen); + } + return out; + } + + /** A single random uppercase token of length in {@code [minLen, maxLen]}. */ + private static Binary randomToken(Random r, int minLen, int maxLen) { + int len = minLen + r.nextInt(maxLen - minLen + 1); + StringBuilder sb = new StringBuilder(len); + for (int j = 0; j < len; j++) { + sb.append((char) ('A' + r.nextInt(26))); + } + return Binary.fromString(sb.toString()); + } + + /** Mild power-law skew towards lower indices, as real categorical distributions exhibit. */ + private static int skewed(Random r, int n) { + int idx = (int) (n * r.nextDouble() * r.nextDouble()); + return Math.min(idx, n - 1); + } + + /** Sensor-like reading: Gaussian around 15.0 with 8.0 spread, quantized to 0.1. */ + private static double measurement(Random r) { + return Math.round((15.0 + r.nextGaussian() * 8.0) * 10.0) / 10.0; + } + + /** 32-bit sensor-like reading: Gaussian around 15.0 with 8.0 spread, quantized to 0.1. */ + private static float measurementF(Random r) { + return Math.round((15.0f + (float) r.nextGaussian() * 8.0f) * 10.0f) / 10.0f; + } + + /** + * DECIMAL-like value as a 16-byte big-endian FIXED_LEN_BYTE_ARRAY: a bounded unscaled magnitude + * (e.g. a price in cents) placed in the low bytes, leaving the high bytes zero — the sparse, + * leading-zero layout real decimal columns exhibit. + */ + private static Binary decimal16(Random r) { + long unscaled = (long) (r.nextDouble() * 100_000_000L); + byte[] b = new byte[16]; + for (int i = 0; i < 8; i++) { + b[15 - i] = (byte) (unscaled >>> (8 * i)); + } + return Binary.fromConstantByteArray(b); + } + + private static final String[] WORDS = { + "the", "quick", "brown", "fox", "jumps", "over", "lazy", "dog", "parquet", "column", + "data", "value", "record", "field", "schema", "compression", "encoding", "page", "row", "group" + }; + + /** Natural-language-like sentence of 6-12 words drawn from a fixed vocabulary. */ + private static String sentence(Random r) { + int words = 6 + r.nextInt(7); + StringBuilder sb = new StringBuilder(); + for (int i = 0; i < words; i++) { + if (i > 0) { + sb.append(' '); + } + sb.append(WORDS[r.nextInt(WORDS.length)]); + } + return sb.toString(); + } +} diff --git a/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/DataGenerator.java b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/DataGenerator.java index b3b0df0ace..f6435298db 100644 --- a/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/DataGenerator.java +++ b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/DataGenerator.java @@ -19,30 +19,17 @@ package org.apache.parquet.benchmarks; import static java.util.UUID.randomUUID; -import static org.apache.parquet.benchmarks.BenchmarkConstants.BLOCK_SIZE_256M; -import static org.apache.parquet.benchmarks.BenchmarkConstants.BLOCK_SIZE_512M; import static org.apache.parquet.benchmarks.BenchmarkConstants.BLOCK_SIZE_DEFAULT; import static org.apache.parquet.benchmarks.BenchmarkConstants.DICT_PAGE_SIZE; import static org.apache.parquet.benchmarks.BenchmarkConstants.FIXED_LEN_BYTEARRAY_SIZE; import static org.apache.parquet.benchmarks.BenchmarkConstants.ONE_MILLION; -import static org.apache.parquet.benchmarks.BenchmarkConstants.PAGE_SIZE_4M; -import static org.apache.parquet.benchmarks.BenchmarkConstants.PAGE_SIZE_8M; import static org.apache.parquet.benchmarks.BenchmarkConstants.PAGE_SIZE_DEFAULT; +import static org.apache.parquet.benchmarks.BenchmarkFiles.TARGET_DIR; import static org.apache.parquet.benchmarks.BenchmarkFiles.configuration; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS256M_PS4M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS256M_PS8M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS512M_PS4M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS512M_PS8M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_GZIP; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_SNAPPY; import static org.apache.parquet.benchmarks.BenchmarkFiles.targetDir; import static org.apache.parquet.benchmarks.BenchmarkUtils.deleteIfExists; import static org.apache.parquet.benchmarks.BenchmarkUtils.exists; import static org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_2_0; -import static org.apache.parquet.hadoop.metadata.CompressionCodecName.GZIP; -import static org.apache.parquet.hadoop.metadata.CompressionCodecName.SNAPPY; -import static org.apache.parquet.hadoop.metadata.CompressionCodecName.UNCOMPRESSED; import static org.apache.parquet.schema.MessageTypeParser.parseMessageType; import java.io.IOException; @@ -60,80 +47,13 @@ public class DataGenerator { - public void generateAll() { - try { - generateData( - file_1M, - configuration, - PARQUET_2_0, - BLOCK_SIZE_DEFAULT, - PAGE_SIZE_DEFAULT, - FIXED_LEN_BYTEARRAY_SIZE, - UNCOMPRESSED, - ONE_MILLION); - - // generate data for different block and page sizes - generateData( - file_1M_BS256M_PS4M, - configuration, - PARQUET_2_0, - BLOCK_SIZE_256M, - PAGE_SIZE_4M, - FIXED_LEN_BYTEARRAY_SIZE, - UNCOMPRESSED, - ONE_MILLION); - generateData( - file_1M_BS256M_PS8M, - configuration, - PARQUET_2_0, - BLOCK_SIZE_256M, - PAGE_SIZE_8M, - FIXED_LEN_BYTEARRAY_SIZE, - UNCOMPRESSED, - ONE_MILLION); - generateData( - file_1M_BS512M_PS4M, - configuration, - PARQUET_2_0, - BLOCK_SIZE_512M, - PAGE_SIZE_4M, - FIXED_LEN_BYTEARRAY_SIZE, - UNCOMPRESSED, - ONE_MILLION); - generateData( - file_1M_BS512M_PS8M, - configuration, - PARQUET_2_0, - BLOCK_SIZE_512M, - PAGE_SIZE_8M, - FIXED_LEN_BYTEARRAY_SIZE, - UNCOMPRESSED, - ONE_MILLION); - - // generate data for different codecs - // generateData(parquetFile_1M_LZO, configuration, PARQUET_2_0, BLOCK_SIZE_DEFAULT, PAGE_SIZE_DEFAULT, - // FIXED_LEN_BYTEARRAY_SIZE, LZO, ONE_MILLION); - generateData( - file_1M_SNAPPY, - configuration, - PARQUET_2_0, - BLOCK_SIZE_DEFAULT, - PAGE_SIZE_DEFAULT, - FIXED_LEN_BYTEARRAY_SIZE, - SNAPPY, - ONE_MILLION); - generateData( - file_1M_GZIP, - configuration, - PARQUET_2_0, - BLOCK_SIZE_DEFAULT, - PAGE_SIZE_DEFAULT, - FIXED_LEN_BYTEARRAY_SIZE, - GZIP, - ONE_MILLION); - } catch (IOException e) { - throw new RuntimeException(e); - } + /** + * Returns the canonical path of the 1M-row benchmark file for the given codec. Shared by + * {@link #generateAll()}, {@link WriteBenchmarks} and {@link ReadBenchmarks} so that a corpus + * built up front can be reused by the read benchmark instead of being regenerated on demand. + */ + static Path benchmarkFile(CompressionCodecName codec) { + return new Path(TARGET_DIR + "/PARQUET-1M-" + codec.name()); } public void generateData( @@ -200,6 +120,30 @@ public void cleanup() { deleteIfExists(configuration, targetDir); } + /** + * Pre-generates the 1M-row data file for every {@link CompressionCodecName} at the default block + * and page size. This is optional: {@link ReadBenchmarks} generates whichever file it needs on + * demand (and {@link #generateData} skips files that already exist), but building the whole + * corpus up front lets repeated read runs reuse it instead of regenerating each time. + */ + public void generateAll() { + try { + for (CompressionCodecName codec : CompressionCodecName.values()) { + generateData( + benchmarkFile(codec), + configuration, + PARQUET_2_0, + BLOCK_SIZE_DEFAULT, + PAGE_SIZE_DEFAULT, + FIXED_LEN_BYTEARRAY_SIZE, + codec, + ONE_MILLION); + } + } catch (IOException e) { + throw new RuntimeException(e); + } + } + public static void main(String[] args) { DataGenerator generator = new DataGenerator(); if (args.length < 1) { diff --git a/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/ReadBenchmarks.java b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/ReadBenchmarks.java index 2d6e3a52e3..55ef1e82f3 100644 --- a/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/ReadBenchmarks.java +++ b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/ReadBenchmarks.java @@ -18,107 +18,82 @@ */ package org.apache.parquet.benchmarks; +import static org.apache.parquet.benchmarks.BenchmarkConstants.BLOCK_SIZE_DEFAULT; +import static org.apache.parquet.benchmarks.BenchmarkConstants.FIXED_LEN_BYTEARRAY_SIZE; import static org.apache.parquet.benchmarks.BenchmarkConstants.ONE_MILLION; +import static org.apache.parquet.benchmarks.BenchmarkConstants.PAGE_SIZE_DEFAULT; import static org.apache.parquet.benchmarks.BenchmarkFiles.configuration; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS256M_PS4M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS256M_PS8M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS512M_PS4M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS512M_PS8M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_GZIP; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_SNAPPY; +import static org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_2_0; import java.io.IOException; import org.apache.hadoop.fs.Path; import org.apache.parquet.example.data.Group; import org.apache.parquet.hadoop.ParquetReader; import org.apache.parquet.hadoop.example.GroupReadSupport; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; import org.openjdk.jmh.annotations.Benchmark; import org.openjdk.jmh.annotations.BenchmarkMode; import org.openjdk.jmh.annotations.Level; import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.Param; import org.openjdk.jmh.annotations.Scope; import org.openjdk.jmh.annotations.Setup; import org.openjdk.jmh.annotations.State; import org.openjdk.jmh.infra.Blackhole; +/** + * End-to-end read benchmark: reads back 1M rows from a real Parquet file for each compression + * codec, measuring the full read path (filesystem I/O + decompression + decoding). + * + *

Parameterized by {@code codec} so every codec is covered uniformly. This replaces the + * previous layout that had one hand-written {@code @Benchmark} method per codec/block/page + * combination (and therefore only covered UNCOMPRESSED, SNAPPY, GZIP and LZO). The file for the + * selected codec is generated once per trial in {@link #generateFileForRead()} (skipped if it + * already exists), reading the shared {@link DataGenerator#benchmarkFile corpus} that can be + * pre-built for every codec via {@code DataGenerator generate} (see {@code run.sh generate}). Run + * a subset with e.g. {@code -p codec=ZSTD,LZ4_RAW}. + */ @State(Scope.Benchmark) +@BenchmarkMode(Mode.SingleShotTime) public class ReadBenchmarks { - private void read(Path parquetFile, int nRows, Blackhole blackhole) throws IOException { - ParquetReader reader = ParquetReader.builder(new GroupReadSupport(), parquetFile) - .withConf(configuration) - .build(); - for (int i = 0; i < nRows; i++) { - Group group = reader.read(); - blackhole.consume(group.getBinary("binary_field", 0)); - blackhole.consume(group.getInteger("int32_field", 0)); - blackhole.consume(group.getLong("int64_field", 0)); - blackhole.consume(group.getBoolean("boolean_field", 0)); - blackhole.consume(group.getFloat("float_field", 0)); - blackhole.consume(group.getDouble("double_field", 0)); - blackhole.consume(group.getBinary("flba_field", 0)); - blackhole.consume(group.getInt96("int96_field", 0)); - } - reader.close(); - } - - /** - * This needs to be done exactly once. To avoid needlessly regenerating the files for reading, they aren't cleaned - * as part of the benchmark. If the files exist, a message will be printed and they will not be regenerated. - */ - @Setup(Level.Trial) - public void generateFilesForRead() { - new DataGenerator().generateAll(); - } - - @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void read1MRowsDefaultBlockAndPageSizeUncompressed(Blackhole blackhole) throws IOException { - read(file_1M, ONE_MILLION, blackhole); - } + @Param({"UNCOMPRESSED", "SNAPPY", "GZIP", "LZO", "BROTLI", "LZ4", "ZSTD", "LZ4_RAW"}) + public String codec; - @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void read1MRowsBS256MPS4MUncompressed(Blackhole blackhole) throws IOException { - read(file_1M_BS256M_PS4M, ONE_MILLION, blackhole); - } + private Path file; - @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void read1MRowsBS256MPS8MUncompressed(Blackhole blackhole) throws IOException { - read(file_1M_BS256M_PS8M, ONE_MILLION, blackhole); - } - - @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void read1MRowsBS512MPS4MUncompressed(Blackhole blackhole) throws IOException { - read(file_1M_BS512M_PS4M, ONE_MILLION, blackhole); - } - - @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void read1MRowsBS512MPS8MUncompressed(Blackhole blackhole) throws IOException { - read(file_1M_BS512M_PS8M, ONE_MILLION, blackhole); - } - - // TODO how to handle lzo jar? - // @Benchmark - // public void read1MRowsDefaultBlockAndPageSizeLZO(Blackhole blackhole) - // throws IOException - // { - // read(parquetFile_1M_LZO, ONE_MILLION, blackhole); - // } - - @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void read1MRowsDefaultBlockAndPageSizeSNAPPY(Blackhole blackhole) throws IOException { - read(file_1M_SNAPPY, ONE_MILLION, blackhole); + @Setup(Level.Trial) + public void generateFileForRead() throws IOException { + CompressionCodecName codecName = CompressionCodecName.valueOf(codec); + file = DataGenerator.benchmarkFile(codecName); + new DataGenerator() + .generateData( + file, + configuration, + PARQUET_2_0, + BLOCK_SIZE_DEFAULT, + PAGE_SIZE_DEFAULT, + FIXED_LEN_BYTEARRAY_SIZE, + codecName, + ONE_MILLION); } @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void read1MRowsDefaultBlockAndPageSizeGZIP(Blackhole blackhole) throws IOException { - read(file_1M_GZIP, ONE_MILLION, blackhole); + public void read1MRows(Blackhole blackhole) throws IOException { + try (ParquetReader reader = ParquetReader.builder(new GroupReadSupport(), file) + .withConf(configuration) + .build()) { + for (int i = 0; i < ONE_MILLION; i++) { + Group group = reader.read(); + blackhole.consume(group.getBinary("binary_field", 0)); + blackhole.consume(group.getInteger("int32_field", 0)); + blackhole.consume(group.getLong("int64_field", 0)); + blackhole.consume(group.getBoolean("boolean_field", 0)); + blackhole.consume(group.getFloat("float_field", 0)); + blackhole.consume(group.getDouble("double_field", 0)); + blackhole.consume(group.getBinary("flba_field", 0)); + blackhole.consume(group.getInt96("int96_field", 0)); + } + } } } diff --git a/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/WriteBenchmarks.java b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/WriteBenchmarks.java index 41f961de44..655694c4bf 100644 --- a/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/WriteBenchmarks.java +++ b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/WriteBenchmarks.java @@ -18,156 +18,81 @@ */ package org.apache.parquet.benchmarks; -import static org.apache.parquet.benchmarks.BenchmarkConstants.BLOCK_SIZE_256M; -import static org.apache.parquet.benchmarks.BenchmarkConstants.BLOCK_SIZE_512M; import static org.apache.parquet.benchmarks.BenchmarkConstants.BLOCK_SIZE_DEFAULT; import static org.apache.parquet.benchmarks.BenchmarkConstants.FIXED_LEN_BYTEARRAY_SIZE; import static org.apache.parquet.benchmarks.BenchmarkConstants.ONE_MILLION; -import static org.apache.parquet.benchmarks.BenchmarkConstants.PAGE_SIZE_4M; -import static org.apache.parquet.benchmarks.BenchmarkConstants.PAGE_SIZE_8M; import static org.apache.parquet.benchmarks.BenchmarkConstants.PAGE_SIZE_DEFAULT; import static org.apache.parquet.benchmarks.BenchmarkFiles.configuration; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS256M_PS4M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS256M_PS8M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS512M_PS4M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_BS512M_PS8M; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_GZIP; -import static org.apache.parquet.benchmarks.BenchmarkFiles.file_1M_SNAPPY; import static org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_2_0; -import static org.apache.parquet.hadoop.metadata.CompressionCodecName.GZIP; -import static org.apache.parquet.hadoop.metadata.CompressionCodecName.SNAPPY; -import static org.apache.parquet.hadoop.metadata.CompressionCodecName.UNCOMPRESSED; import static org.openjdk.jmh.annotations.Scope.Thread; import java.io.IOException; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.openjdk.jmh.annotations.AuxCounters; import org.openjdk.jmh.annotations.Benchmark; import org.openjdk.jmh.annotations.BenchmarkMode; import org.openjdk.jmh.annotations.Level; import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.Param; import org.openjdk.jmh.annotations.Setup; import org.openjdk.jmh.annotations.State; +/** + * End-to-end write benchmark: writes 1M rows to a real Parquet file for each compression + * codec, measuring the full write path (encoding + compression + filesystem I/O) as the primary + * time metric, plus the resulting file size as the {@code compressedBytes} secondary metric. + * + *

Parameterized by {@code codec} so every codec is covered uniformly. This replaces the + * previous layout that had one hand-written {@code @Benchmark} method per codec/block/page + * combination (and therefore only covered UNCOMPRESSED, SNAPPY, GZIP and LZO). Block and page + * size are fixed at their defaults; add {@code @Param} fields for them if that dimension is + * needed. Run a subset with e.g. {@code -p codec=ZSTD,LZ4_RAW}. + * + *

Reading the size metric. {@code compressedBytes} is a JMH {@link AuxCounters} of type + * {@link AuxCounters.Type#EVENTS}, reported as the sum over measurement iterations; + * divide by the {@code Cnt} column for the per-file size in bytes (it is deterministic per codec). + */ @State(Thread) +@BenchmarkMode(Mode.SingleShotTime) public class WriteBenchmarks { - private DataGenerator dataGenerator = new DataGenerator(); - @Setup(Level.Iteration) - public void setup() { - // clean existing test data at the beginning of each iteration - dataGenerator.cleanup(); - } + @Param({"UNCOMPRESSED", "SNAPPY", "GZIP", "LZO", "BROTLI", "LZ4", "ZSTD", "LZ4_RAW"}) + public String codec; - @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void write1MRowsDefaultBlockAndPageSizeUncompressed() throws IOException { - dataGenerator.generateData( - file_1M, - configuration, - PARQUET_2_0, - BLOCK_SIZE_DEFAULT, - PAGE_SIZE_DEFAULT, - FIXED_LEN_BYTEARRAY_SIZE, - UNCOMPRESSED, - ONE_MILLION); - } + private final DataGenerator dataGenerator = new DataGenerator(); - @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void write1MRowsBS256MPS4MUncompressed() throws IOException { - dataGenerator.generateData( - file_1M_BS256M_PS4M, - configuration, - PARQUET_2_0, - BLOCK_SIZE_256M, - PAGE_SIZE_4M, - FIXED_LEN_BYTEARRAY_SIZE, - UNCOMPRESSED, - ONE_MILLION); - } + /** Exposes the resulting file size to JMH as a secondary metric. */ + @AuxCounters(AuxCounters.Type.EVENTS) + @State(Thread) + public static class Sizes { + public long compressedBytes; - @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void write1MRowsBS256MPS8MUncompressed() throws IOException { - dataGenerator.generateData( - file_1M_BS256M_PS8M, - configuration, - PARQUET_2_0, - BLOCK_SIZE_256M, - PAGE_SIZE_8M, - FIXED_LEN_BYTEARRAY_SIZE, - UNCOMPRESSED, - ONE_MILLION); + @Setup(Level.Iteration) + public void reset() { + compressedBytes = 0; + } } - @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void write1MRowsBS512MPS4MUncompressed() throws IOException { - dataGenerator.generateData( - file_1M_BS512M_PS4M, - configuration, - PARQUET_2_0, - BLOCK_SIZE_512M, - PAGE_SIZE_4M, - FIXED_LEN_BYTEARRAY_SIZE, - UNCOMPRESSED, - ONE_MILLION); - } - - @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void write1MRowsBS512MPS8MUncompressed() throws IOException { - dataGenerator.generateData( - file_1M_BS512M_PS8M, - configuration, - PARQUET_2_0, - BLOCK_SIZE_512M, - PAGE_SIZE_8M, - FIXED_LEN_BYTEARRAY_SIZE, - UNCOMPRESSED, - ONE_MILLION); - } - - // TODO how to handle lzo jar? - // @Benchmark - // public void write1MRowsDefaultBlockAndPageSizeLZO() - // throws IOException - // { - // dataGenerator.generateData(parquetFile_1M_LZO, - // configuration, - // WriterVersion.PARQUET_2_0, - // BLOCK_SIZE_DEFAULT, - // PAGE_SIZE_DEFAULT, - // FIXED_LEN_BYTEARRAY_SIZE, - // LZO, - // ONE_MILLION); - // } - - @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void write1MRowsDefaultBlockAndPageSizeSNAPPY() throws IOException { - dataGenerator.generateData( - file_1M_SNAPPY, - configuration, - PARQUET_2_0, - BLOCK_SIZE_DEFAULT, - PAGE_SIZE_DEFAULT, - FIXED_LEN_BYTEARRAY_SIZE, - SNAPPY, - ONE_MILLION); + @Setup(Level.Iteration) + public void setup() { + // clean existing test data at the beginning of each iteration + dataGenerator.cleanup(); } @Benchmark - @BenchmarkMode(Mode.SingleShotTime) - public void write1MRowsDefaultBlockAndPageSizeGZIP() throws IOException { + public void write1MRows(Sizes sizes) throws IOException { + Path out = DataGenerator.benchmarkFile(CompressionCodecName.valueOf(codec)); dataGenerator.generateData( - file_1M_GZIP, + out, configuration, PARQUET_2_0, BLOCK_SIZE_DEFAULT, PAGE_SIZE_DEFAULT, FIXED_LEN_BYTEARRAY_SIZE, - GZIP, + CompressionCodecName.valueOf(codec), ONE_MILLION); + sizes.compressedBytes = + out.getFileSystem(configuration).getFileStatus(out).getLen(); } } diff --git a/parquet-common/src/test/java/org/apache/parquet/bytes/TestBytesInput.java b/parquet-common/src/test/java/org/apache/parquet/bytes/TestBytesInput.java index 6aca1811fe..1c26855d87 100644 --- a/parquet-common/src/test/java/org/apache/parquet/bytes/TestBytesInput.java +++ b/parquet-common/src/test/java/org/apache/parquet/bytes/TestBytesInput.java @@ -426,4 +426,41 @@ private void validateToByteBufferIsInternal(Supplier factory) { verify(allocatorMock, never()).allocate(anyInt()); verify(callbackMock, never()).accept(any()); } + + // ---- Tests for ByteArrayBytesInput.toByteArray() zero-copy optimization ---- + + @Test + public void testByteArrayBytesInputToByteArrayZeroCopyFullArray() throws IOException { + byte[] data = new byte[100]; + RANDOM.nextBytes(data); + BytesInput bi = BytesInput.from(data, 0, data.length); + + // When offset=0 and length=array.length, toByteArray() should return the same array instance + byte[] result = bi.toByteArray(); + assertSame("Expected zero-copy (same array instance) for full-array BytesInput", data, result); + } + + @Test + public void testByteArrayBytesInputToByteArrayCopiesForSubArray() throws IOException { + byte[] data = new byte[100]; + RANDOM.nextBytes(data); + BytesInput bi = BytesInput.from(data, 10, 50); + + byte[] result = bi.toByteArray(); + assertEquals("Sub-array toByteArray() should have correct length", 50, result.length); + byte[] expected = new byte[50]; + System.arraycopy(data, 10, expected, 0, 50); + assertArrayEquals("Sub-array toByteArray() content mismatch", expected, result); + } + + @Test + public void testByteArrayBytesInputToByteArrayFromSimpleFactory() throws IOException { + byte[] data = new byte[200]; + RANDOM.nextBytes(data); + // BytesInput.from(byte[]) delegates to from(byte[], 0, length) + BytesInput bi = BytesInput.from(data); + + byte[] result = bi.toByteArray(); + assertSame("Expected zero-copy for BytesInput.from(byte[])", data, result); + } } diff --git a/parquet-hadoop/pom.xml b/parquet-hadoop/pom.xml index 0426d9ff63..6cc397c855 100644 --- a/parquet-hadoop/pom.xml +++ b/parquet-hadoop/pom.xml @@ -159,6 +159,7 @@ aircompressor 2.0.3 + commons-pool commons-pool @@ -174,6 +175,13 @@ zstd-jni ${zstd-jni.version} + + com.aayushatharva.brotli4j + brotli4j + ${brotli4j.version} + runtime + true + com.google.guava @@ -246,33 +254,4 @@ - - - - non-aarch64 - - - !aarch64 - - - - - jitpack.io - https://jitpack.io - Jitpack.io repository - - - - - - com.github.rdblue - brotli-codec - ${brotli-codec.version} - runtime - true - - - - - diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java index c9391201f4..8298b53b89 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java @@ -19,45 +19,59 @@ package org.apache.parquet.hadoop; import com.github.luben.zstd.Zstd; +import com.github.luben.zstd.ZstdCompressCtx; +import com.github.luben.zstd.ZstdDecompressCtx; +import io.airlift.compress.lz4.Lz4Compressor; +import io.airlift.compress.lz4.Lz4Decompressor; +import io.airlift.compress.lz4.Lz4HadoopStreams; +import io.airlift.compress.lzo.LzoHadoopStreams; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; +import java.io.OutputStream; +import java.lang.reflect.Method; import java.nio.ByteBuffer; import java.util.Collections; import java.util.HashMap; import java.util.Map; -import java.util.Objects; +import java.util.zip.Deflater; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.io.compress.CodecPool; import org.apache.hadoop.io.compress.CompressionCodec; -import org.apache.hadoop.io.compress.CompressionOutputStream; -import org.apache.hadoop.io.compress.Compressor; -import org.apache.hadoop.io.compress.Decompressor; -import org.apache.hadoop.io.compress.zlib.ZlibCompressor; import org.apache.hadoop.util.ReflectionUtils; import org.apache.parquet.Preconditions; import org.apache.parquet.bytes.ByteBufferAllocator; +import org.apache.parquet.bytes.ByteBufferInputStream; import org.apache.parquet.bytes.BytesInput; import org.apache.parquet.compression.CompressionCodecFactory; import org.apache.parquet.conf.HadoopParquetConfiguration; import org.apache.parquet.conf.ParquetConfiguration; -import org.apache.parquet.hadoop.codec.Lz4RawCodec; import org.apache.parquet.hadoop.codec.ZstandardCodec; import org.apache.parquet.hadoop.metadata.CompressionCodecName; import org.apache.parquet.hadoop.util.ConfigurationUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.xerial.snappy.Snappy; public class CodecFactory implements CompressionCodecFactory { private static final Logger LOG = LoggerFactory.getLogger(CodecFactory.class); + /** + * Cache of Hadoop {@link CompressionCodec} instances keyed by codec class name (and optional + * compression level). Retained for backwards compatibility with subclasses that override + * {@link #createCompressor(CompressionCodecName)} / {@link #createDecompressor(CompressionCodecName)} + * and rely on {@link #getCodec(CompressionCodecName)}. The default implementation no longer uses + * the Hadoop codec abstraction and does not populate this map. + * + * @deprecated the default implementation no longer resolves codecs through Hadoop; this field is + * kept only for binary/source compatibility and will be removed in 2.0.0. + */ + @Deprecated protected static final Map CODEC_BY_NAME = Collections.synchronizedMap(new HashMap()); - static final String GZIP_COMPRESS_LEVEL = "zlib.compress.level"; - static final String BROTLI_COMPRESS_QUALITY = "compression.brotli.quality"; - private final Map compressors = new HashMap<>(); private final Map decompressors = new HashMap<>(); @@ -68,6 +82,92 @@ public class CodecFactory implements CompressionCodecFactory { @Deprecated protected final Configuration configuration; + /** + * Reflection-based helper for brotli4j (runtime-only dependency). + * Initialized eagerly at class-load time; all fields are null if + * brotli4j is not on the classpath. + * + *

Uses {@code Encoder.compress(byte[], Encoder.Parameters)} for compression and + * {@code Decoder.decompress(byte[], int, int)} for decompression — the latter returns + * {@code byte[]} directly and avoids loading {@code DirectDecompress} which references + * {@code io.netty.buffer.ByteBuf} (optional Netty dependency not on our classpath). + */ + static final class Brotli4j { + static final boolean AVAILABLE; + // Encoder.compress(byte[], Object/*Encoder.Parameters*/) -> byte[] + private static final Method COMPRESS; + // Decoder.decompress(byte[], int/*offset*/, int/*length*/) -> byte[] + private static final Method DECOMPRESS; + // Encoder.Parameters class + private static final Class PARAMS_CLASS; + // Encoder.Parameters.setQuality(int) -> Encoder.Parameters + private static final Method SET_QUALITY; + + static { + boolean loaded = false; + Method compress = null, decompress = null, setQuality = null; + Class paramsClass = null; + try { + // Load native library + Class loader = Class.forName("com.aayushatharva.brotli4j.Brotli4jLoader"); + loader.getMethod("ensureAvailability").invoke(null); + + // Encoder.compress(byte[], Encoder.Parameters) -> byte[] + paramsClass = Class.forName("com.aayushatharva.brotli4j.encoder.Encoder$Parameters"); + Class encoder = Class.forName("com.aayushatharva.brotli4j.encoder.Encoder"); + compress = encoder.getMethod("compress", byte[].class, paramsClass); + + // Decoder.decompress(byte[], int, int) -> byte[] + // This avoids loading DirectDecompress which references io.netty.buffer.ByteBuf + Class decoder = Class.forName("com.aayushatharva.brotli4j.decoder.Decoder"); + decompress = decoder.getMethod("decompress", byte[].class, int.class, int.class); + + // Encoder.Parameters.setQuality(int) -> Encoder.Parameters + setQuality = paramsClass.getMethod("setQuality", int.class); + + loaded = true; + } catch (Throwable t) { + // brotli4j not available -- BROTLI is unsupported; requesting it will throw + // UnsupportedOperationException (there is no Hadoop codec fallback). + LOG.info("brotli4j not available, BROTLI codec will be unsupported: {}", t.toString()); + } + AVAILABLE = loaded; + COMPRESS = compress; + DECOMPRESS = decompress; + PARAMS_CLASS = paramsClass; + SET_QUALITY = setQuality; + } + + /** Create an {@code Encoder.Parameters} instance with the given quality. */ + static Object newParams(int quality) { + try { + Object params = PARAMS_CLASS.getConstructor().newInstance(); + SET_QUALITY.invoke(params, quality); + return params; + } catch (ReflectiveOperationException e) { + throw new RuntimeException("Failed to create Brotli encoder parameters", e); + } + } + + /** Compress using {@code Encoder.compress(byte[], Encoder.Parameters)}. */ + static byte[] compress(byte[] input, Object params) throws IOException { + try { + return (byte[]) COMPRESS.invoke(null, input, params); + } catch (ReflectiveOperationException e) { + throw new IOException("Brotli compression failed", e); + } + } + + /** Decompress using {@code Decoder.decompress(byte[], offset, length)}. */ + static byte[] decompress(byte[] input) throws IOException { + try { + return (byte[]) DECOMPRESS.invoke(null, input, 0, input.length); + } catch (ReflectiveOperationException e) { + throw new IOException("Brotli decompression failed", e); + } + } + } + static final BytesDecompressor NO_OP_DECOMPRESSOR = new BytesDecompressor() { @Override public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) { @@ -162,103 +262,6 @@ public static CodecFactory createDirectCodecFactory( return new DirectCodecFactory(config, allocator, pageSize); } - class HeapBytesDecompressor extends BytesDecompressor { - - private final CompressionCodec codec; - private final Decompressor decompressor; - - HeapBytesDecompressor(CompressionCodec codec) { - this.codec = Objects.requireNonNull(codec); - decompressor = CodecPool.getDecompressor(codec); - } - - @Override - public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException { - final BytesInput decompressed; - if (decompressor != null) { - decompressor.reset(); - } - InputStream is = codec.createInputStream(bytes.toInputStream(), decompressor); - - // Eagerly materialize the decompressed stream for codecs that require all input in a single buffer. - // ZSTD: releases off-heap resources early to avoid fragmentation (see parquet-format#398). - // LZ4_RAW: requires one-shot decompression; the lazy StreamBytesInput.writeInto() path reads via - // Channels.newChannel() in ~8KB chunks, causing the decompressor to be called with an undersized - // output buffer (see #3478). - if (codec instanceof ZstandardCodec || codec instanceof Lz4RawCodec) { - decompressed = BytesInput.copy(BytesInput.from(is, decompressedSize)); - is.close(); - } else { - decompressed = BytesInput.from(is, decompressedSize); - } - return decompressed; - } - - @Override - public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) - throws IOException { - Preconditions.checkArgument( - input.remaining() >= compressedSize, "Not enough bytes available in the input buffer"); - int origLimit = input.limit(); - int origPosition = input.position(); - input.limit(origPosition + compressedSize); - ByteBuffer decompressed = - decompress(BytesInput.from(input), decompressedSize).toByteBuffer(); - output.put(decompressed); - input.limit(origLimit); - input.position(origPosition + compressedSize); - } - - public void release() { - if (decompressor != null) { - CodecPool.returnDecompressor(decompressor); - } - } - } - - /** - * Encapsulates the logic around hadoop compression - */ - class HeapBytesCompressor extends BytesCompressor { - - private final CompressionCodec codec; - private final Compressor compressor; - private final ByteArrayOutputStream compressedOutBuffer; - private final CompressionCodecName codecName; - - HeapBytesCompressor(CompressionCodecName codecName, CompressionCodec codec) { - this.codecName = codecName; - this.codec = Objects.requireNonNull(codec); - this.compressor = CodecPool.getCompressor(codec); - this.compressedOutBuffer = new ByteArrayOutputStream(pageSize); - } - - @Override - public BytesInput compress(BytesInput bytes) throws IOException { - compressedOutBuffer.reset(); - if (compressor != null) { - // null compressor for non-native gzip - compressor.reset(); - } - try (CompressionOutputStream cos = codec.createOutputStream(compressedOutBuffer, compressor)) { - bytes.writeAllTo(cos); - cos.finish(); - } - return BytesInput.from(compressedOutBuffer); - } - - @Override - public void release() { - if (compressor != null) { - CodecPool.returnCompressor(compressor); - } - } - - public CompressionCodecName getCodecName() { - return codecName; - } - } - @Override public BytesCompressor getCompressor(CompressionCodecName codecName) { String key = cacheKey(codecName); @@ -292,22 +295,67 @@ public BytesDecompressor getDecompressor(CompressionCodecName codecName) { } protected BytesCompressor createCompressor(CompressionCodecName codecName) { - return compressorForCodec(codecName, getCodec(codecName)); - } - - protected BytesDecompressor createDecompressor(CompressionCodecName codecName) { - CompressionCodec codec = getCodec(codecName); - return codec == null ? NO_OP_DECOMPRESSOR : new HeapBytesDecompressor(codec); + switch (codecName) { + case UNCOMPRESSED: + return NO_OP_COMPRESSOR; + case SNAPPY: + return new SnappyBytesCompressor(); + case ZSTD: + return new ZstdBytesCompressor( + conf.getInt( + ZstandardCodec.PARQUET_COMPRESS_ZSTD_LEVEL, + ZstandardCodec.DEFAULT_PARQUET_COMPRESS_ZSTD_LEVEL), + conf.getInt( + ZstandardCodec.PARQUET_COMPRESS_ZSTD_WORKERS, + ZstandardCodec.DEFAULTPARQUET_COMPRESS_ZSTD_WORKERS)); + case LZ4_RAW: + return new Lz4RawBytesCompressor(); + case LZ4: + return new Lz4BytesCompressor(pageSize); + case GZIP: + int gzipLevel = conf.getInt("zlib.compress.level", Deflater.DEFAULT_COMPRESSION); + return new GzipBytesCompressor(gzipLevel, pageSize); + case LZO: + return new LzoBytesCompressor(pageSize); + case BROTLI: + if (Brotli4j.AVAILABLE) { + int brotliQuality = conf.getInt("compression.brotli.quality", 1); + return new BrotliBytesCompressor(brotliQuality); + } + throw new UnsupportedOperationException( + "BROTLI codec requires brotli4j on the classpath (com.aayushatharva.brotli4j)"); + default: + throw new UnsupportedOperationException("Codec not supported: " + codecName); + } } protected BytesCompressor createCompressorAtLevel(CompressionCodecName codecName, int level) { - return compressorForCodec(codecName, getCodecAtLevel(codecName, level)); - } - - private BytesCompressor compressorForCodec(CompressionCodecName codecName, CompressionCodec codec) { - return codec == null ? NO_OP_COMPRESSOR : new HeapBytesCompressor(codecName, codec); + switch (codecName) { + case ZSTD: + validateZstdLevel(level); + return new ZstdBytesCompressor( + level, + conf.getInt( + ZstandardCodec.PARQUET_COMPRESS_ZSTD_WORKERS, + ZstandardCodec.DEFAULTPARQUET_COMPRESS_ZSTD_WORKERS)); + case GZIP: + validateGzipLevel(level); + return new GzipBytesCompressor(level, pageSize); + case BROTLI: + if (Brotli4j.AVAILABLE) { + validateBrotliLevel(level); + return new BrotliBytesCompressor(level); + } + throw new UnsupportedOperationException( + "BROTLI codec requires brotli4j on the classpath (com.aayushatharva.brotli4j)"); + default: + // Codec does not support compression levels; ignore the level and use the default. + LOG.warn("Compression level {} is not supported for codec {} and will be ignored.", level, codecName); + return createCompressor(codecName); + } } + /** Validates a ZSTD compression level against the range supported by the zstd library. */ static void validateZstdLevel(int level) { if (level < Zstd.minCompressionLevel() || level > Zstd.maxCompressionLevel()) { throw new BadConfigurationException( @@ -324,97 +372,53 @@ private static void validateBrotliLevel(int level) { } private static void validateGzipLevel(int level) { - if (level != -1 && (level < 0 || level > 9)) { + if (level != Deflater.DEFAULT_COMPRESSION && (level < 0 || level > 9)) { throw new BadConfigurationException("Unsupported GZIP compression level: " + level + ". Valid range is 0 (no compression) to 9 (best compression), or -1 for default."); } } - private static ZlibCompressor.CompressionLevel zlibCompressionLevel(int level) { - switch (level) { - case -1: - return ZlibCompressor.CompressionLevel.DEFAULT_COMPRESSION; - case 0: - return ZlibCompressor.CompressionLevel.NO_COMPRESSION; - case 1: - return ZlibCompressor.CompressionLevel.BEST_SPEED; - case 2: - return ZlibCompressor.CompressionLevel.TWO; - case 3: - return ZlibCompressor.CompressionLevel.THREE; - case 4: - return ZlibCompressor.CompressionLevel.FOUR; - case 5: - return ZlibCompressor.CompressionLevel.FIVE; - case 6: - return ZlibCompressor.CompressionLevel.SIX; - case 7: - return ZlibCompressor.CompressionLevel.SEVEN; - case 8: - return ZlibCompressor.CompressionLevel.EIGHT; - case 9: - return ZlibCompressor.CompressionLevel.BEST_COMPRESSION; - default: - throw new BadConfigurationException("Unsupported GZIP compression level: " + level - + ". Valid range is 0 (no compression) to 9 (best compression), or -1 for default."); - } - } - - /** - * Returns a {@link CompressionCodec} instance configured at the specified compression level. - * A level-specific {@link Configuration} snapshot is built so that the codec is initialized - * with the requested level rather than the global configuration value. - * For codecs that do not support levels the method falls back to {@link #getCodec(CompressionCodecName)}. - */ - private CompressionCodec getCodecAtLevel(CompressionCodecName codecName, int level) { - String codecClassName = codecName.getHadoopCompressionCodecClassName(); - if (codecClassName == null) { - return null; - } - String key = cacheKey(codecName, level); - CompressionCodec codec = CODEC_BY_NAME.get(key); - if (codec != null) { - return codec; - } - // Build a Configuration snapshot with the level explicitly set for this codec. - Configuration levelConf = new Configuration(ConfigurationUtil.createHadoopConfiguration(conf)); + protected BytesDecompressor createDecompressor(CompressionCodecName codecName) { switch (codecName) { + case UNCOMPRESSED: + return NO_OP_DECOMPRESSOR; + case SNAPPY: + return new SnappyBytesDecompressor(); case ZSTD: - validateZstdLevel(level); - levelConf.setInt(ZstandardCodec.PARQUET_COMPRESS_ZSTD_LEVEL, level); - break; + return new ZstdBytesDecompressor(); + case LZ4_RAW: + return new Lz4RawBytesDecompressor(); + case LZ4: + return new Lz4BytesDecompressor(); case GZIP: - validateGzipLevel(level); - levelConf.setEnum(GZIP_COMPRESS_LEVEL, zlibCompressionLevel(level)); - break; + return new GzipBytesDecompressor(); + case LZO: + return new LzoBytesDecompressor(); case BROTLI: - validateBrotliLevel(level); - levelConf.setInt(BROTLI_COMPRESS_QUALITY, level); - break; + if (Brotli4j.AVAILABLE) { + return new BrotliBytesDecompressor(); + } + throw new UnsupportedOperationException( + "BROTLI codec requires brotli4j on the classpath (com.aayushatharva.brotli4j)"); default: - // Codec does not support levels; fall back to the default codec instance. - LOG.warn("Compression level {} is not supported for codec {} and will be ignored.", level, codecName); - return getCodec(codecName); - } - try { - Class codecClass; - try { - codecClass = Class.forName(codecClassName); - } catch (ClassNotFoundException e) { - codecClass = new Configuration(false).getClassLoader().loadClass(codecClassName); - } - codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, levelConf); - CODEC_BY_NAME.put(key, codec); - return codec; - } catch (ClassNotFoundException e) { - throw new BadConfigurationException("Class " + codecClassName + " was not found", e); + throw new UnsupportedOperationException("Codec not supported: " + codecName); } } /** + * Resolve the Hadoop {@link CompressionCodec} corresponding to the given Parquet codec name. + * + *

The default {@link CodecFactory} no longer routes compression/decompression through the + * Hadoop codec abstraction (see the optimized {@code *BytesCompressor}/{@code *BytesDecompressor} + * implementations in this class). This method is retained only for backwards compatibility with + * third-party subclasses that override {@link #createCompressor(CompressionCodecName)} or + * {@link #createDecompressor(CompressionCodecName)} and delegate to it. + * * @param codecName the requested codec * @return the corresponding hadoop codec. null if UNCOMPRESSED + * @deprecated the default implementation no longer uses Hadoop codecs; will be removed in 2.0.0. */ + @Deprecated protected CompressionCodec getCodec(CompressionCodecName codecName) { String codecClassName = codecName.getHadoopCompressionCodecClassName(); if (codecClassName == null) { @@ -447,10 +451,10 @@ private String cacheKey(CompressionCodecName codecName) { String level = null; switch (codecName) { case GZIP: - level = conf.get(GZIP_COMPRESS_LEVEL); + level = conf.get("zlib.compress.level"); break; case BROTLI: - level = conf.get(BROTLI_COMPRESS_QUALITY); + level = conf.get("compression.brotli.quality"); break; case ZSTD: level = conf.get("parquet.compression.codec.zstd.level"); @@ -503,4 +507,628 @@ public abstract void decompress(ByteBuffer input, int compressedSize, ByteBuffer public abstract void release(); } + + // ---- Optimized Snappy compressor/decompressor using direct JNI calls ---- + + /** + * Compresses using Snappy's byte-array JNI API directly, bypassing the Hadoop + * stream abstraction. This avoids intermediate direct ByteBuffer copies and + * reduces the compression to a single native call per page. + */ + static class SnappyBytesCompressor extends BytesCompressor { + private byte[] outputBuffer; + + @Override + public BytesInput compress(BytesInput bytes) throws IOException { + byte[] input = bytes.toByteArray(); + int maxLen = Snappy.maxCompressedLength(input.length); + if (outputBuffer == null || outputBuffer.length < maxLen) { + outputBuffer = new byte[maxLen]; + } + int compressed = Snappy.compress(input, 0, input.length, outputBuffer, 0); + return BytesInput.from(outputBuffer, 0, compressed); + } + + @Override + public CompressionCodecName getCodecName() { + return CompressionCodecName.SNAPPY; + } + + @Override + public void release() { + outputBuffer = null; + } + } + + /** + * Decompresses using Snappy's JNI API directly. The {@link ByteBuffer} overload uses + * {@link Snappy#uncompress(ByteBuffer, ByteBuffer)} which, for direct buffers, passes + * native memory addresses straight to the snappy library with no JNI array pinning or + * intermediate copies. + */ + static class SnappyBytesDecompressor extends BytesDecompressor { + @Override + public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException { + byte[] input = bytes.toByteArray(); + byte[] output = new byte[decompressedSize]; + Snappy.uncompress(input, 0, input.length, output, 0); + return BytesInput.from(output); + } + + @Override + public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) + throws IOException { + int origInputLimit = input.limit(); + input.limit(input.position() + compressedSize); + int origOutputLimit = output.limit(); + output.limit(output.position() + decompressedSize); + // Use slices so native API works on independent buffers; advance positions manually. + Snappy.uncompress(input.slice(), output.slice()); + input.position(input.limit()); + input.limit(origInputLimit); + output.position(output.limit()); + output.limit(origOutputLimit); + } + + @Override + public void release() {} + } + + // ---- Optimized ZSTD compressor/decompressor using zstd-jni context API directly ---- + + /** + * Compresses using a reusable {@link ZstdCompressCtx}, bypassing the Hadoop codec + * framework ({@code ZstandardCodec}, {@code CodecPool}, {@code CompressionOutputStream} + * wrapper). The context is created once at construction and reused across calls, + * avoiding per-call JNI context creation, internal buffer allocation, and Java stream + * overhead. Multi-threaded compression via {@code workers > 0} is supported through + * {@link ZstdCompressCtx#setWorkers(int)}. + * + *

Measured effect vs. the previous streaming path is modest and data-dependent: roughly + * parity to ~1.3x faster for most page shapes/sizes, with the largest gains on small, + * compressible pages. High-entropy (barely compressible) ~1MB pages can be slightly slower. + * See {@code CompressionBenchmark} (parquet-benchmarks) for reproducible numbers across data + * shapes; earlier "1.5-3.4x" figures were an artifact of an unrepresentative benchmark data + * generator (see GH-3530). + */ + static class ZstdBytesCompressor extends BytesCompressor { + private final ZstdCompressCtx context; + private byte[] outputBuffer; + + ZstdBytesCompressor(int level, int workers) { + this.context = new ZstdCompressCtx(); + this.context.setLevel(level); + if (workers > 0) { + this.context.setWorkers(workers); + } + } + + @Override + public BytesInput compress(BytesInput bytes) throws IOException { + byte[] input = bytes.toByteArray(); + int maxLen = (int) Zstd.compressBound(input.length); + if (outputBuffer == null || outputBuffer.length < maxLen) { + outputBuffer = new byte[maxLen]; + } + int compressed = context.compressByteArray(outputBuffer, 0, outputBuffer.length, input, 0, input.length); + return BytesInput.from(outputBuffer, 0, compressed); + } + + @Override + public CompressionCodecName getCodecName() { + return CompressionCodecName.ZSTD; + } + + @Override + public void release() { + context.close(); + outputBuffer = null; + } + } + + /** + * Decompresses using a reusable {@link ZstdDecompressCtx}, bypassing the Hadoop + * codec framework. The context is created once at construction and reused across + * calls, avoiding per-call JNI context creation, internal buffer allocation, and + * Java stream overhead. The {@link ByteBuffer} overload uses + * {@link Zstd#decompress(ByteBuffer, ByteBuffer)} to pass buffers directly to the + * native library without intermediate copies. + */ + static class ZstdBytesDecompressor extends BytesDecompressor { + private final ZstdDecompressCtx context; + + ZstdBytesDecompressor() { + this.context = new ZstdDecompressCtx(); + } + + @Override + public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException { + byte[] input = bytes.toByteArray(); + byte[] output = new byte[decompressedSize]; + int decompressed = context.decompressByteArray(output, 0, decompressedSize, input, 0, input.length); + if (decompressed != decompressedSize) { + throw new IOException("Unexpected decompressed size: " + decompressed + " != " + decompressedSize); + } + return BytesInput.from(output); + } + + @Override + public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) + throws IOException { + int origInputLimit = input.limit(); + input.limit(input.position() + compressedSize); + int origOutputLimit = output.limit(); + output.limit(output.position() + decompressedSize); + // Zstd.decompress uses (dst, src) parameter order, matching the native zstd convention. + // Use slices so native API works on independent buffers; advance positions manually. + Zstd.decompress(output.slice(), input.slice()); + input.position(input.limit()); + input.limit(origInputLimit); + output.position(output.limit()); + output.limit(origOutputLimit); + } + + @Override + public void release() { + context.close(); + } + } + + // ---- Optimized LZ4_RAW compressor/decompressor using airlift LZ4 directly ---- + + /** + * Compresses using airlift's LZ4 compressor via reusable direct ByteBuffers, bypassing the + * Hadoop stream abstraction. As with the decompressor, direct buffers favor compressible pages; + * a plain heap-array encode is faster only on high-entropy pages. See {@code CompressionBenchmark} + * (parquet-benchmarks) and GH-3530. + */ + static class Lz4RawBytesCompressor extends BytesCompressor { + private final Lz4Compressor compressor = new Lz4Compressor(); + private ByteBuffer directInputBuf; + private ByteBuffer directOutputBuf; + + @Override + public BytesInput compress(BytesInput bytes) throws IOException { + byte[] input = bytes.toByteArray(); + int maxLen = compressor.maxCompressedLength(input.length); + + // Grow reusable direct input buffer if needed + if (directInputBuf == null || directInputBuf.capacity() < input.length) { + directInputBuf = ByteBuffer.allocateDirect(input.length); + } + directInputBuf.clear(); + directInputBuf.put(input); + directInputBuf.flip(); + + // Grow reusable direct output buffer if needed + if (directOutputBuf == null || directOutputBuf.capacity() < maxLen) { + directOutputBuf = ByteBuffer.allocateDirect(maxLen); + } + directOutputBuf.clear(); + + compressor.compress(directInputBuf, directOutputBuf); + int compressedSize = directOutputBuf.position(); + + // Copy result to heap byte array + directOutputBuf.flip(); + byte[] output = new byte[compressedSize]; + directOutputBuf.get(output); + return BytesInput.from(output, 0, compressedSize); + } + + @Override + public CompressionCodecName getCodecName() { + return CompressionCodecName.LZ4_RAW; + } + + @Override + public void release() { + directInputBuf = null; + directOutputBuf = null; + } + } + + /** + * Decompresses using airlift's LZ4 decompressor via reusable direct ByteBuffers. + * + *

Staging through direct buffers lets aircompressor's native match-copy loop use raw pointers, + * which is faster for compressible (match-heavy) pages — the common case for encoded Parquet + * data. The trade-off is the extra heap<->direct copies, which make this slower than a plain + * heap-array decode for high-entropy pages (where SNAPPY/ZSTD-style heap decoding is ~1.8x + * faster at 1MB). Isolated per-case benchmarks show direct wins on small compressible pages and + * loses on high-entropy ones; the direct path is kept as the better fit for typical + * dictionary/RLE-encoded pages. See {@code CompressionBenchmark} (parquet-benchmarks) and + * GH-3530. (SNAPPY and ZSTD decompress into heap arrays.) + */ + static class Lz4RawBytesDecompressor extends BytesDecompressor { + private final Lz4Decompressor decompressor = new Lz4Decompressor(); + private ByteBuffer directInputBuf; + private ByteBuffer directOutputBuf; + + @Override + public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException { + int inputSize = Math.toIntExact(bytes.size()); + + // Grow reusable direct input buffer if needed + if (directInputBuf == null || directInputBuf.capacity() < inputSize) { + directInputBuf = ByteBuffer.allocateDirect(inputSize); + } + directInputBuf.clear().limit(inputSize); + // toByteArray() is zero-copy for ByteArrayBytesInput (returns backing array directly) + directInputBuf.put(bytes.toByteArray(), 0, inputSize); + directInputBuf.flip(); + + // Grow reusable direct output buffer if needed + if (directOutputBuf == null || directOutputBuf.capacity() < decompressedSize) { + directOutputBuf = ByteBuffer.allocateDirect(decompressedSize); + } + directOutputBuf.clear().limit(decompressedSize); + + decompressor.decompress(directInputBuf.slice(), directOutputBuf.slice()); + + // Copy result to heap — returning a ByteArrayBytesInput allows callers to + // get the byte[] via toByteArray() without an additional copy. + byte[] output = new byte[decompressedSize]; + directOutputBuf.position(0).limit(decompressedSize); + directOutputBuf.get(output); + return BytesInput.from(output); + } + + @Override + public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) + throws IOException { + int origInputLimit = input.limit(); + input.limit(input.position() + compressedSize); + int origOutputLimit = output.limit(); + output.limit(output.position() + decompressedSize); + // Use slices so native API works on independent buffers; advance positions manually. + decompressor.decompress(input.slice(), output.slice()); + input.position(input.limit()); + input.limit(origInputLimit); + output.position(output.limit()); + output.limit(origOutputLimit); + } + + @Override + public void release() { + directInputBuf = null; + directOutputBuf = null; + } + } + + // ---- Optimized GZIP compressor/decompressor using JDK GZIPOutputStream/GZIPInputStream directly ---- + + /** + * Compresses using {@link GZIPOutputStream} directly, bypassing Hadoop's + * GzipCodec and the associated codec pool / stream wrapper overhead. + * + *

Note: this implementation always uses Java's built-in zlib via + * {@link GZIPOutputStream}. It does not use Hadoop native libraries, + * so hardware-accelerated compression via Intel ISA-L will not be used even if + * the native libraries are installed. The overhead reduction from bypassing the + * Hadoop codec framework typically outweighs the ISA-L advantage for the page + * sizes used by Parquet. + */ + static class GzipBytesCompressor extends BytesCompressor { + // The JDK default GZIP stream buffer is only 512 B, which forces many tiny inflate/deflate + // calls on multi-hundred-KB Parquet pages; a 64 KB buffer removes that overhead. + private static final int GZIP_BUFFER_SIZE = 64 * 1024; + + private final int level; + private final ByteArrayOutputStream baos; + + GzipBytesCompressor(int level, int pageSize) { + this.level = level; + this.baos = new ByteArrayOutputStream(pageSize); + } + + @Override + public BytesInput compress(BytesInput bytes) throws IOException { + baos.reset(); + try (GZIPOutputStream gos = new GZIPOutputStream(baos, GZIP_BUFFER_SIZE) { + { + def.setLevel(level); + } + }) { + bytes.writeAllTo(gos); + } + return BytesInput.from(baos); + } + + @Override + public CompressionCodecName getCodecName() { + return CompressionCodecName.GZIP; + } + + @Override + public void release() {} + } + + /** + * Decompresses using {@link GZIPInputStream} directly, bypassing Hadoop's + * GzipCodec and the associated codec pool / stream wrapper overhead. + * CRC32 and size verification is handled by the JDK implementation. + * + *

Note: this implementation always uses Java's built-in zlib via + * {@link GZIPInputStream}. It does not use Hadoop native libraries, + * so hardware-accelerated decompression via Intel ISA-L will not be used even if + * the native libraries are installed. + */ + static class GzipBytesDecompressor extends BytesDecompressor { + private static final int GZIP_BUFFER_SIZE = 64 * 1024; + + @Override + public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException { + try (GZIPInputStream gis = new GZIPInputStream(bytes.toInputStream(), GZIP_BUFFER_SIZE)) { + byte[] output = new byte[decompressedSize]; + int offset = 0; + while (offset < decompressedSize) { + int read = gis.read(output, offset, decompressedSize - offset); + if (read < 0) { + throw new IOException( + "Unexpected end of GZIP stream at offset " + offset + " of " + decompressedSize); + } + offset += read; + } + return BytesInput.from(output); + } + } + + @Override + public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) + throws IOException { + // Wrap the input ByteBuffer slice in an InputStream to avoid allocating a temp byte array. + // GZIPInputStream is stream-based so we still need a temp output array. + ByteBuffer inputSlice = input.slice(); + inputSlice.limit(compressedSize); + try (GZIPInputStream gis = new GZIPInputStream(ByteBufferInputStream.wrap(inputSlice), GZIP_BUFFER_SIZE)) { + byte[] outputBytes = new byte[decompressedSize]; + int offset = 0; + while (offset < decompressedSize) { + int read = gis.read(outputBytes, offset, decompressedSize - offset); + if (read < 0) { + throw new IOException( + "Unexpected end of GZIP stream at offset " + offset + " of " + decompressedSize); + } + offset += read; + } + output.put(outputBytes); + } + input.position(input.position() + compressedSize); + } + + @Override + public void release() {} + } + + // ---- Optimized LZO compressor/decompressor using aircompressor's Hadoop-framed LZO directly ---- + + /** + * Compresses using aircompressor's LZO Hadoop-framed streams directly, + * bypassing the GPL-licensed {@code com.hadoop.compression.lzo.LzoCodec} and + * the associated Hadoop codec pool / stream wrapper overhead. The framing + * format (big-endian length-prefixed blocks) is wire-compatible with Hadoop's + * LzoCodec, so files produced by this compressor are readable by any standard + * Parquet reader. + */ + static class LzoBytesCompressor extends BytesCompressor { + private static final LzoHadoopStreams LZO_STREAMS = new LzoHadoopStreams(); + private final ByteArrayOutputStream baos; + + LzoBytesCompressor(int pageSize) { + this.baos = new ByteArrayOutputStream(pageSize); + } + + @Override + public BytesInput compress(BytesInput bytes) throws IOException { + baos.reset(); + try (OutputStream los = LZO_STREAMS.createOutputStream(baos)) { + bytes.writeAllTo(los); + } + return BytesInput.from(baos); + } + + @Override + public CompressionCodecName getCodecName() { + return CompressionCodecName.LZO; + } + + @Override + public void release() {} + } + + /** + * Decompresses using aircompressor's LZO Hadoop-framed streams directly, + * bypassing the GPL-licensed Hadoop LzoCodec. Reads the same big-endian + * length-prefixed block framing that Hadoop's LzoCodec produces. + */ + static class LzoBytesDecompressor extends BytesDecompressor { + private static final LzoHadoopStreams LZO_STREAMS = new LzoHadoopStreams(); + + @Override + public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException { + try (InputStream lis = LZO_STREAMS.createInputStream(bytes.toInputStream())) { + byte[] output = new byte[decompressedSize]; + int offset = 0; + while (offset < decompressedSize) { + int read = lis.read(output, offset, decompressedSize - offset); + if (read < 0) { + throw new IOException( + "Unexpected end of LZO stream at offset " + offset + " of " + decompressedSize); + } + offset += read; + } + return BytesInput.from(output); + } + } + + @Override + public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) + throws IOException { + ByteBuffer inputSlice = input.slice(); + inputSlice.limit(compressedSize); + try (InputStream lis = LZO_STREAMS.createInputStream(ByteBufferInputStream.wrap(inputSlice))) { + byte[] outputBytes = new byte[decompressedSize]; + int offset = 0; + while (offset < decompressedSize) { + int read = lis.read(outputBytes, offset, decompressedSize - offset); + if (read < 0) { + throw new IOException( + "Unexpected end of LZO stream at offset " + offset + " of " + decompressedSize); + } + offset += read; + } + output.put(outputBytes); + } + input.position(input.position() + compressedSize); + } + + @Override + public void release() {} + } + + // ---- Deprecated LZ4 (Hadoop-framed) using aircompressor's Hadoop-compatible LZ4 streams ---- + + /** + * Compresses using aircompressor's Hadoop-framed LZ4 streams. This produces the same + * block-framed format as Hadoop's {@code org.apache.hadoop.io.compress.Lz4Codec}, i.e. the + * legacy {@link CompressionCodecName#LZ4} Parquet codec (distinct from the newer, one-shot + * {@link CompressionCodecName#LZ4_RAW}). LZ4 is deprecated in favour of LZ4_RAW; this exists so + * that files using the old codec remain readable and writable after the Hadoop codec path was + * removed. + */ + static class Lz4BytesCompressor extends BytesCompressor { + private static final Lz4HadoopStreams LZ4_STREAMS = new Lz4HadoopStreams(); + private final ByteArrayOutputStream baos; + + Lz4BytesCompressor(int pageSize) { + this.baos = new ByteArrayOutputStream(pageSize); + } + + @Override + public BytesInput compress(BytesInput bytes) throws IOException { + baos.reset(); + try (OutputStream los = LZ4_STREAMS.createOutputStream(baos)) { + bytes.writeAllTo(los); + } + return BytesInput.from(baos); + } + + @Override + public CompressionCodecName getCodecName() { + return CompressionCodecName.LZ4; + } + + @Override + public void release() {} + } + + /** + * Decompresses the legacy Hadoop-framed {@link CompressionCodecName#LZ4} format using + * aircompressor's Hadoop-framed LZ4 streams. + */ + static class Lz4BytesDecompressor extends BytesDecompressor { + private static final Lz4HadoopStreams LZ4_STREAMS = new Lz4HadoopStreams(); + + @Override + public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException { + try (InputStream lis = LZ4_STREAMS.createInputStream(bytes.toInputStream())) { + byte[] output = new byte[decompressedSize]; + int offset = 0; + while (offset < decompressedSize) { + int read = lis.read(output, offset, decompressedSize - offset); + if (read < 0) { + throw new IOException( + "Unexpected end of LZ4 stream at offset " + offset + " of " + decompressedSize); + } + offset += read; + } + return BytesInput.from(output); + } + } + + @Override + public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) + throws IOException { + ByteBuffer inputSlice = input.slice(); + inputSlice.limit(compressedSize); + try (InputStream lis = LZ4_STREAMS.createInputStream(ByteBufferInputStream.wrap(inputSlice))) { + byte[] outputBytes = new byte[decompressedSize]; + int offset = 0; + while (offset < decompressedSize) { + int read = lis.read(outputBytes, offset, decompressedSize - offset); + if (read < 0) { + throw new IOException( + "Unexpected end of LZ4 stream at offset " + offset + " of " + decompressedSize); + } + offset += read; + } + output.put(outputBytes); + } + input.position(input.position() + compressedSize); + } + + @Override + public void release() {} + } + + /** + * Brotli compressor using brotli4j ({@code com.aayushatharva.brotli4j}) via reflection. + * Single-call byte-array API — no streaming overhead. Default quality=1 + * matches the old jbrotli default and gives a good speed/ratio trade-off. + */ + static class BrotliBytesCompressor extends BytesCompressor { + private final Object params; + + BrotliBytesCompressor(int quality) { + this.params = Brotli4j.newParams(quality); + } + + @Override + public BytesInput compress(BytesInput bytes) throws IOException { + byte[] input = bytes.toByteArray(); + byte[] compressed = Brotli4j.compress(input, params); + return BytesInput.from(compressed); + } + + @Override + public CompressionCodecName getCodecName() { + return CompressionCodecName.BROTLI; + } + + @Override + public void release() {} + } + + /** + * Brotli decompressor using brotli4j ({@code com.aayushatharva.brotli4j}) via reflection. + * Single-call byte-array API. For the ByteBuffer overload the input slice + * is copied to a heap array, decompressed, and the result put into the + * output buffer — Brotli is slow enough that the copy overhead is negligible. + */ + static class BrotliBytesDecompressor extends BytesDecompressor { + + @Override + public BytesInput decompress(BytesInput bytes, int uncompressedSize) throws IOException { + byte[] compressed = bytes.toByteArray(); + byte[] decompressed = Brotli4j.decompress(compressed); + return BytesInput.from(decompressed); + } + + @Override + public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) + throws IOException { + ByteBuffer inputSlice = input.slice(); + inputSlice.limit(compressedSize); + byte[] compressedBytes = new byte[compressedSize]; + inputSlice.get(compressedBytes); + + byte[] decompressed = Brotli4j.decompress(compressedBytes); + output.put(decompressed); + input.position(input.position() + compressedSize); + } + + @Override + public void release() {} + } } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java index 312ccedf8f..626715b7d5 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java @@ -56,7 +56,8 @@ class DirectCodecFactory extends CodecFactory implements AutoCloseable { private final ByteBufferAllocator allocator; - // Any of these can be null depending on the version of hadoop on the classpath + // Any of these can be null depending on the version of hadoop on the classpath. + // Only used by the deprecated IndirectDecompressor/FullDirectDecompressor/DirectCodecPool shims. private static final Class DIRECT_DECOMPRESSION_CODEC_CLASS; private static final Method DECOMPRESS_METHOD; private static final Method CREATE_DIRECT_DECOMPRESSOR_METHOD; @@ -103,8 +104,14 @@ protected BytesCompressor createCompressor(final CompressionCodecName codecName) return new SnappyCompressor(); case ZSTD: return new ZstdCompressor(); - // todo: create class similar to the SnappyCompressor for zlib and exclude it as - // snappy is above since it also generates allocateDirect calls. + case LZ4_RAW: + return new Lz4RawCompressor(); + case BROTLI: + if (Brotli4j.AVAILABLE) { + return new BrotliDirectCompressor(); + } + return super.createCompressor(codecName); + case LZO: default: return super.createCompressor(codecName); } @@ -131,17 +138,19 @@ protected BytesDecompressor createDecompressor(final CompressionCodecName codecN return new SnappyDecompressor(); case ZSTD: return new ZstdDecompressor(); - default: - CompressionCodec codec = getCodec(codecName); - if (codec == null) { - return NO_OP_DECOMPRESSOR; - } - DirectCodecPool.CodecPool pool = DirectCodecPool.INSTANCE.codec(codec); - if (pool.supportsDirectDecompression()) { - return new FullDirectDecompressor(pool.borrowDirectDecompressor()); - } else { - return new IndirectDecompressor(pool.borrowDecompressor()); + case LZ4_RAW: + return new Lz4RawDecompressor(); + case BROTLI: + if (Brotli4j.AVAILABLE) { + return new BrotliDirectDecompressor(); } + // fall through to super (which throws UnsupportedOperationException) + case GZIP: + case LZO: + case UNCOMPRESSED: + return super.createDecompressor(codecName); + default: + return super.createDecompressor(codecName); } } @@ -149,50 +158,6 @@ public void close() { release(); } - /** - * Wrapper around legacy hadoop compressors that do not implement a direct memory - * based version of the decompression algorithm. - */ - public class IndirectDecompressor extends BytesDecompressor { - private final Decompressor decompressor; - - public IndirectDecompressor(CompressionCodec codec) { - this(DirectCodecPool.INSTANCE.codec(codec).borrowDecompressor()); - } - - private IndirectDecompressor(Decompressor decompressor) { - this.decompressor = decompressor; - } - - @Override - public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException { - decompressor.reset(); - byte[] inputBytes = bytes.toByteArray(); - decompressor.setInput(inputBytes, 0, inputBytes.length); - byte[] output = new byte[decompressedSize]; - decompressor.decompress(output, 0, decompressedSize); - return BytesInput.from(output); - } - - @Override - public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) - throws IOException { - - decompressor.reset(); - byte[] inputBytes = new byte[compressedSize]; - input.get(inputBytes); - decompressor.setInput(inputBytes, 0, inputBytes.length); - byte[] outputBytes = new byte[decompressedSize]; - decompressor.decompress(outputBytes, 0, decompressedSize); - output.put(outputBytes); - } - - @Override - public void release() { - DirectCodecPool.INSTANCE.returnDecompressor(decompressor); - } - } - private abstract class BaseDecompressor extends BytesDecompressor { private final ReusingByteBufferAllocator inputAllocator; private final ReusingByteBufferAllocator outputAllocator; @@ -212,8 +177,7 @@ public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOEx ByteBuffer output = outputAllocator.allocate(decompressedSize); int size = decompress(input.slice(), output.slice()); if (size != decompressedSize) { - throw new DirectCodecPool.ParquetCompressionCodecException( - "Unexpected decompressed size: " + size + " != " + decompressedSize); + throw new IOException("Unexpected decompressed size: " + size + " != " + decompressedSize); } output.limit(size); return BytesInput.from(output); @@ -231,8 +195,7 @@ public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, output.limit(output.position() + decompressedSize); int size = decompress(input.slice(), output.slice()); if (size != decompressedSize) { - throw new DirectCodecPool.ParquetCompressionCodecException( - "Unexpected decompressed size: " + size + " != " + decompressedSize); + throw new IOException("Unexpected decompressed size: " + size + " != " + decompressedSize); } input.position(input.limit()); input.limit(origInputLimit); @@ -283,66 +246,6 @@ public void release() { abstract void closeCompressor(); } - /** - * Wrapper around new Hadoop compressors that implement a direct memory - * based version of a particular decompression algorithm. To maintain - * compatibility with Hadoop 1.x these classes that implement - * {@link org.apache.hadoop.io.compress.DirectDecompressionCodec} - * are currently retrieved and have their decompression method invoked - * with reflection. - */ - public class FullDirectDecompressor extends BaseDecompressor { - private final Object decompressor; - - public FullDirectDecompressor(CompressionCodecName codecName) { - this(DirectCodecPool.INSTANCE - .codec(Objects.requireNonNull(getCodec(codecName))) - .borrowDirectDecompressor()); - } - - private FullDirectDecompressor(Object decompressor) { - this.decompressor = decompressor; - } - - @Override - public BytesInput decompress(BytesInput compressedBytes, int decompressedSize) throws IOException { - // Similarly to non-direct decompressors, we reset before use, if possible (see HeapBytesDecompressor) - if (decompressor instanceof Decompressor) { - ((Decompressor) decompressor).reset(); - } - return super.decompress(compressedBytes, decompressedSize); - } - - @Override - public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) - throws IOException { - // Similarly to non-direct decompressors, we reset before use, if possible (see HeapBytesDecompressor) - if (decompressor instanceof Decompressor) { - ((Decompressor) decompressor).reset(); - } - super.decompress(input, compressedSize, output, decompressedSize); - } - - @Override - int decompress(ByteBuffer input, ByteBuffer output) { - int startPos = output.position(); - try { - DECOMPRESS_METHOD.invoke(decompressor, input, output); - } catch (IllegalAccessException | InvocationTargetException e) { - throw new DirectCodecPool.ParquetCompressionCodecException(e); - } - int size = output.position() - startPos; - // Some decompressors flip the output buffer, some don't: - // Let's rely on the limit if the position did not change - return size == 0 ? output.limit() : size; - } - - @Override - void closeDecompressor() { - DirectCodecPool.INSTANCE.returnDirectDecompressor(decompressor); - } - } - /** * @deprecated Use {@link CodecFactory#NO_OP_DECOMPRESSOR} instead */ @@ -419,6 +322,26 @@ void closeDecompressor() { } } + /** + * Direct-memory LZ4_RAW decompressor using airlift's LZ4 decompressor with + * direct ByteBuffers. + */ + private class Lz4RawDecompressor extends BaseDecompressor { + private final io.airlift.compress.lz4.Lz4Decompressor decompressor = + new io.airlift.compress.lz4.Lz4Decompressor(); + + @Override + int decompress(ByteBuffer input, ByteBuffer output) { + decompressor.decompress(input, output); + return output.position(); + } + + @Override + void closeDecompressor() { + // no-op + } + } + private class ZstdCompressor extends BaseCompressor { private final ZstdCompressCtx context; @@ -461,6 +384,95 @@ void closeCompressor() { } } + /** + * Direct-memory LZ4_RAW compressor using airlift's LZ4 compressor with + * direct ByteBuffers, avoiding the stream-based heap path. + */ + private class Lz4RawCompressor extends BaseCompressor { + private final io.airlift.compress.lz4.Lz4Compressor compressor = new io.airlift.compress.lz4.Lz4Compressor(); + + @Override + public CompressionCodecName getCodecName() { + return CompressionCodecName.LZ4_RAW; + } + + @Override + int maxCompressedSize(int size) { + return compressor.maxCompressedLength(size); + } + + @Override + int compress(ByteBuffer input, ByteBuffer output) { + compressor.compress(input, output); + return output.position(); + } + + @Override + void closeCompressor() { + // no-op + } + } + + /** + * Direct-memory Brotli decompressor using brotli4j via reflection. + * brotli4j only exposes a byte-array API, so input/output are copied through heap arrays. + * Brotli is slow enough that the copy overhead is negligible. + */ + private class BrotliDirectDecompressor extends BaseDecompressor { + + @Override + int decompress(ByteBuffer input, ByteBuffer output) throws IOException { + byte[] compressedBytes = new byte[input.remaining()]; + input.get(compressedBytes); + byte[] decompressed = Brotli4j.decompress(compressedBytes); + output.put(decompressed); + return decompressed.length; + } + + @Override + void closeDecompressor() { + // no-op + } + } + + /** + * Direct-memory Brotli compressor using brotli4j via reflection. + * Uses quality=1 by default (fast compression, matching the old jbrotli default). + * brotli4j only exposes a byte-array API, so input/output are copied through heap arrays. + */ + private class BrotliDirectCompressor extends BaseCompressor { + private final Object params; + + BrotliDirectCompressor() { + this.params = Brotli4j.newParams(1); + } + + @Override + public CompressionCodecName getCodecName() { + return CompressionCodecName.BROTLI; + } + + @Override + int maxCompressedSize(int size) { + // Brotli worst case: input size + (input size >> 2) + 1K overhead for small inputs + return size + (size >> 2) + 1024; + } + + @Override + int compress(ByteBuffer input, ByteBuffer output) throws IOException { + byte[] inputBytes = new byte[input.remaining()]; + input.get(inputBytes); + byte[] compressed = Brotli4j.compress(inputBytes, params); + output.put(compressed); + return compressed.length; + } + + @Override + void closeCompressor() { + // no-op + } + } + /** * @deprecated Use {@link CodecFactory#NO_OP_COMPRESSOR} instead */ @@ -485,6 +497,125 @@ public void release() { } } + /** + * Wrapper around legacy hadoop compressors that do not implement a direct memory + * based version of the decompression algorithm. + * + * @deprecated retained for backwards compatibility only. The default direct codec factory no + * longer routes decompression through Hadoop codecs; will be removed in 2.0.0. + */ + @Deprecated + public class IndirectDecompressor extends BytesDecompressor { + private final Decompressor decompressor; + + public IndirectDecompressor(CompressionCodec codec) { + this(DirectCodecPool.INSTANCE.codec(codec).borrowDecompressor()); + } + + private IndirectDecompressor(Decompressor decompressor) { + this.decompressor = decompressor; + } + + @Override + public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException { + decompressor.reset(); + byte[] inputBytes = bytes.toByteArray(); + decompressor.setInput(inputBytes, 0, inputBytes.length); + byte[] output = new byte[decompressedSize]; + decompressor.decompress(output, 0, decompressedSize); + return BytesInput.from(output); + } + + @Override + public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) + throws IOException { + + decompressor.reset(); + byte[] inputBytes = new byte[compressedSize]; + input.get(inputBytes); + decompressor.setInput(inputBytes, 0, inputBytes.length); + byte[] outputBytes = new byte[decompressedSize]; + decompressor.decompress(outputBytes, 0, decompressedSize); + output.put(outputBytes); + } + + @Override + public void release() { + DirectCodecPool.INSTANCE.returnDecompressor(decompressor); + } + } + + /** + * Wrapper around new Hadoop compressors that implement a direct memory + * based version of a particular decompression algorithm. To maintain + * compatibility with Hadoop 1.x these classes that implement + * {@link org.apache.hadoop.io.compress.DirectDecompressionCodec} + * are currently retrieved and have their decompression method invoked + * with reflection. + * + * @deprecated retained for backwards compatibility only. The default direct codec factory no + * longer routes decompression through Hadoop codecs; will be removed in 2.0.0. + */ + @Deprecated + public class FullDirectDecompressor extends BaseDecompressor { + private final Object decompressor; + + public FullDirectDecompressor(CompressionCodecName codecName) { + this(DirectCodecPool.INSTANCE + .codec(Objects.requireNonNull(getCodec(codecName))) + .borrowDirectDecompressor()); + } + + private FullDirectDecompressor(Object decompressor) { + this.decompressor = decompressor; + } + + @Override + public BytesInput decompress(BytesInput compressedBytes, int decompressedSize) throws IOException { + // Similarly to non-direct decompressors, we reset before use, if possible (see HeapBytesDecompressor) + if (decompressor instanceof Decompressor) { + ((Decompressor) decompressor).reset(); + } + return super.decompress(compressedBytes, decompressedSize); + } + + @Override + public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) + throws IOException { + // Similarly to non-direct decompressors, we reset before use, if possible (see HeapBytesDecompressor) + if (decompressor instanceof Decompressor) { + ((Decompressor) decompressor).reset(); + } + super.decompress(input, compressedSize, output, decompressedSize); + } + + @Override + int decompress(ByteBuffer input, ByteBuffer output) { + int startPos = output.position(); + try { + DECOMPRESS_METHOD.invoke(decompressor, input, output); + } catch (IllegalAccessException | InvocationTargetException e) { + throw new DirectCodecPool.ParquetCompressionCodecException(e); + } + int size = output.position() - startPos; + // Some decompressors flip the output buffer, some don't: + // Let's rely on the limit if the position did not change + return size == 0 ? output.limit() : size; + } + + @Override + void closeDecompressor() { + DirectCodecPool.INSTANCE.returnDirectDecompressor(decompressor); + } + } + + /** + * Object pool of Hadoop compressors/decompressors. + * + * @deprecated retained for backwards compatibility only. The default direct codec factory no + * longer pools Hadoop codecs; will be removed in 2.0.0. + */ + @Deprecated static class DirectCodecPool { public static final DirectCodecPool INSTANCE = new DirectCodecPool(); diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestCompressionInterop.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestCompressionInterop.java new file mode 100644 index 0000000000..eae06993e4 --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestCompressionInterop.java @@ -0,0 +1,713 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.parquet.hadoop; + +import static org.apache.parquet.schema.MessageTypeParser.parseMessageType; +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; + +import io.airlift.compress.lz4.Lz4Codec; +import io.airlift.compress.lzo.LzoCodec; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.lang.reflect.Constructor; +import java.nio.ByteBuffer; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.Random; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.io.compress.CodecPool; +import org.apache.hadoop.io.compress.CompressionCodec; +import org.apache.hadoop.io.compress.CompressionOutputStream; +import org.apache.hadoop.io.compress.Compressor; +import org.apache.hadoop.io.compress.Decompressor; +import org.apache.parquet.Preconditions; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.compression.CompressionCodecFactory; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.hadoop.api.ReadSupport; +import org.apache.parquet.hadoop.codec.Lz4RawCodec; +import org.apache.parquet.hadoop.codec.ZstandardCodec; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.hadoop.example.GroupReadSupport; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.io.InputFile; +import org.apache.parquet.io.LocalInputFile; +import org.apache.parquet.io.LocalOutputFile; +import org.apache.parquet.io.OutputFile; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.MessageType; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * End-to-end interoperability tests between the new direct compression path + * (bypassing the Hadoop codec abstraction) and the legacy Hadoop + * {@link CompressionCodec} streaming path. + * + *

These tests verify that: + *

    + *
  • Files written with the Hadoop codec path can be read correctly with the new direct path
    • + *
  • Files written with the new direct path can be read correctly with the Hadoop codec path
    • + *
  • Each codec still round-trips correctly through the new direct path alone
    • + *
+ * + *

This guarantees that the wire format produced/consumed by the optimized direct + * implementations is byte-compatible with the Hadoop-based implementation, so compressed + * Parquet files remain readable regardless of which implementation produced them. + * + *

The legacy path is simulated by {@link HadoopOnlyCodecFactory}, which wraps a Hadoop + * {@link CompressionCodec} in a stream-based compressor/decompressor (the pre-GH-3530 + * behaviour). The Hadoop codec classes used are Parquet's own {@code SnappyCodec}, + * {@code ZstandardCodec}, {@code Lz4RawCodec} and Hadoop's {@code GzipCodec}, resolved via + * {@link CodecFactory#getCodec(CompressionCodecName)}. For LZO and BROTLI the original Hadoop + * codec classes are not on the test classpath, so: + *

    + *
  • LZO: aircompressor's {@link LzoCodec} (Hadoop-compatible framing) is used.
    • + *
  • BROTLI: a stream compressor/decompressor backed by brotli4j's + * {@code BrotliOutputStream}/{@code BrotliInputStream} is used.
    • + *
+ */ +public class TestCompressionInterop { + + private static final Logger LOG = LoggerFactory.getLogger(TestCompressionInterop.class); + + private static final int PAGE_SIZE = 64 * 1024; + private static final int ROW_GROUP_SIZE = 256 * 1024; + private static final int NUM_RECORDS = 500; + + private static final MessageType SCHEMA = parseMessageType("message test { " + + "required binary binary_field; " + + "required int32 int32_field; " + + "required int64 int64_field; " + + "required boolean boolean_field; " + + "required float float_field; " + + "required double double_field; " + + "required fixed_len_byte_array(3) flba_field; " + + "required int96 int96_field; " + + "} "); + + /** + * All codecs that have a direct implementation in CodecFactory. + * LZO uses aircompressor's LzoCodec (Hadoop-compatible) as the Hadoop-path substitute. + * BROTLI uses brotli4j's streaming API as the Hadoop-path substitute. + */ + private static final CompressionCodecName[] ALL_CODECS = { + CompressionCodecName.SNAPPY, + CompressionCodecName.GZIP, + CompressionCodecName.ZSTD, + CompressionCodecName.LZ4_RAW, + CompressionCodecName.LZ4, + CompressionCodecName.LZO, + CompressionCodecName.BROTLI, + }; + + @Rule + public TemporaryFolder tempFolder = new TemporaryFolder(); + + // ---- Round-trip tests for each codec (new direct path only) ---- + + @Test + public void roundTrip_SNAPPY() throws Exception { + testRoundTrip(CompressionCodecName.SNAPPY); + } + + @Test + public void roundTrip_GZIP() throws Exception { + testRoundTrip(CompressionCodecName.GZIP); + } + + @Test + public void roundTrip_ZSTD() throws Exception { + testRoundTrip(CompressionCodecName.ZSTD); + } + + @Test + public void roundTrip_LZ4_RAW() throws Exception { + testRoundTrip(CompressionCodecName.LZ4_RAW); + } + + @Test + public void roundTrip_LZO() throws Exception { + testRoundTrip(CompressionCodecName.LZO); + } + + @Test + public void roundTrip_BROTLI() throws Exception { + testRoundTrip(CompressionCodecName.BROTLI); + } + + @Test + public void roundTripAllCodecs() throws Exception { + for (CompressionCodecName codec : ALL_CODECS) { + LOG.info("Testing round-trip for codec: {}", codec); + testRoundTrip(codec); + } + } + + @Test + public void roundTrip_multiRowGroup() throws Exception { + for (CompressionCodecName codec : ALL_CODECS) { + LOG.info("Testing multi-row-group round-trip for codec: {}", codec); + testRoundTripMultiRowGroup(codec); + } + } + + // ---- Write with Hadoop path, read with Direct path ---- + + @Test + public void writeHadoopReadDirect_SNAPPY() throws Exception { + testWriteHadoopReadDirect(CompressionCodecName.SNAPPY); + } + + @Test + public void writeHadoopReadDirect_GZIP() throws Exception { + testWriteHadoopReadDirect(CompressionCodecName.GZIP); + } + + @Test + public void writeHadoopReadDirect_ZSTD() throws Exception { + testWriteHadoopReadDirect(CompressionCodecName.ZSTD); + } + + @Test + public void writeHadoopReadDirect_LZ4_RAW() throws Exception { + testWriteHadoopReadDirect(CompressionCodecName.LZ4_RAW); + } + + @Test + public void writeHadoopReadDirect_LZO() throws Exception { + testWriteHadoopReadDirect(CompressionCodecName.LZO); + } + + @Test + public void writeHadoopReadDirect_BROTLI() throws Exception { + testWriteHadoopReadDirect(CompressionCodecName.BROTLI); + } + + // ---- Write with Direct path, read with Hadoop path ---- + + @Test + public void writeDirectReadHadoop_SNAPPY() throws Exception { + testWriteDirectReadHadoop(CompressionCodecName.SNAPPY); + } + + @Test + public void writeDirectReadHadoop_GZIP() throws Exception { + testWriteDirectReadHadoop(CompressionCodecName.GZIP); + } + + @Test + public void writeDirectReadHadoop_ZSTD() throws Exception { + testWriteDirectReadHadoop(CompressionCodecName.ZSTD); + } + + @Test + public void writeDirectReadHadoop_LZ4_RAW() throws Exception { + testWriteDirectReadHadoop(CompressionCodecName.LZ4_RAW); + } + + @Test + public void writeDirectReadHadoop_LZO() throws Exception { + testWriteDirectReadHadoop(CompressionCodecName.LZO); + } + + @Test + public void writeDirectReadHadoop_BROTLI() throws Exception { + testWriteDirectReadHadoop(CompressionCodecName.BROTLI); + } + + // ---- Bidirectional test: both directions for all codecs ---- + + @Test + public void bidirectionalInteropAllCodecs() throws Exception { + for (CompressionCodecName codec : ALL_CODECS) { + LOG.info("Testing bidirectional interop for codec: {}", codec); + testWriteHadoopReadDirect(codec); + testWriteDirectReadHadoop(codec); + } + } + + // ---- Multi-row-group interop to validate across row group boundaries ---- + + @Test + public void writeHadoopReadDirect_multiRowGroup() throws Exception { + for (CompressionCodecName codec : ALL_CODECS) { + testInteropMultiRowGroup(codec, /* writeWithHadoop= */ true); + } + } + + @Test + public void writeDirectReadHadoop_multiRowGroup() throws Exception { + for (CompressionCodecName codec : ALL_CODECS) { + testInteropMultiRowGroup(codec, /* writeWithHadoop= */ false); + } + } + + // ---- Implementation ---- + + private void testRoundTrip(CompressionCodecName codec) throws Exception { + Configuration conf = new Configuration(); + Path file = tempFolder.newFolder().toPath().resolve("roundtrip_" + codec.name() + ".parquet"); + + CodecFactory factory = new CodecFactory(conf, PAGE_SIZE); + List expectedRecords = writeFile(file, codec, factory); + factory.release(); + + CodecFactory readFactory = new CodecFactory(conf, PAGE_SIZE); + List actualRecords = readFile(file, readFactory); + readFactory.release(); + + assertRecordsEqual(expectedRecords, actualRecords, codec.name() + " round-trip"); + } + + private void testRoundTripMultiRowGroup(CompressionCodecName codec) throws Exception { + Configuration conf = new Configuration(); + Path file = tempFolder.newFolder().toPath().resolve("roundtrip_mrg_" + codec.name() + ".parquet"); + + // Use a small row group size to force multiple row groups + int smallRowGroupSize = 4 * 1024; + + CodecFactory writeFactory = new CodecFactory(conf, PAGE_SIZE); + List expectedRecords = writeFile(file, codec, writeFactory, smallRowGroupSize, 1000); + writeFactory.release(); + + CodecFactory readFactory = new CodecFactory(conf, PAGE_SIZE); + List actualRecords = readFile(file, readFactory); + readFactory.release(); + + assertRecordsEqual(expectedRecords, actualRecords, codec.name() + " multi-row-group round-trip"); + } + + private void testWriteHadoopReadDirect(CompressionCodecName codec) throws Exception { + Configuration conf = new Configuration(); + Path file = tempFolder.newFolder().toPath().resolve("hadoop_write_" + codec.name() + ".parquet"); + + // Write using the legacy Hadoop codec path + CompressionCodecFactory hadoopFactory = new HadoopOnlyCodecFactory(conf, PAGE_SIZE); + List expectedRecords = writeFile(file, codec, hadoopFactory); + hadoopFactory.release(); + + // Read using the new direct (default) codec path + CodecFactory directFactory = new CodecFactory(conf, PAGE_SIZE); + List actualRecords = readFile(file, directFactory); + directFactory.release(); + + assertRecordsEqual(expectedRecords, actualRecords, codec.name() + " hadoop->direct"); + } + + private void testWriteDirectReadHadoop(CompressionCodecName codec) throws Exception { + Configuration conf = new Configuration(); + Path file = tempFolder.newFolder().toPath().resolve("direct_write_" + codec.name() + ".parquet"); + + // Write using the new direct (default) codec path + CodecFactory directFactory = new CodecFactory(conf, PAGE_SIZE); + List expectedRecords = writeFile(file, codec, directFactory); + directFactory.release(); + + // Read using the legacy Hadoop codec path + CompressionCodecFactory hadoopFactory = new HadoopOnlyCodecFactory(conf, PAGE_SIZE); + List actualRecords = readFile(file, hadoopFactory); + hadoopFactory.release(); + + assertRecordsEqual(expectedRecords, actualRecords, codec.name() + " direct->hadoop"); + } + + private void testInteropMultiRowGroup(CompressionCodecName codec, boolean writeWithHadoop) throws Exception { + Configuration conf = new Configuration(); + String prefix = writeWithHadoop ? "hadoop_mrg_" : "direct_mrg_"; + Path file = tempFolder.newFolder().toPath().resolve(prefix + codec.name() + ".parquet"); + + // Use a small row group size to force multiple row groups + int smallRowGroupSize = 4 * 1024; + + CompressionCodecFactory writeFactory = + writeWithHadoop ? new HadoopOnlyCodecFactory(conf, PAGE_SIZE) : new CodecFactory(conf, PAGE_SIZE); + CompressionCodecFactory readFactory = + writeWithHadoop ? new CodecFactory(conf, PAGE_SIZE) : new HadoopOnlyCodecFactory(conf, PAGE_SIZE); + + List expectedRecords = writeFile(file, codec, writeFactory, smallRowGroupSize, 1000); + writeFactory.release(); + + List actualRecords = readFile(file, readFactory); + readFactory.release(); + + assertRecordsEqual(expectedRecords, actualRecords, codec.name() + " multi-row-group " + prefix); + } + + private List writeFile(Path file, CompressionCodecName codec, CompressionCodecFactory factory) + throws IOException { + return writeFile(file, codec, factory, ROW_GROUP_SIZE, NUM_RECORDS); + } + + private List writeFile( + Path file, CompressionCodecName codec, CompressionCodecFactory factory, int rowGroupSize, int numRecords) + throws IOException { + SimpleGroupFactory groupFactory = new SimpleGroupFactory(SCHEMA); + List records = generateRecords(groupFactory, numRecords); + + OutputFile outputFile = new LocalOutputFile(file); + try (ParquetWriter writer = ExampleParquetWriter.builder(outputFile) + .withType(SCHEMA) + .withCompressionCodec(codec) + .withCodecFactory(factory) + .withRowGroupSize(rowGroupSize) + .withPageSize(PAGE_SIZE) + .withWriteMode(ParquetFileWriter.Mode.CREATE) + .build()) { + for (Group record : records) { + writer.write(record); + } + } + + return records; + } + + private List readFile(Path file, CompressionCodecFactory factory) throws IOException { + List records = new ArrayList<>(); + InputFile inputFile = new LocalInputFile(file); + try (ParquetReader reader = new ParquetReader.Builder(inputFile) { + @Override + protected ReadSupport getReadSupport() { + return new GroupReadSupport(); + } + }.withCodecFactory(factory).build()) { + Group record; + while ((record = reader.read()) != null) { + records.add(record); + } + } + return records; + } + + private List generateRecords(SimpleGroupFactory factory, int numRecords) { + List records = new ArrayList<>(numRecords); + Random random = new Random(42); // fixed seed for reproducibility + + for (int i = 0; i < numRecords; i++) { + byte[] binaryData = new byte[10 + random.nextInt(50)]; + random.nextBytes(binaryData); + + byte[] flbaData = new byte[3]; + random.nextBytes(flbaData); + + byte[] int96Data = new byte[12]; + random.nextBytes(int96Data); + + records.add(factory.newGroup() + .append("binary_field", Binary.fromConstantByteArray(binaryData)) + .append("int32_field", random.nextInt()) + .append("int64_field", random.nextLong()) + .append("boolean_field", random.nextBoolean()) + .append("float_field", random.nextFloat()) + .append("double_field", random.nextDouble()) + .append("flba_field", Binary.fromConstantByteArray(flbaData)) + .append("int96_field", Binary.fromConstantByteArray(int96Data))); + } + return records; + } + + private void assertRecordsEqual(List expected, List actual, String context) { + assertEquals("Record count mismatch for " + context, expected.size(), actual.size()); + for (int i = 0; i < expected.size(); i++) { + Group exp = expected.get(i); + Group act = actual.get(i); + String msg = context + " record " + i; + + assertArrayEquals( + msg + " binary_field", + exp.getBinary("binary_field", 0).getBytes(), + act.getBinary("binary_field", 0).getBytes()); + assertEquals(msg + " int32_field", exp.getInteger("int32_field", 0), act.getInteger("int32_field", 0)); + assertEquals(msg + " int64_field", exp.getLong("int64_field", 0), act.getLong("int64_field", 0)); + assertEquals( + msg + " boolean_field", exp.getBoolean("boolean_field", 0), act.getBoolean("boolean_field", 0)); + assertEquals(msg + " float_field", exp.getFloat("float_field", 0), act.getFloat("float_field", 0), 0.0f); + assertEquals( + msg + " double_field", exp.getDouble("double_field", 0), act.getDouble("double_field", 0), 0.0d); + assertArrayEquals( + msg + " flba_field", + exp.getBinary("flba_field", 0).getBytes(), + act.getBinary("flba_field", 0).getBytes()); + assertArrayEquals( + msg + " int96_field", + exp.getInt96("int96_field", 0).getBytes(), + act.getInt96("int96_field", 0).getBytes()); + } + } + + // ---- Hadoop-path simulation factory ---- + + /** + * A CodecFactory that always uses the Hadoop {@link CompressionCodec} streaming path, + * bypassing the optimized direct implementations. This reproduces the behaviour before + * the GH-3530 optimization so that interop with the direct path can be validated. + * + *

The Hadoop-stream compressor/decompressor wrappers are embedded here (they were + * removed from production {@code CodecFactory}) so that this test remains a genuine + * exercise of the Hadoop codec path. + */ + static class HadoopOnlyCodecFactory extends CodecFactory { + HadoopOnlyCodecFactory(Configuration conf, int pageSize) { + super(conf, pageSize); + } + + @Override + protected BytesCompressor createCompressor(CompressionCodecName codecName) { + switch (codecName) { + case UNCOMPRESSED: + return NO_OP_COMPRESSOR; + case LZO: + // Use aircompressor's LzoCodec which implements Hadoop's CompressionCodec + return new HeapBytesCompressor(codecName, new LzoCodec()); + case LZ4: + // Use aircompressor's Hadoop-framed Lz4Codec (matches org.apache.hadoop...Lz4Codec) + return new HeapBytesCompressor(codecName, new Lz4Codec()); + case BROTLI: + if (CodecFactory.Brotli4j.AVAILABLE) { + return new BrotliStreamCompressor(); + } + // fall through if brotli4j not available + default: + CompressionCodec codec = getCodec(codecName); + if (codec == null) { + return NO_OP_COMPRESSOR; + } + return new HeapBytesCompressor(codecName, codec); + } + } + + @Override + protected BytesDecompressor createDecompressor(CompressionCodecName codecName) { + switch (codecName) { + case UNCOMPRESSED: + return NO_OP_DECOMPRESSOR; + case LZO: + // Use aircompressor's LzoCodec which implements Hadoop's CompressionCodec + return new HeapBytesDecompressor(new LzoCodec()); + case LZ4: + // Use aircompressor's Hadoop-framed Lz4Codec (matches org.apache.hadoop...Lz4Codec) + return new HeapBytesDecompressor(new Lz4Codec()); + case BROTLI: + if (CodecFactory.Brotli4j.AVAILABLE) { + return new BrotliStreamDecompressor(); + } + // fall through if brotli4j not available + default: + CompressionCodec codec = getCodec(codecName); + if (codec == null) { + return NO_OP_DECOMPRESSOR; + } + return new HeapBytesDecompressor(codec); + } + } + + /** Hadoop stream-based compressor (pre-GH-3530 behaviour). */ + class HeapBytesCompressor extends BytesCompressor { + private final CompressionCodec codec; + private final Compressor compressor; + private final ByteArrayOutputStream compressedOutBuffer; + private final CompressionCodecName codecName; + + HeapBytesCompressor(CompressionCodecName codecName, CompressionCodec codec) { + this.codecName = codecName; + this.codec = Objects.requireNonNull(codec); + this.compressor = CodecPool.getCompressor(codec); + this.compressedOutBuffer = new ByteArrayOutputStream(pageSize); + } + + @Override + public BytesInput compress(BytesInput bytes) throws IOException { + compressedOutBuffer.reset(); + if (compressor != null) { + // null compressor for non-native gzip + compressor.reset(); + } + try (CompressionOutputStream cos = codec.createOutputStream(compressedOutBuffer, compressor)) { + bytes.writeAllTo(cos); + cos.finish(); + } + return BytesInput.from(compressedOutBuffer); + } + + @Override + public CompressionCodecName getCodecName() { + return codecName; + } + + @Override + public void release() { + if (compressor != null) { + CodecPool.returnCompressor(compressor); + } + } + } + + /** Hadoop stream-based decompressor (pre-GH-3530 behaviour). */ + class HeapBytesDecompressor extends BytesDecompressor { + private final CompressionCodec codec; + private final Decompressor decompressor; + + HeapBytesDecompressor(CompressionCodec codec) { + this.codec = Objects.requireNonNull(codec); + this.decompressor = CodecPool.getDecompressor(codec); + } + + @Override + public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException { + final BytesInput decompressed; + if (decompressor != null) { + decompressor.reset(); + } + InputStream is = codec.createInputStream(bytes.toInputStream(), decompressor); + + // Eagerly materialize for codecs that require all input in a single buffer (see #3478). + if (codec instanceof ZstandardCodec || codec instanceof Lz4RawCodec) { + decompressed = BytesInput.copy(BytesInput.from(is, decompressedSize)); + is.close(); + } else { + decompressed = BytesInput.from(is, decompressedSize); + } + return decompressed; + } + + @Override + public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) + throws IOException { + Preconditions.checkArgument( + input.remaining() >= compressedSize, "Not enough bytes available in the input buffer"); + int origLimit = input.limit(); + int origPosition = input.position(); + input.limit(origPosition + compressedSize); + ByteBuffer decompressed = + decompress(BytesInput.from(input), decompressedSize).toByteBuffer(); + output.put(decompressed); + input.limit(origLimit); + input.position(origPosition + compressedSize); + } + + @Override + public void release() { + if (decompressor != null) { + CodecPool.returnDecompressor(decompressor); + } + } + } + } + + /** + * Stream-based Brotli compressor using brotli4j's BrotliOutputStream via reflection. + * This mimics what a Hadoop BrotliCodec would do: wrap the output in a BrotliOutputStream. + */ + static class BrotliStreamCompressor extends CodecFactory.BytesCompressor { + private static final Constructor BROTLI_OS_CTOR; + + static { + Constructor ctor = null; + try { + Class bosClass = Class.forName("com.aayushatharva.brotli4j.encoder.BrotliOutputStream"); + ctor = bosClass.getConstructor(OutputStream.class); + } catch (Throwable t) { + // will be null — checked before use + } + BROTLI_OS_CTOR = ctor; + } + + @Override + public BytesInput compress(BytesInput bytes) throws IOException { + try { + ByteArrayOutputStream baos = new ByteArrayOutputStream((int) bytes.size()); + OutputStream bos = (OutputStream) BROTLI_OS_CTOR.newInstance(baos); + bytes.writeAllTo(bos); + bos.close(); + return BytesInput.from(baos.toByteArray()); + } catch (ReflectiveOperationException e) { + throw new IOException("Brotli stream compression failed", e); + } + } + + @Override + public CompressionCodecName getCodecName() { + return CompressionCodecName.BROTLI; + } + + @Override + public void release() {} + } + + /** + * Stream-based Brotli decompressor using brotli4j's BrotliInputStream via reflection. + * This mimics what a Hadoop BrotliCodec would do: wrap the input in a BrotliInputStream. + */ + static class BrotliStreamDecompressor extends CodecFactory.BytesDecompressor { + private static final Constructor BROTLI_IS_CTOR; + + static { + Constructor ctor = null; + try { + Class bisClass = Class.forName("com.aayushatharva.brotli4j.decoder.BrotliInputStream"); + ctor = bisClass.getConstructor(InputStream.class); + } catch (Throwable t) { + // will be null — checked before use + } + BROTLI_IS_CTOR = ctor; + } + + @Override + public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException { + try { + InputStream bis = (InputStream) BROTLI_IS_CTOR.newInstance(bytes.toInputStream()); + byte[] output = new byte[decompressedSize]; + int offset = 0; + while (offset < decompressedSize) { + int read = bis.read(output, offset, decompressedSize - offset); + if (read < 0) { + throw new IOException( + "Unexpected end of Brotli stream at offset " + offset + " of " + decompressedSize); + } + offset += read; + } + bis.close(); + return BytesInput.from(output); + } catch (ReflectiveOperationException e) { + throw new IOException("Brotli stream decompression failed", e); + } + } + + @Override + public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize) + throws IOException { + byte[] compressed = new byte[compressedSize]; + input.get(compressed); + BytesInput decompressed = decompress(BytesInput.from(compressed), decompressedSize); + output.put(decompressed.toByteArray()); + } + + @Override + public void release() {} + } +} diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDirectCodecFactory.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDirectCodecFactory.java index e4be00947a..f56d38920e 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDirectCodecFactory.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDirectCodecFactory.java @@ -33,7 +33,6 @@ import java.util.Random; import java.util.Set; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.io.compress.CompressionCodec; import org.apache.parquet.bytes.ByteBufferAllocator; import org.apache.parquet.bytes.ByteBufferReleaser; import org.apache.parquet.bytes.BytesInput; @@ -42,6 +41,7 @@ import org.apache.parquet.bytes.TrackingByteBufferAllocator; import org.apache.parquet.compression.CompressionCodecFactory.BytesInputCompressor; import org.apache.parquet.compression.CompressionCodecFactory.BytesInputDecompressor; +import org.apache.parquet.hadoop.codec.ZstandardCodec; import org.apache.parquet.hadoop.metadata.CompressionCodecName; import org.junit.Assert; import org.junit.Test; @@ -85,14 +85,6 @@ private void test(int size, CompressionCodecName codec, boolean useOnHeapCompres final BytesInputCompressor heapCompressor = heapCodecFactory.getCompressor(codec); final BytesInputDecompressor heapDecompressor = heapCodecFactory.getDecompressor(codec); - if (codec == LZ4_RAW) { - // Hadoop codecs support direct decompressors only if the related native libraries are available. - // This is not the case for our CI so let's rely on LZ4_RAW where the implementation is our own. - Assert.assertTrue( - String.format("The hadoop codec %s should support direct decompression", codec), - directDecompressor instanceof DirectCodecFactory.FullDirectDecompressor); - } - final BytesInput directCompressed; if (useOnHeapCompression) { directCompressed = directCompressor.compress(BytesInput.from(rawArr)); @@ -219,13 +211,7 @@ public void compressionCodecs() { final int[] sizes = {4 * 1024, 1 * 1024 * 1024}; final boolean[] comp = {true, false}; Set codecsToSkip = new HashSet<>(); - codecsToSkip.add(LZO); // not distributed because it is GPL codecsToSkip.add(LZ4); // not distributed in the default version of Hadoop - final String arch = System.getProperty("os.arch"); - if ("aarch64".equals(arch)) { - // PARQUET-1975 brotli-codec does not have natives for ARM64 architectures - codecsToSkip.add(BROTLI); - } for (final int size : sizes) { for (final boolean useOnHeapComp : comp) { @@ -241,54 +227,402 @@ public void compressionCodecs() { } } - static class PublicCodecFactory extends CodecFactory { - // To make getCodec public + @Test + public void compressionLevelGzip() throws IOException { + Configuration config_zlib_1 = new Configuration(); + config_zlib_1.set("zlib.compress.level", "1"); + + Configuration config_zlib_9 = new Configuration(); + config_zlib_9.set("zlib.compress.level", "9"); + + // Generate compressible data so different levels produce different sizes + byte[] data = new byte[64 * 1024]; + new Random(42).nextBytes(data); + + final CodecFactory codecFactory_1 = new CodecFactory(config_zlib_1, pageSize); + final CodecFactory codecFactory_9 = new CodecFactory(config_zlib_9, pageSize); + + BytesInputCompressor compressor_1 = codecFactory_1.getCompressor(CompressionCodecName.GZIP); + BytesInputCompressor compressor_9 = codecFactory_9.getCompressor(CompressionCodecName.GZIP); + + long size_1 = compressor_1.compress(BytesInput.from(data)).size(); + long size_9 = compressor_9.compress(BytesInput.from(data)).size(); + + // Level 9 should produce smaller (or equal) output than level 1 + Assert.assertTrue("Expected level 9 (" + size_9 + ") <= level 1 (" + size_1 + ")", size_9 <= size_1); + + codecFactory_1.release(); + codecFactory_9.release(); + } + + @Test + public void compressionLevelZstd() throws IOException { + Configuration config_zstd_1 = new Configuration(); + config_zstd_1.set("parquet.compression.codec.zstd.level", "1"); + + Configuration config_zstd_19 = new Configuration(); + config_zstd_19.set("parquet.compression.codec.zstd.level", "19"); + + // Generate compressible data so different levels produce different sizes + byte[] data = new byte[64 * 1024]; + new Random(42).nextBytes(data); - public PublicCodecFactory(Configuration configuration, int pageSize) { - super(configuration, pageSize); + final CodecFactory codecFactory_1 = new CodecFactory(config_zstd_1, pageSize); + final CodecFactory codecFactory_19 = new CodecFactory(config_zstd_19, pageSize); + + BytesInputCompressor compressor_1 = codecFactory_1.getCompressor(CompressionCodecName.ZSTD); + BytesInputCompressor compressor_19 = codecFactory_19.getCompressor(CompressionCodecName.ZSTD); + + long size_1 = compressor_1.compress(BytesInput.from(data)).size(); + long size_19 = compressor_19.compress(BytesInput.from(data)).size(); + + // Level 19 should produce smaller (or equal) output than level 1 + Assert.assertTrue("Expected level 19 (" + size_19 + ") <= level 1 (" + size_1 + ")", size_19 <= size_1); + + codecFactory_1.release(); + codecFactory_19.release(); + } + + // ---- Tests for empty input (0 bytes) through direct compressor/decompressor path ---- + + @Test + public void emptyInputRoundTrip() throws IOException { + // Codecs that have direct bypass implementations in CodecFactory + CompressionCodecName[] directCodecs = {SNAPPY, ZSTD, LZ4_RAW, GZIP, LZO, BROTLI}; + for (CompressionCodecName codec : directCodecs) { + CodecFactory factory = new CodecFactory(new Configuration(), pageSize); + BytesInputCompressor compressor = factory.getCompressor(codec); + BytesInputDecompressor decompressor = factory.getDecompressor(codec); + + BytesInput compressed = compressor.compress(BytesInput.from(new byte[0])); + BytesInput decompressed = decompressor.decompress(compressed, 0); + Assert.assertEquals("Empty input round-trip failed for " + codec, 0, decompressed.toByteArray().length); + + compressor.release(); + decompressor.release(); + factory.release(); } + } + + // ---- Tests for GZIP consecutive compressions with a single compressor instance ---- + + @Test + public void gzipConsecutiveCompressionsProduceCorrectResults() throws IOException { + CodecFactory factory = new CodecFactory(new Configuration(), pageSize); + BytesInputCompressor compressor = factory.getCompressor(GZIP); + BytesInputDecompressor decompressor = factory.getDecompressor(GZIP); - public org.apache.hadoop.io.compress.CompressionCodec getCodec(CompressionCodecName name) { - return super.getCodec(name); + Random r = new Random(99); + for (int i = 0; i < 10; i++) { + byte[] data = new byte[4096 + i * 1024]; + r.nextBytes(data); + + BytesInput compressed = compressor.compress(BytesInput.from(data)); + BytesInput decompressed = decompressor.decompress(compressed, data.length); + Assert.assertArrayEquals( + "GZIP consecutive round-trip failed on iteration " + i, data, decompressed.toByteArray()); } + + compressor.release(); + decompressor.release(); + factory.release(); } + // ---- Tests for buffer reuse safety in Snappy/LZ4_RAW compressors ---- + @Test - public void cachingKeysGzip() { - Configuration config_zlib_2 = new Configuration(); - config_zlib_2.set("zlib.compress.level", "2"); + public void snappyCompressorBufferReuseSafety() throws IOException { + verifyCompressorOutputCopiedBeforeReuse(SNAPPY); + } - Configuration config_zlib_5 = new Configuration(); - config_zlib_5.set("zlib.compress.level", "5"); + @Test + public void lz4RawCompressorBufferReuseSafety() throws IOException { + verifyCompressorOutputCopiedBeforeReuse(LZ4_RAW); + } - final CodecFactory codecFactory_2 = new PublicCodecFactory(config_zlib_2, pageSize); - final CodecFactory codecFactory_5 = new PublicCodecFactory(config_zlib_5, pageSize); + /** + * Verifies that the caller can safely copy the compressed output before the next + * compress() call overwrites the internal buffer. This is the documented contract + * for compressors that reuse output buffers. + */ + private void verifyCompressorOutputCopiedBeforeReuse(CompressionCodecName codec) throws IOException { + CodecFactory factory = new CodecFactory(new Configuration(), pageSize); + BytesInputCompressor compressor = factory.getCompressor(codec); + BytesInputDecompressor decompressor = factory.getDecompressor(codec); + + byte[] data1 = new byte[4096]; + byte[] data2 = new byte[4096]; + new Random(1).nextBytes(data1); + new Random(2).nextBytes(data2); + + // Compress first, copy result immediately + BytesInput compressed1 = compressor.compress(BytesInput.from(data1)); + byte[] compressed1Bytes = compressed1.toByteArray(); + + // Compress second (may overwrite internal buffer) + BytesInput compressed2 = compressor.compress(BytesInput.from(data2)); + byte[] compressed2Bytes = compressed2.toByteArray(); + + // Both should decompress correctly from the copied bytes + BytesInput decompressed1 = decompressor.decompress(BytesInput.from(compressed1Bytes), data1.length); + Assert.assertArrayEquals(codec + " buffer reuse: first input corrupted", data1, decompressed1.toByteArray()); + + BytesInput decompressed2 = decompressor.decompress(BytesInput.from(compressed2Bytes), data2.length); + Assert.assertArrayEquals(codec + " buffer reuse: second input corrupted", data2, decompressed2.toByteArray()); + + compressor.release(); + decompressor.release(); + factory.release(); + } - CompressionCodec codec_2_1 = codecFactory_2.getCodec(CompressionCodecName.GZIP); - CompressionCodec codec_2_2 = codecFactory_2.getCodec(CompressionCodecName.GZIP); - CompressionCodec codec_5_1 = codecFactory_5.getCodec(CompressionCodecName.GZIP); + // ---- Tests for ZSTD bufferPool config propagation through new direct compressor ---- - Assert.assertEquals(codec_2_1, codec_2_2); - Assert.assertNotEquals(codec_2_1, codec_5_1); + @Test + public void zstdBufferPoolEnabledRoundTrip() throws IOException { + Configuration conf = new Configuration(); + conf.setBoolean(ZstandardCodec.PARQUET_COMPRESS_ZSTD_BUFFERPOOL_ENABLED, true); + verifyZstdRoundTrip(conf, "bufferPool=true"); } @Test - public void cachingKeysZstd() { - Configuration config_zstd_2 = new Configuration(); - config_zstd_2.set("parquet.compression.codec.zstd.level", "2"); + public void zstdBufferPoolDisabledRoundTrip() throws IOException { + Configuration conf = new Configuration(); + conf.setBoolean(ZstandardCodec.PARQUET_COMPRESS_ZSTD_BUFFERPOOL_ENABLED, false); + verifyZstdRoundTrip(conf, "bufferPool=false"); + } + + /** + * Verifies ZSTD round-trip with different bufferPool configurations through the + * new direct ZstdBytesCompressor/ZstdBytesDecompressor path in CodecFactory. + */ + private void verifyZstdRoundTrip(Configuration conf, String label) throws IOException { + CodecFactory factory = new CodecFactory(conf, pageSize); + BytesInputCompressor compressor = factory.getCompressor(ZSTD); + BytesInputDecompressor decompressor = factory.getDecompressor(ZSTD); + + byte[] data = new byte[64 * 1024]; + new Random(42).nextBytes(data); - Configuration config_zstd_5 = new Configuration(); - config_zstd_5.set("parquet.compression.codec.zstd.level", "5"); + BytesInput compressed = compressor.compress(BytesInput.from(data)); + BytesInput decompressed = decompressor.decompress(compressed, data.length); + Assert.assertArrayEquals("ZSTD round-trip failed with " + label, data, decompressed.toByteArray()); - final CodecFactory codecFactory_2 = new PublicCodecFactory(config_zstd_2, pageSize); - final CodecFactory codecFactory_5 = new PublicCodecFactory(config_zstd_5, pageSize); + compressor.release(); + decompressor.release(); + factory.release(); + } - CompressionCodec codec_2_1 = codecFactory_2.getCodec(CompressionCodecName.ZSTD); - CompressionCodec codec_2_2 = codecFactory_2.getCodec(CompressionCodecName.ZSTD); - CompressionCodec codec_5_1 = codecFactory_5.getCodec(CompressionCodecName.ZSTD); + // ---- Tests for ZSTD workers config propagation ---- - Assert.assertEquals(codec_2_1, codec_2_2); - Assert.assertNotEquals(codec_2_1, codec_5_1); + @Test + public void zstdWorkersConfigRoundTrip() throws IOException { + Configuration conf = new Configuration(); + conf.setInt(ZstandardCodec.PARQUET_COMPRESS_ZSTD_WORKERS, 2); + CodecFactory factory = new CodecFactory(conf, pageSize); + BytesInputCompressor compressor = factory.getCompressor(ZSTD); + BytesInputDecompressor decompressor = factory.getDecompressor(ZSTD); + + byte[] data = new byte[64 * 1024]; + new Random(42).nextBytes(data); + + BytesInput compressed = compressor.compress(BytesInput.from(data)); + BytesInput decompressed = decompressor.decompress(compressed, data.length); + Assert.assertArrayEquals("ZSTD round-trip failed with workers=2", data, decompressed.toByteArray()); + + compressor.release(); + decompressor.release(); + factory.release(); + } + + // ---- Tests for ZSTD level through the direct CodecFactory path ---- + + @Test + public void zstdLevelConfigThroughDirectPath() throws IOException { + Configuration confLow = new Configuration(); + confLow.setInt(ZstandardCodec.PARQUET_COMPRESS_ZSTD_LEVEL, 1); + + Configuration confHigh = new Configuration(); + confHigh.setInt(ZstandardCodec.PARQUET_COMPRESS_ZSTD_LEVEL, 19); + + byte[] data = new byte[64 * 1024]; + new Random(42).nextBytes(data); + + CodecFactory factoryLow = new CodecFactory(confLow, pageSize); + CodecFactory factoryHigh = new CodecFactory(confHigh, pageSize); + + long sizeLow = + factoryLow.getCompressor(ZSTD).compress(BytesInput.from(data)).size(); + long sizeHigh = + factoryHigh.getCompressor(ZSTD).compress(BytesInput.from(data)).size(); + + Assert.assertTrue( + "Expected ZSTD level 19 (" + sizeHigh + ") <= level 1 (" + sizeLow + ")", sizeHigh <= sizeLow); + + factoryLow.release(); + factoryHigh.release(); + } + + // ---- Tests for GZIP level through the direct CodecFactory path ---- + + @Test + public void gzipLevelConfigThroughDirectPath() throws IOException { + Configuration confLow = new Configuration(); + confLow.setInt("zlib.compress.level", 1); + + Configuration confHigh = new Configuration(); + confHigh.setInt("zlib.compress.level", 9); + + byte[] data = new byte[64 * 1024]; + new Random(42).nextBytes(data); + + CodecFactory factoryLow = new CodecFactory(confLow, pageSize); + CodecFactory factoryHigh = new CodecFactory(confHigh, pageSize); + + BytesInputCompressor compLow = factoryLow.getCompressor(GZIP); + BytesInputCompressor compHigh = factoryHigh.getCompressor(GZIP); + + long sizeLow = compLow.compress(BytesInput.from(data)).size(); + long sizeHigh = compHigh.compress(BytesInput.from(data)).size(); + + Assert.assertTrue("Expected GZIP level 9 (" + sizeHigh + ") <= level 1 (" + sizeLow + ")", sizeHigh <= sizeLow); + + // Also verify round-trip for both levels + BytesInputDecompressor decompLow = factoryLow.getDecompressor(GZIP); + BytesInputDecompressor decompHigh = factoryHigh.getDecompressor(GZIP); + + Assert.assertArrayEquals( + data, + decompLow + .decompress(compLow.compress(BytesInput.from(data)), data.length) + .toByteArray()); + Assert.assertArrayEquals( + data, + decompHigh + .decompress(compHigh.compress(BytesInput.from(data)), data.length) + .toByteArray()); + + factoryLow.release(); + factoryHigh.release(); + } + + // ---- Tests for BROTLI direct bypass (when native lib available) ---- + + @Test + public void brotliDirectFactoryRoundTrip() throws IOException { + // Test through the DirectCodecFactory path where BROTLI bypass lives + try (TrackingByteBufferAllocator alloc = TrackingByteBufferAllocator.wrap(new DirectByteBufferAllocator())) { + CodecFactory directFactory = CodecFactory.createDirectCodecFactory(new Configuration(), alloc, pageSize); + BytesInputCompressor compressor = directFactory.getCompressor(BROTLI); + BytesInputDecompressor decompressor = directFactory.getDecompressor(BROTLI); + + // Use compressible data (repeated patterns) so compression is verifiable + byte[] data = new byte[16 * 1024]; + for (int i = 0; i < data.length; i++) { + data[i] = (byte) (i % 251); + } + + BytesInput compressed = compressor.compress(BytesInput.from(data)); + BytesInput decompressed = decompressor.decompress(compressed, data.length); + Assert.assertArrayEquals("BROTLI direct round-trip failed", data, decompressed.toByteArray()); + + // Test multiple consecutive compressions to verify state management + for (int i = 0; i < 5; i++) { + byte[] moreData = new byte[8 * 1024 + i * 1024]; + for (int j = 0; j < moreData.length; j++) { + moreData[j] = (byte) ((j + i) % 251); + } + BytesInput moreCompressed = compressor.compress(BytesInput.from(moreData)); + BytesInput moreDecompressed = decompressor.decompress(moreCompressed, moreData.length); + Assert.assertArrayEquals( + "BROTLI direct round-trip failed on iteration " + i, moreData, moreDecompressed.toByteArray()); + } + + compressor.release(); + decompressor.release(); + directFactory.release(); + } + } + + // ---- Test for cross-factory interop with new direct codecs ---- + + @Test + public void crossFactoryInteropAllDirectCodecs() throws IOException { + CompressionCodecName[] codecs = {SNAPPY, ZSTD, LZ4_RAW, GZIP, LZO, BROTLI}; + + byte[] data = new byte[32 * 1024]; + new Random(42).nextBytes(data); + + CodecFactory heapFactory = new CodecFactory(new Configuration(), pageSize); + try (TrackingByteBufferAllocator alloc = TrackingByteBufferAllocator.wrap(new DirectByteBufferAllocator())) { + CodecFactory directFactory = CodecFactory.createDirectCodecFactory(new Configuration(), alloc, pageSize); + + for (CompressionCodecName codec : codecs) { + // heap compress -> direct decompress + BytesInput heapCompressed = heapFactory.getCompressor(codec).compress(BytesInput.from(data)); + BytesInput directDecompressed = + directFactory.getDecompressor(codec).decompress(heapCompressed, data.length); + Assert.assertArrayEquals(codec + " heap->direct failed", data, directDecompressed.toByteArray()); + + // direct compress -> heap decompress + BytesInput directCompressed = directFactory.getCompressor(codec).compress(BytesInput.from(data)); + BytesInput heapDecompressed = + heapFactory.getDecompressor(codec).decompress(directCompressed, data.length); + Assert.assertArrayEquals(codec + " direct->heap failed", data, heapDecompressed.toByteArray()); + } + + directFactory.release(); + } + heapFactory.release(); + } + + @Test + public void zstdCompressorBufferReuseSafety() throws IOException { + verifyCompressorOutputCopiedBeforeReuse(ZSTD); + } + + @Test + public void brotliQualityConfigProducesValidOutput() throws IOException { + byte[] data = new byte[16 * 1024]; + for (int i = 0; i < data.length; i++) { + data[i] = (byte) (i % 251); + } + + for (int quality : new int[] {0, 1, 6, 11}) { + Configuration conf = new Configuration(); + conf.setInt("compression.brotli.quality", quality); + CodecFactory factory = new CodecFactory(conf, pageSize); + BytesInputCompressor compressor = factory.getCompressor(BROTLI); + BytesInputDecompressor decompressor = factory.getDecompressor(BROTLI); + + BytesInput compressed = compressor.compress(BytesInput.from(data)); + BytesInput decompressed = decompressor.decompress(compressed, data.length); + Assert.assertArrayEquals( + "BROTLI quality=" + quality + " round-trip failed", data, decompressed.toByteArray()); + + compressor.release(); + decompressor.release(); + factory.release(); + } + } + + @Test + public void singleByteInputRoundTrip() throws IOException { + CompressionCodecName[] codecs = {SNAPPY, ZSTD, LZ4_RAW, GZIP, LZO, BROTLI}; + byte[] data = new byte[] {42}; + + for (CompressionCodecName codec : codecs) { + CodecFactory factory = new CodecFactory(new Configuration(), pageSize); + BytesInputCompressor compressor = factory.getCompressor(codec); + BytesInputDecompressor decompressor = factory.getDecompressor(codec); + + BytesInput compressed = compressor.compress(BytesInput.from(data)); + BytesInput decompressed = decompressor.decompress(compressed, 1); + Assert.assertArrayEquals("Single-byte round-trip failed for " + codec, data, decompressed.toByteArray()); + + compressor.release(); + decompressor.release(); + factory.release(); + } } @Test @@ -391,16 +725,17 @@ public void directFactoryLeveledZstdRoundTrip() throws IOException { CodecFactory direct = CodecFactory.createDirectCodecFactory(new Configuration(), new DirectByteBufferAllocator(), pageSize); try { - // Sanity: the heap factory's leveled path yields a HeapBytesCompressor. + // Sanity: the heap factory's leveled path uses the heap byte-array ZSTD compressor. Assert.assertTrue( - "heap factory should produce a HeapBytesCompressor", - heap.getCompressor(ZSTD, 3) instanceof CodecFactory.HeapBytesCompressor); + "heap factory should produce a heap ZstdBytesCompressor", + heap.getCompressor(ZSTD, 3) instanceof CodecFactory.ZstdBytesCompressor); - // The direct factory must not fall back to the heap/Hadoop path for leveled ZSTD/SNAPPY. + // The direct factory must stay on its direct compressors for leveled ZSTD/SNAPPY, + // not fall back to the heap byte-array implementations. BytesInputCompressor directZstd = direct.getCompressor(ZSTD, 3); Assert.assertFalse( - "direct factory ZSTD(level) should not fall back to HeapBytesCompressor", - directZstd instanceof CodecFactory.HeapBytesCompressor); + "direct factory ZSTD(level) should not fall back to the heap compressor", + directZstd instanceof CodecFactory.ZstdBytesCompressor); Assert.assertEquals(ZSTD, directZstd.getCodecName()); Assert.assertEquals( "leveled ZSTD should use the same direct compressor type as the no-level path", @@ -409,8 +744,8 @@ public void directFactoryLeveledZstdRoundTrip() throws IOException { BytesInputCompressor directSnappy = direct.getCompressor(SNAPPY, 5); Assert.assertFalse( - "direct factory SNAPPY(level) should not fall back to HeapBytesCompressor", - directSnappy instanceof CodecFactory.HeapBytesCompressor); + "direct factory SNAPPY(level) should not fall back to the heap compressor", + directSnappy instanceof CodecFactory.SnappyBytesCompressor); Assert.assertEquals( "leveled SNAPPY should use the same direct compressor type as the no-level path", direct.getCompressor(SNAPPY).getClass(), diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestZstandardCodec.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestZstandardCodec.java index 541faf4f6a..ea85f08aa2 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestZstandardCodec.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestZstandardCodec.java @@ -101,8 +101,6 @@ private byte[] decompress(ZstandardCodec codec, BytesInput bytes, int uncompress @Test public void testZstdConfWithMr() throws Exception { long fileSizeLowLevel = runMrWithConf(1); - // Clear the cache so that a new codec can be created with new configuration - CodecFactory.CODEC_BY_NAME.clear(); long fileSizeHighLevel = runMrWithConf(22); Assert.assertTrue(fileSizeLowLevel > fileSizeHighLevel); } diff --git a/pom.xml b/pom.xml index 62bdb32623..8e885ed305 100644 --- a/pom.xml +++ b/pom.xml @@ -96,10 +96,9 @@ 1.7.33 1.11.5 33.6.0-jre - 0.1.1 + 1.23.0 3.27.7 5.14.4 - 3.27.7 5.23.0 0.27ea1 3.6.1