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): + * + *
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: + * + *
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 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
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: + *
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: + *
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