wgtmac commented on code in PR #3570:
URL: https://github.com/apache/parquet-java/pull/3570#discussion_r4092200325
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java:
##########
@@ -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(conf.getInt("compression.brotli.quality", 1));
Review Comment:
The level-aware BROTLI path falls back to the heap compressor, while the
no-level path uses the direct compressor. Could we keep this path direct when
brotli4j is available?
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java:
##########
@@ -292,22 +314,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 = gzipLevel(conf.get("zlib.compress.level"));
+ return new GzipBytesCompressor(gzipLevel, pageSize);
+ case LZO:
+ return new LzoBytesCompressor(pageSize);
+ case BROTLI:
+ if (Brotli4j.AVAILABLE) {
+ int brotliQuality = conf.getInt("compression.brotli.quality", 1);
+ validateBrotliLevel(brotliQuality);
+ return new BrotliBytesCompressor(brotliQuality);
+ }
+ return new HadoopBytesCompressor(codecName, getCodec(codecName),
pageSize);
+ 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)");
Review Comment:
The no-level BROTLI path can still use the Hadoop codec when brotli4j is
missing. This level-aware path throws instead. Could we keep the fallback for
callers that set a Brotli level?
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java:
##########
@@ -503,4 +564,764 @@ public abstract void decompress(ByteBuffer input, int
compressedSize, ByteBuffer
public abstract void release();
}
+
+ private static class HadoopBytesCompressor extends BytesCompressor {
+ private final CompressionCodecName codecName;
+ private final CompressionCodec codec;
+ private final Compressor compressor;
+ private final ByteArrayOutputStream output;
+
+ HadoopBytesCompressor(CompressionCodecName codecName, CompressionCodec
codec, int pageSize) {
+ this.codecName = codecName;
+ this.codec = Objects.requireNonNull(codec);
+ this.compressor = CodecPool.getCompressor(codec);
+ this.output = new ByteArrayOutputStream(pageSize);
+ }
+
+ @Override
+ public BytesInput compress(BytesInput bytes) throws IOException {
+ output.reset();
+ if (compressor != null) {
+ compressor.reset();
+ }
+ try (CompressionOutputStream stream = codec.createOutputStream(output,
compressor)) {
+ bytes.writeAllTo(stream);
+ stream.finish();
+ }
+ return BytesInput.from(output);
+ }
+
+ @Override
+ public CompressionCodecName getCodecName() {
+ return codecName;
+ }
+
+ @Override
+ public void release() {
+ if (compressor != null) {
+ CodecPool.returnCompressor(compressor);
+ }
+ }
+ }
+
+ private static class HadoopBytesDecompressor extends BytesDecompressor {
+ private final CompressionCodec codec;
+ private final Decompressor decompressor;
+
+ HadoopBytesDecompressor(CompressionCodec codec) {
+ this.codec = Objects.requireNonNull(codec);
+ this.decompressor = CodecPool.getDecompressor(codec);
+ }
+
+ @Override
+ public BytesInput decompress(BytesInput bytes, int decompressedSize)
throws IOException {
+ if (decompressor != null) {
+ decompressor.reset();
+ }
+ try (InputStream stream = codec.createInputStream(bytes.toInputStream(),
decompressor)) {
+ byte[] output = new byte[decompressedSize];
+ int offset = 0;
+ while (offset < decompressedSize) {
+ int read = stream.read(output, offset, decompressedSize - offset);
+ if (read < 0) {
+ throw new IOException(
+ "Unexpected end of compressed stream at " + offset + " of " +
decompressedSize);
+ }
+ offset += read;
+ }
+ ensureEndOfStream(stream, decompressedSize,
codec.getClass().getSimpleName());
+ 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);
+ output.put(decompress(BytesInput.from(inputSlice),
decompressedSize).toByteBuffer());
+ input.position(input.position() + compressedSize);
+ }
+
+ @Override
+ public void release() {
+ if (decompressor != null) {
+ CodecPool.returnDecompressor(decompressor);
+ }
+ }
+ }
+
+ // ---- 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];
+ int decompressed = Snappy.uncompress(input, 0, input.length, output, 0);
+ 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 {
+ ByteBuffer inputSlice = input.slice();
+ inputSlice.limit(compressedSize);
+ ByteBuffer outputSlice = output.slice();
+ outputSlice.limit(decompressedSize);
+ int decompressed = Snappy.uncompress(inputSlice, outputSlice);
+ if (decompressed != decompressedSize) {
Review Comment:
This ByteBuffer overload also accepts heap buffers, but Snappy requires
direct buffers here. The ZSTD path below has the same issue. Could we keep heap
buffers working and add a test for both codecs?
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java:
##########
@@ -503,4 +564,764 @@ public abstract void decompress(ByteBuffer input, int
compressedSize, ByteBuffer
public abstract void release();
}
+
+ private static class HadoopBytesCompressor extends BytesCompressor {
+ private final CompressionCodecName codecName;
+ private final CompressionCodec codec;
+ private final Compressor compressor;
+ private final ByteArrayOutputStream output;
+
+ HadoopBytesCompressor(CompressionCodecName codecName, CompressionCodec
codec, int pageSize) {
+ this.codecName = codecName;
+ this.codec = Objects.requireNonNull(codec);
+ this.compressor = CodecPool.getCompressor(codec);
+ this.output = new ByteArrayOutputStream(pageSize);
+ }
+
+ @Override
+ public BytesInput compress(BytesInput bytes) throws IOException {
+ output.reset();
+ if (compressor != null) {
+ compressor.reset();
+ }
+ try (CompressionOutputStream stream = codec.createOutputStream(output,
compressor)) {
+ bytes.writeAllTo(stream);
+ stream.finish();
+ }
+ return BytesInput.from(output);
+ }
+
+ @Override
+ public CompressionCodecName getCodecName() {
+ return codecName;
+ }
+
+ @Override
+ public void release() {
+ if (compressor != null) {
+ CodecPool.returnCompressor(compressor);
+ }
+ }
+ }
+
+ private static class HadoopBytesDecompressor extends BytesDecompressor {
+ private final CompressionCodec codec;
+ private final Decompressor decompressor;
+
+ HadoopBytesDecompressor(CompressionCodec codec) {
+ this.codec = Objects.requireNonNull(codec);
+ this.decompressor = CodecPool.getDecompressor(codec);
+ }
+
+ @Override
+ public BytesInput decompress(BytesInput bytes, int decompressedSize)
throws IOException {
+ if (decompressor != null) {
+ decompressor.reset();
+ }
+ try (InputStream stream = codec.createInputStream(bytes.toInputStream(),
decompressor)) {
+ byte[] output = new byte[decompressedSize];
+ int offset = 0;
+ while (offset < decompressedSize) {
+ int read = stream.read(output, offset, decompressedSize - offset);
+ if (read < 0) {
+ throw new IOException(
+ "Unexpected end of compressed stream at " + offset + " of " +
decompressedSize);
+ }
+ offset += read;
+ }
+ ensureEndOfStream(stream, decompressedSize,
codec.getClass().getSimpleName());
+ 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);
+ output.put(decompress(BytesInput.from(inputSlice),
decompressedSize).toByteBuffer());
+ input.position(input.position() + compressedSize);
+ }
+
+ @Override
+ public void release() {
+ if (decompressor != null) {
+ CodecPool.returnDecompressor(decompressor);
+ }
+ }
+ }
+
+ // ---- 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];
+ int decompressed = Snappy.uncompress(input, 0, input.length, output, 0);
+ 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 {
+ ByteBuffer inputSlice = input.slice();
+ inputSlice.limit(compressedSize);
+ ByteBuffer outputSlice = output.slice();
+ outputSlice.limit(decompressedSize);
+ int decompressed = Snappy.uncompress(inputSlice, outputSlice);
+ if (decompressed != decompressedSize) {
+ throw new IOException("Unexpected decompressed size: " + decompressed
+ " != " + decompressedSize);
+ }
+ input.position(input.position() + compressedSize);
+ output.position(output.position() + decompressedSize);
+ }
+
+ @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)}.
+ *
+ * <p>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 = Math.toIntExact(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 {
+ ByteBuffer inputSlice = input.slice();
+ inputSlice.limit(compressedSize);
+ ByteBuffer outputSlice = output.slice();
+ outputSlice.limit(decompressedSize);
+ // Zstd.decompress uses (dst, src) parameter order, matching the native
zstd convention.
+ int decompressed = Zstd.decompress(outputSlice, inputSlice);
+ if (decompressed != decompressedSize) {
+ throw new IOException("Unexpected decompressed size: " + decompressed
+ " != " + decompressedSize);
+ }
+ input.position(input.position() + compressedSize);
+ output.position(output.position() + decompressedSize);
+ }
+
+ @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.
+ *
+ * <p>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);
+ 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);
+
+ ByteBuffer outputSlice = directOutputBuf.slice();
+ decompressor.decompress(directInputBuf.slice(), outputSlice);
+ int decompressed = outputSlice.position();
+ if (decompressed != decompressedSize) {
+ throw new IOException("Unexpected decompressed size: " + decompressed
+ " != " + decompressedSize);
+ }
+
+ // 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 {
+ ByteBuffer inputSlice = input.slice();
+ inputSlice.limit(compressedSize);
+ ByteBuffer outputSlice = output.slice();
+ outputSlice.limit(decompressedSize);
+ decompressor.decompress(inputSlice, outputSlice);
+ int decompressed = outputSlice.position();
+ if (decompressed != decompressedSize) {
+ throw new IOException("Unexpected decompressed size: " + decompressed
+ " != " + decompressedSize);
+ }
+ input.position(input.position() + compressedSize);
+ output.position(output.position() + decompressedSize);
+ }
+
+ @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.
+ *
+ * <p>Note: this implementation always uses Java's built-in zlib via
+ * {@link GZIPOutputStream}. It does <em>not</em> 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.
+ *
+ * <p>Note: this implementation always uses Java's built-in zlib via
+ * {@link GZIPInputStream}. It does <em>not</em> 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;
+ }
+ ensureEndOfStream(gis, decompressedSize, "GZIP");
+ 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;
+ }
+ ensureEndOfStream(gis, decompressedSize, "GZIP");
+ 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);
Review Comment:
This path ignores the existing Hadoop LZO strategy, level, and buffer-size
settings. Could we preserve those settings where possible, or document the
change and test old files made with non-default settings?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
