From a82f5715f1c0f64c6e098766c01487f7c35c138c Mon Sep 17 00:00:00 2001 From: sahmadsabri Date: Mon, 3 Aug 2026 21:53:15 -0500 Subject: [PATCH 01/12] Implement GzipByteBuffDecompressor and integrate with ReusableStreamGzipCodec --- .../io/compress/GzipByteBuffDecompressor.java | 160 ++++++++++ .../GzipHFileDecompressionContext.java | 66 ++++ .../io/compress/ReusableStreamGzipCodec.java | 19 +- .../TestGzipByteBuffDecompressor.java | 299 ++++++++++++++++++ .../io/compress/TestHFileCompressionGzip.java | 72 +++++ 5 files changed, 615 insertions(+), 1 deletion(-) create mode 100644 hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java create mode 100644 hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipHFileDecompressionContext.java create mode 100644 hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java create mode 100644 hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java new file mode 100644 index 000000000000..17970eff3e51 --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java @@ -0,0 +1,160 @@ +/* + * 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.hadoop.hbase.io.compress; + +import edu.umd.cs.findbugs.annotations.Nullable; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.zip.CRC32; +import java.util.zip.DataFormatException; +import java.util.zip.Inflater; +import org.apache.hadoop.hbase.nio.ByteBuff; +import org.apache.hadoop.hbase.nio.SingleByteBuff; +import org.apache.yetus.audience.InterfaceAudience; + +/** + * Glue for ByteBuffDecompressor on top of {@link Inflater}. Only supports gzip members with the + * fixed ten-byte header that {@link ReusableStreamGzipCodec} (and Hadoop's native zlib gzip + * compressor) always writes, i.e. no FEXTRA/FNAME/FCOMMENT/FHCRC, since that is the only format + * HBase ever produces on the compression side. + */ +@InterfaceAudience.Private +public class GzipByteBuffDecompressor implements ByteBuffDecompressor { + + private static final int GZIP_HEADER_LENGTH = 10; + private static final int GZIP_TRAILER_LENGTH = 8; + private static final byte GZIP_MAGIC_0 = (byte) 0x1f; + private static final byte GZIP_MAGIC_1 = (byte) 0x8b; + + private final Inflater inflater = new Inflater(true); + // Intended to be set to false by some unit tests + private boolean allowByteBuffDecompression; + + GzipByteBuffDecompressor() { + allowByteBuffDecompression = true; + } + + @Override + public boolean canDecompress(ByteBuff output, ByteBuff input) { + return allowByteBuffDecompression && output instanceof SingleByteBuff + && input instanceof SingleByteBuff; + } + + @Override + public int decompress(ByteBuff output, ByteBuff input, int inputLen) throws IOException { + if (!(output instanceof SingleByteBuff) || !(input instanceof SingleByteBuff)) { + throw new IllegalStateException( + "At least one buffer is not a SingleByteBuff, this is not supported"); + } + if (inputLen < GZIP_HEADER_LENGTH + GZIP_TRAILER_LENGTH) { + throw new IOException("Input of length " + inputLen + " is too short to be a gzip member"); + } + + ByteBuffer nioInput = input.nioByteBuffers()[0]; + int inputStart = nioInput.position(); + if (nioInput.get(inputStart) != GZIP_MAGIC_0 || nioInput.get(inputStart + 1) != GZIP_MAGIC_1) { + throw new IOException("Not a gzip member, bad magic bytes"); + } + + ByteBuffer nioOutput = output.nioByteBuffers()[0]; + + // Isolate the raw DEFLATE payload (strip the fixed header and the CRC32/ISIZE trailer) into + // its own view so Inflater can consume it without disturbing nioInput's own position/limit. + ByteBuffer deflateStream = nioInput.duplicate(); + deflateStream.limit(inputStart + inputLen - GZIP_TRAILER_LENGTH); + deflateStream.position(inputStart + GZIP_HEADER_LENGTH); + + inflater.reset(); + inflater.setInput(deflateStream); + int outputStart = nioOutput.position(); + try { + while (!inflater.finished()) { + if (inflater.inflate(nioOutput) == 0) { + if (inflater.finished()) { + break; + } + if (inflater.needsInput()) { + throw new IOException("Unexpected end of gzip stream"); + } + if (!nioOutput.hasRemaining()) { + throw new IOException("Output buffer is too small for the decompressed gzip stream"); + } + } + } + } catch (DataFormatException e) { + throw new IOException("Invalid gzip stream", e); + } + + int decompressedLength = nioOutput.position() - outputStart; + verifyTrailer(nioInput, inputStart, inputLen, nioOutput, outputStart, decompressedLength); + + nioInput.position(inputStart + inputLen); + return decompressedLength; + } + + /** + * {@link Inflater} runs in nowrap mode and never looks at the gzip header or trailer, so this is + * the only place the CRC32 and ISIZE fields of the trailer are ever checked. Catches the case + * where the raw DEFLATE payload decoded "successfully" (no {@link DataFormatException}) but + * produced the wrong bytes or the wrong number of bytes. + */ + private void verifyTrailer(ByteBuffer nioInput, int inputStart, int inputLen, + ByteBuffer nioOutput, int outputStart, int decompressedLength) throws IOException { + ByteBuffer trailer = nioInput.duplicate().order(ByteOrder.LITTLE_ENDIAN); + trailer.position(inputStart + inputLen - GZIP_TRAILER_LENGTH); + int expectedCrc32 = trailer.getInt(); + int expectedISize = trailer.getInt(); + + if (decompressedLength != expectedISize) { + throw new IOException("Decompressed length " + decompressedLength + + " does not match gzip trailer ISIZE " + expectedISize); + } + + CRC32 crc32 = new CRC32(); + ByteBuffer writtenOutput = nioOutput.duplicate(); + writtenOutput.limit(nioOutput.position()); + writtenOutput.position(outputStart); + crc32.update(writtenOutput); + if ((int) crc32.getValue() != expectedCrc32) { + throw new IOException( + "Decompressed data's CRC32 does not match gzip trailer CRC32, " + "data is corrupt"); + } + } + + @Override + public void reinit(@Nullable Compression.HFileDecompressionContext newHFileDecompressionContext) { + if (newHFileDecompressionContext == null) { + return; + } + if (!(newHFileDecompressionContext instanceof GzipHFileDecompressionContext)) { + throw new IllegalArgumentException( + "GzipByteBuffDecompressor#reinit() was given an HFileDecompressionContext that was not " + + "a GzipHFileDecompressionContext, this should never happen"); + } + GzipHFileDecompressionContext gzipContext = + (GzipHFileDecompressionContext) newHFileDecompressionContext; + allowByteBuffDecompression = gzipContext.isAllowByteBuffDecompression(); + } + + @Override + public void close() { + inflater.end(); + } + +} diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipHFileDecompressionContext.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipHFileDecompressionContext.java new file mode 100644 index 000000000000..fc94bda2ceea --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipHFileDecompressionContext.java @@ -0,0 +1,66 @@ +/* + * 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.hadoop.hbase.io.compress; + +import java.io.IOException; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.util.ClassSize; +import org.apache.yetus.audience.InterfaceAudience; + +/** + * Holds HFile-level settings used by GzipByteBuffDecompressor. It's expensive to pull these from a + * Configuration object every time we decompress a block, so pull them upon opening an HFile, and + * reuse them in every block that gets decompressed. + */ +@InterfaceAudience.Private +public final class GzipHFileDecompressionContext extends Compression.HFileDecompressionContext { + + public static final long FIXED_OVERHEAD = + ClassSize.estimateBase(GzipHFileDecompressionContext.class, false); + + // Intended to be set to false by some unit tests + private final boolean allowByteBuffDecompression; + + private GzipHFileDecompressionContext(boolean allowByteBuffDecompression) { + this.allowByteBuffDecompression = allowByteBuffDecompression; + } + + public boolean isAllowByteBuffDecompression() { + return allowByteBuffDecompression; + } + + public static GzipHFileDecompressionContext fromConfiguration(Configuration conf) { + return new GzipHFileDecompressionContext( + conf.getBoolean("hbase.io.compress.gz.allowByteBuffDecompression", true)); + } + + @Override + public void close() throws IOException { + } + + @Override + public long heapSize() { + return FIXED_OVERHEAD; + } + + @Override + public String toString() { + return "GzipHFileDecompressionContext{allowByteBuffDecompression=" + allowByteBuffDecompression + + '}'; + } +} diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/ReusableStreamGzipCodec.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/ReusableStreamGzipCodec.java index 0b3b3afbfc58..46982a46a0d7 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/ReusableStreamGzipCodec.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/ReusableStreamGzipCodec.java @@ -22,6 +22,7 @@ import java.io.OutputStream; import java.util.Arrays; import java.util.zip.GZIPOutputStream; +import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.util.JVM; import org.apache.hadoop.io.compress.CompressionOutputStream; import org.apache.hadoop.io.compress.CompressorStream; @@ -35,7 +36,7 @@ * Fixes an inefficiency in Hadoop's Gzip codec, allowing to reuse compression streams. */ @InterfaceAudience.Private -public class ReusableStreamGzipCodec extends GzipCodec { +public class ReusableStreamGzipCodec extends GzipCodec implements ByteBuffDecompressionCodec { private static final Logger LOG = LoggerFactory.getLogger(Compression.class); @@ -185,4 +186,20 @@ public CompressionOutputStream createOutputStream(OutputStream out) throws IOExc return new ReusableGzipOutputStream(out); } + @Override + public ByteBuffDecompressor createByteBuffDecompressor() { + return new GzipByteBuffDecompressor(); + } + + @Override + public Class getByteBuffDecompressorType() { + return GzipByteBuffDecompressor.class; + } + + @Override + public Compression.HFileDecompressionContext + getDecompressionContextFromConfiguration(Configuration conf) { + return GzipHFileDecompressionContext.fromConfiguration(conf); + } + } diff --git a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java new file mode 100644 index 000000000000..8fe93120a302 --- /dev/null +++ b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java @@ -0,0 +1,299 @@ +/* + * 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.hadoop.hbase.io.compress; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Arrays; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HBaseClassTestRule; +import org.apache.hadoop.hbase.nio.ByteBuff; +import org.apache.hadoop.hbase.nio.MultiByteBuff; +import org.apache.hadoop.hbase.nio.SingleByteBuff; +import org.apache.hadoop.hbase.testclassification.SmallTests; +import org.apache.hadoop.hbase.util.Bytes; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +@Category(SmallTests.class) +public class TestGzipByteBuffDecompressor { + + @ClassRule + public static final HBaseClassTestRule CLASS_RULE = + HBaseClassTestRule.forClass(TestGzipByteBuffDecompressor.class); + + /* + * "HBase is fun to use and very fast" compressed as a single gzip member via GZIPOutputStream, + * i.e. exactly the framing GzipByteBuffDecompressor expects: a fixed 10-byte header (no + * FEXTRA/FNAME/FCOMMENT/FHCRC), a raw DEFLATE stream, and an 8-byte CRC32/ISIZE trailer. + */ + private static final byte[] COMPRESSED_PAYLOAD = Bytes.fromHex( + "1f8b08000000000000fff3704a2c4e55c82c56482bcd5328c9572805f212f35214ca528b2a15d2128b4b006edf170321000000"); + + @Test + public void testCapabilities() { + ByteBuff emptySingleHeapBuff = new SingleByteBuff(ByteBuffer.allocate(0)); + ByteBuff emptyMultiHeapBuff = new MultiByteBuff(ByteBuffer.allocate(0), ByteBuffer.allocate(0)); + ByteBuff emptySingleDirectBuff = new SingleByteBuff(ByteBuffer.allocateDirect(0)); + ByteBuff emptyMultiDirectBuff = + new MultiByteBuff(ByteBuffer.allocateDirect(0), ByteBuffer.allocateDirect(0)); + + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + assertTrue(decompressor.canDecompress(emptySingleHeapBuff, emptySingleHeapBuff)); + assertTrue(decompressor.canDecompress(emptySingleDirectBuff, emptySingleDirectBuff)); + assertTrue(decompressor.canDecompress(emptySingleHeapBuff, emptySingleDirectBuff)); + assertTrue(decompressor.canDecompress(emptySingleDirectBuff, emptySingleHeapBuff)); + assertFalse(decompressor.canDecompress(emptyMultiHeapBuff, emptyMultiHeapBuff)); + assertFalse(decompressor.canDecompress(emptyMultiDirectBuff, emptyMultiDirectBuff)); + assertFalse(decompressor.canDecompress(emptySingleHeapBuff, emptyMultiHeapBuff)); + assertFalse(decompressor.canDecompress(emptySingleDirectBuff, emptyMultiDirectBuff)); + } + } + + @Test + public void testDecompressHeapToHeap() throws IOException { + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); + int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + assertEquals("HBase is fun to use and very fast", + Bytes.toString(output.toBytes(0, decompressedSize))); + } + } + + @Test + public void testDecompressDirectToDirect() throws IOException { + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); + ByteBuff input = new SingleByteBuff(ByteBuffer.allocateDirect(COMPRESSED_PAYLOAD.length)); + input.put(COMPRESSED_PAYLOAD); + input.rewind(); + int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + assertEquals("HBase is fun to use and very fast", + Bytes.toString(output.toBytes(0, decompressedSize))); + } + } + + @Test + public void testDecompressDirectToHeap() throws IOException { + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = new SingleByteBuff(ByteBuffer.allocateDirect(COMPRESSED_PAYLOAD.length)); + input.put(COMPRESSED_PAYLOAD); + input.rewind(); + int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + assertEquals("HBase is fun to use and very fast", + Bytes.toString(output.toBytes(0, decompressedSize))); + } + } + + @Test + public void testDecompressHeapToDirect() throws IOException { + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); + ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); + int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + assertEquals("HBase is fun to use and very fast", + Bytes.toString(output.toBytes(0, decompressedSize))); + } + } + + @Test + public void testDecompressFailsOnTooShortInput() { + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = new SingleByteBuff(ByteBuffer.allocate(10)); + decompressor.decompress(output, input, 10); + fail("Expected an IOException because the input is too short to be a gzip member"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("too short to be a gzip member")); + } + } + + @Test + public void testDecompressFailsOnBadMagicBytes() { + byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); + corrupted[0] ^= (byte) 0xff; + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(corrupted)); + decompressor.decompress(output, input, corrupted.length); + fail("Expected an IOException because the magic bytes are wrong"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("bad magic bytes")); + } + } + + @Test + public void testDecompressFailsWhenOutputBufferTooSmall() { + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(10)); + ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); + decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + fail("Expected an IOException because the output buffer is too small"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("Output buffer is too small")); + } + } + + @Test + public void testDecompressFailsOnCorruptedCrc32() { + byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); + // First 4 bytes of the 8-byte trailer are the CRC32, leave ISIZE (the last 4 bytes) alone. + corrupted[corrupted.length - 8] ^= (byte) 0xff; + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(corrupted)); + decompressor.decompress(output, input, corrupted.length); + fail("Expected an IOException because the trailer's CRC32 no longer matches"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("CRC32")); + } + } + + @Test + public void testDecompressFailsOnCorruptedIsize() { + byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); + // Last 4 bytes of the 8-byte trailer are the ISIZE. + corrupted[corrupted.length - 4] ^= (byte) 0xff; + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(corrupted)); + decompressor.decompress(output, input, corrupted.length); + fail("Expected an IOException because the trailer's ISIZE no longer matches"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("ISIZE")); + } + } + + @Test + public void testDecompressSucceedsRepeatedlyOnTheSameDecompressor() throws IOException { + // Mirrors how CodecPool actually uses these: one instance is reused across many blocks, so the + // trailer (CRC32/ISIZE) verification must produce a correct result on every call, not just the + // first. + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + for (int i = 0; i < 3; i++) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); + int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + assertEquals("HBase is fun to use and very fast", + Bytes.toString(output.toBytes(0, decompressedSize))); + } + } + } + + @Test + public void testDecompressorIsStillUsableAfterAPreviousCallThrows() throws IOException { + byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); + // First 4 bytes of the 8-byte trailer are the CRC32. + corrupted[corrupted.length - 8] ^= (byte) 0xff; + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + ByteBuff badOutput = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff badInput = new SingleByteBuff(ByteBuffer.wrap(corrupted)); + try { + decompressor.decompress(badOutput, badInput, corrupted.length); + fail("Expected an IOException because the trailer's CRC32 no longer matches"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("CRC32")); + } + + // A prior failure must not leave the shared Inflater/CRC32 state corrupted for the next, + // valid call. + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); + int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + assertEquals("HBase is fun to use and very fast", + Bytes.toString(output.toBytes(0, decompressedSize))); + } + } + + /** + * This is the exact gate {@code HFileBlockDefaultDecodingContext#canDecompressViaByteBuff} relies + * on to decide between ByteBuff decompression and the stream path, driven end-to-end from the + * {@code hbase.io.compress.gz.allowByteBuffDecompression} config flag. + */ + @Test + public void testReinitControlsByteBuffDecompressionViaConfigFlag() { + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); + + Configuration conf = new Configuration(false); + conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", false); + decompressor.reinit(GzipHFileDecompressionContext.fromConfiguration(conf)); + assertFalse("Block reader must fall back to stream decompression when the config flag " + + "disables ByteBuff decompression", decompressor.canDecompress(output, input)); + + conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", true); + decompressor.reinit(GzipHFileDecompressionContext.fromConfiguration(conf)); + assertTrue("Block reader must use ByteBuff decompression when the config flag is enabled", + decompressor.canDecompress(output, input)); + + // The default, with no config value set, must also allow ByteBuff decompression. + decompressor + .reinit(GzipHFileDecompressionContext.fromConfiguration(new Configuration(false))); + assertTrue(decompressor.canDecompress(output, input)); + } + } + + @Test + public void testReinitWithNullContextIsNoOp() { + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); + + Configuration conf = new Configuration(false); + conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", false); + decompressor.reinit(GzipHFileDecompressionContext.fromConfiguration(conf)); + assertFalse(decompressor.canDecompress(output, input)); + + decompressor.reinit(null); + assertFalse("reinit(null) must not reset allowByteBuffDecompression back to the default", + decompressor.canDecompress(output, input)); + } + } + + @Test + public void testReinitFailsOnWrongContextType() { + Compression.HFileDecompressionContext wrongContext = + new Compression.HFileDecompressionContext() { + @Override + public void close() { + } + + @Override + public long heapSize() { + return 0; + } + }; + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + decompressor.reinit(wrongContext); + fail("Expected an IllegalArgumentException because the context was not a " + + "GzipHFileDecompressionContext"); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains("GzipHFileDecompressionContext")); + } + } + +} diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java new file mode 100644 index 000000000000..06651648aca0 --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java @@ -0,0 +1,72 @@ +/* + * 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.hadoop.hbase.io.compress; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.HBaseClassTestRule; +import org.apache.hadoop.hbase.HBaseTestingUtil; +import org.apache.hadoop.hbase.testclassification.IOTests; +import org.apache.hadoop.hbase.testclassification.SmallTests; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +@Category({ IOTests.class, SmallTests.class }) +public class TestHFileCompressionGzip extends HFileTestBase { + + @ClassRule + public static final HBaseClassTestRule CLASS_RULE = + HBaseClassTestRule.forClass(TestHFileCompressionGzip.class); + + private static Configuration conf; + + @BeforeClass + public static void setUpBeforeClass() throws Exception { + HFileTestBase.setUpBeforeClass(); + } + + @Before + public void setUp() throws Exception { + conf = TEST_UTIL.getConfiguration(); + HFileTestBase.setUpBeforeClass(); + } + + @Test + public void testWithStreamDecompression() throws Exception { + conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", false); + Compression.Algorithm.GZ.reload(conf); + + Path path = + new Path(TEST_UTIL.getDataTestDir(), HBaseTestingUtil.getRandomUUID().toString() + ".hfile"); + doTest(conf, path, Compression.Algorithm.GZ); + } + + @Test + public void testWithByteBuffDecompression() throws Exception { + conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", true); + Compression.Algorithm.GZ.reload(conf); + + Path path = + new Path(TEST_UTIL.getDataTestDir(), HBaseTestingUtil.getRandomUUID().toString() + ".hfile"); + doTest(conf, path, Compression.Algorithm.GZ); + } + +} From e76f52dfff8cb6cb81025add411cdc346bf519e0 Mon Sep 17 00:00:00 2001 From: sahmadsabri Date: Tue, 4 Aug 2026 12:03:59 -0500 Subject: [PATCH 02/12] Update Java compile source and release target to version 11 in pom.xml --- pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index dcd7400835bc..1937b5e564e9 100644 --- a/pom.xml +++ b/pom.xml @@ -540,8 +540,8 @@ ${project.build.finalName}.tar.gz yyyy-MM-dd'T'HH:mm ${maven.build.timestamp} - 1.8 - 8 + 11 + 11 3.5.0 From 9a5f09c4342aef4a0db0945e4b4cad0def524189 Mon Sep 17 00:00:00 2001 From: sahmadsabri Date: Tue, 4 Aug 2026 12:50:13 -0500 Subject: [PATCH 03/12] Fix import statement for HBaseTestingUtility in TestHFileCompressionGzip --- .../hadoop/hbase/io/compress/TestHFileCompressionGzip.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java index 06651648aca0..9182df48d698 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java @@ -20,7 +20,7 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.HBaseClassTestRule; -import org.apache.hadoop.hbase.HBaseTestingUtil; +import org.apache.hadoop.hbase.HBaseTestingUtility; import org.apache.hadoop.hbase.testclassification.IOTests; import org.apache.hadoop.hbase.testclassification.SmallTests; import org.junit.Before; @@ -55,7 +55,7 @@ public void testWithStreamDecompression() throws Exception { Compression.Algorithm.GZ.reload(conf); Path path = - new Path(TEST_UTIL.getDataTestDir(), HBaseTestingUtil.getRandomUUID().toString() + ".hfile"); + new Path(TEST_UTIL.getDataTestDir(), HBaseTestingUtility.getRandomUUID().toString() + ".hfile"); doTest(conf, path, Compression.Algorithm.GZ); } @@ -65,7 +65,7 @@ public void testWithByteBuffDecompression() throws Exception { Compression.Algorithm.GZ.reload(conf); Path path = - new Path(TEST_UTIL.getDataTestDir(), HBaseTestingUtil.getRandomUUID().toString() + ".hfile"); + new Path(TEST_UTIL.getDataTestDir(), HBaseTestingUtility.getRandomUUID().toString() + ".hfile"); doTest(conf, path, Compression.Algorithm.GZ); } From 163fe92d1161255e2101d136d1a4222b5ba8ba2d Mon Sep 17 00:00:00 2001 From: sahmadsabri Date: Tue, 4 Aug 2026 12:54:18 -0500 Subject: [PATCH 04/12] Refactor path initialization in test classes for improved readability --- .../hadoop/hbase/client/TestOperationInterceptor.java | 1 + .../hbase/io/compress/TestHFileCompressionGzip.java | 8 ++++---- .../master/balancer/TestUnattainableBalancerCostGoal.java | 4 ++-- 3 files changed, 7 insertions(+), 6 deletions(-) diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestOperationInterceptor.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestOperationInterceptor.java index d2a76817e647..ddafd774f941 100644 --- a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestOperationInterceptor.java +++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestOperationInterceptor.java @@ -21,6 +21,7 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNotSame; import static org.junit.jupiter.api.Assertions.fail; + import java.io.IOException; import java.util.ArrayList; import java.util.List; diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java index 9182df48d698..1f689235b5a0 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java @@ -54,8 +54,8 @@ public void testWithStreamDecompression() throws Exception { conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", false); Compression.Algorithm.GZ.reload(conf); - Path path = - new Path(TEST_UTIL.getDataTestDir(), HBaseTestingUtility.getRandomUUID().toString() + ".hfile"); + Path path = new Path(TEST_UTIL.getDataTestDir(), + HBaseTestingUtility.getRandomUUID().toString() + ".hfile"); doTest(conf, path, Compression.Algorithm.GZ); } @@ -64,8 +64,8 @@ public void testWithByteBuffDecompression() throws Exception { conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", true); Compression.Algorithm.GZ.reload(conf); - Path path = - new Path(TEST_UTIL.getDataTestDir(), HBaseTestingUtility.getRandomUUID().toString() + ".hfile"); + Path path = new Path(TEST_UTIL.getDataTestDir(), + HBaseTestingUtility.getRandomUUID().toString() + ".hfile"); doTest(conf, path, Compression.Algorithm.GZ); } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/balancer/TestUnattainableBalancerCostGoal.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/balancer/TestUnattainableBalancerCostGoal.java index 5e95564b6fee..cf3f241cab89 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/balancer/TestUnattainableBalancerCostGoal.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/balancer/TestUnattainableBalancerCostGoal.java @@ -24,7 +24,6 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Set; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseClassTestRule; import org.apache.hadoop.hbase.ServerName; @@ -33,7 +32,6 @@ import org.apache.hadoop.hbase.client.RegionInfoBuilder; import org.apache.hadoop.hbase.testclassification.MasterTests; import org.apache.hadoop.hbase.testclassification.MediumTests; -import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableSet; import org.junit.BeforeClass; import org.junit.ClassRule; import org.junit.Test; @@ -41,6 +39,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableSet; + /** * If your minCostNeedsBalance is set too low, then the balancer should still eventually stop making * moves as further cost improvements become impossible, and balancer plan calculation becomes From 9c42417461260b6cb247e7a828da8cfeb06fd1ac Mon Sep 17 00:00:00 2001 From: sahmadsabri Date: Tue, 4 Aug 2026 23:44:54 -0500 Subject: [PATCH 05/12] Refactor GzipByteBuffDecompressor to use Hadoop's native ZlibDecompressor and improve error handling --- .../io/compress/GzipByteBuffDecompressor.java | 127 +++++-------- .../io/compress/ReusableStreamGzipCodec.java | 2 +- .../TestGzipByteBuffDecompressor.java | 176 +++++++++--------- pom.xml | 4 +- 4 files changed, 143 insertions(+), 166 deletions(-) diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java index 17970eff3e51..6c216827f9a9 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java @@ -20,44 +20,47 @@ import edu.umd.cs.findbugs.annotations.Nullable; import java.io.IOException; import java.nio.ByteBuffer; -import java.nio.ByteOrder; -import java.util.zip.CRC32; -import java.util.zip.DataFormatException; -import java.util.zip.Inflater; import org.apache.hadoop.hbase.nio.ByteBuff; import org.apache.hadoop.hbase.nio.SingleByteBuff; -import org.apache.yetus.audience.InterfaceAudience; +import org.apache.hadoop.io.compress.zlib.ZlibDecompressor; /** - * Glue for ByteBuffDecompressor on top of {@link Inflater}. Only supports gzip members with the - * fixed ten-byte header that {@link ReusableStreamGzipCodec} (and Hadoop's native zlib gzip - * compressor) always writes, i.e. no FEXTRA/FNAME/FCOMMENT/FHCRC, since that is the only format - * HBase ever produces on the compression side. + * Glue for ByteBuffDecompressor on top of Hadoop's native + * {@link ZlibDecompressor.ZlibDirectDecompressor}. */ -@InterfaceAudience.Private +@InterfaceAudienc.Private public class GzipByteBuffDecompressor implements ByteBuffDecompressor { private static final int GZIP_HEADER_LENGTH = 10; private static final int GZIP_TRAILER_LENGTH = 8; - private static final byte GZIP_MAGIC_0 = (byte) 0x1f; - private static final byte GZIP_MAGIC_1 = (byte) 0x8b; - private final Inflater inflater = new Inflater(true); + @Nullable + private final ZlibDecompressor.ZlibDirectDecompressor decompressor; // Intended to be set to false by some unit tests private boolean allowByteBuffDecompression; - GzipByteBuffDecompressor() { + GzipByteBuffDecompressor(boolean nativeZlibLoaded) { + decompressor = nativeZlibLoaded + ? new ZlibDecompressor.ZlibDirectDecompressor(ZlibDecompressor.CompressionHeader.GZIP_FORMAT, + 0) + : null; allowByteBuffDecompression = true; } @Override public boolean canDecompress(ByteBuff output, ByteBuff input) { - return allowByteBuffDecompression && output instanceof SingleByteBuff - && input instanceof SingleByteBuff; + return decompressor != null && allowByteBuffDecompression && output instanceof SingleByteBuff + && input instanceof SingleByteBuff && output.nioByteBuffers()[0].isDirect() + && input.nioByteBuffers()[0].isDirect(); } @Override public int decompress(ByteBuff output, ByteBuff input, int inputLen) throws IOException { + if (decompressor == null) { + throw new IllegalStateException( + "GzipByteBuffDecompressor#decompress() was called but Hadoop's native zlib library is " + + "not loaded, this should never happen since canDecompress() would have returned false"); + } if (!(output instanceof SingleByteBuff) || !(input instanceof SingleByteBuff)) { throw new IllegalStateException( "At least one buffer is not a SingleByteBuff, this is not supported"); @@ -67,74 +70,44 @@ public int decompress(ByteBuff output, ByteBuff input, int inputLen) throws IOEx } ByteBuffer nioInput = input.nioByteBuffers()[0]; - int inputStart = nioInput.position(); - if (nioInput.get(inputStart) != GZIP_MAGIC_0 || nioInput.get(inputStart + 1) != GZIP_MAGIC_1) { - throw new IOException("Not a gzip member, bad magic bytes"); - } - ByteBuffer nioOutput = output.nioByteBuffers()[0]; + if (!nioInput.isDirect() || !nioOutput.isDirect()) { + throw new IllegalStateException( + "At least one buffer is not direct, this is not supported by the native zlib decompressor"); + } - // Isolate the raw DEFLATE payload (strip the fixed header and the CRC32/ISIZE trailer) into - // its own view so Inflater can consume it without disturbing nioInput's own position/limit. - ByteBuffer deflateStream = nioInput.duplicate(); - deflateStream.limit(inputStart + inputLen - GZIP_TRAILER_LENGTH); - deflateStream.position(inputStart + GZIP_HEADER_LENGTH); - - inflater.reset(); - inflater.setInput(deflateStream); + int inputStart = nioInput.position(); int outputStart = nioOutput.position(); - try { - while (!inflater.finished()) { - if (inflater.inflate(nioOutput) == 0) { - if (inflater.finished()) { - break; - } - if (inflater.needsInput()) { - throw new IOException("Unexpected end of gzip stream"); - } - if (!nioOutput.hasRemaining()) { - throw new IOException("Output buffer is too small for the decompressed gzip stream"); - } + + // Duplicate so the decompressor can advance its own position without disturbing nioInput. + // The native decompressor consumes the whole gzip member, including the header and the + // CRC32/ISIZE trailer, and validates the trailer itself. + ByteBuffer gzipMember = nioInput.duplicate(); + gzipMember.limit(inputStart + inputLen); + + decompressor.reset(); + while (!decompressor.finished()) { + int outputRemainingBefore = nioOutput.remaining(); + try { + decompressor.decompress(gzipMember, nioOutput); + } catch (IOException e) { + throw new IOException("Invalid gzip stream: " + e.getMessage(), e); + } + // No progress means either the output buffer is full or the gzip member is truncated. + if (nioOutput.remaining() == outputRemainingBefore && !decompressor.finished()) { + if (!nioOutput.hasRemaining()) { + throw new IOException("Output buffer is too small for the decompressed gzip stream"); } + throw new IOException("Unexpected end of gzip stream"); } - } catch (DataFormatException e) { - throw new IOException("Invalid gzip stream", e); } - int decompressedLength = nioOutput.position() - outputStart; - verifyTrailer(nioInput, inputStart, inputLen, nioOutput, outputStart, decompressedLength); - - nioInput.position(inputStart + inputLen); - return decompressedLength; - } - - /** - * {@link Inflater} runs in nowrap mode and never looks at the gzip header or trailer, so this is - * the only place the CRC32 and ISIZE fields of the trailer are ever checked. Catches the case - * where the raw DEFLATE payload decoded "successfully" (no {@link DataFormatException}) but - * produced the wrong bytes or the wrong number of bytes. - */ - private void verifyTrailer(ByteBuffer nioInput, int inputStart, int inputLen, - ByteBuffer nioOutput, int outputStart, int decompressedLength) throws IOException { - ByteBuffer trailer = nioInput.duplicate().order(ByteOrder.LITTLE_ENDIAN); - trailer.position(inputStart + inputLen - GZIP_TRAILER_LENGTH); - int expectedCrc32 = trailer.getInt(); - int expectedISize = trailer.getInt(); - - if (decompressedLength != expectedISize) { - throw new IOException("Decompressed length " + decompressedLength - + " does not match gzip trailer ISIZE " + expectedISize); + if (gzipMember.hasRemaining()) { + throw new IOException("Unexpected trailing bytes after decompressing gzip stream"); } - CRC32 crc32 = new CRC32(); - ByteBuffer writtenOutput = nioOutput.duplicate(); - writtenOutput.limit(nioOutput.position()); - writtenOutput.position(outputStart); - crc32.update(writtenOutput); - if ((int) crc32.getValue() != expectedCrc32) { - throw new IOException( - "Decompressed data's CRC32 does not match gzip trailer CRC32, " + "data is corrupt"); - } + nioInput.position(inputStart + inputLen); + return nioOutput.position() - outputStart; } @Override @@ -154,7 +127,9 @@ public void reinit(@Nullable Compression.HFileDecompressionContext newHFileDecom @Override public void close() { - inflater.end(); + if (decompressor != null) { + decompressor.end(); + } } } diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/ReusableStreamGzipCodec.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/ReusableStreamGzipCodec.java index 46982a46a0d7..08276d7f0732 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/ReusableStreamGzipCodec.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/ReusableStreamGzipCodec.java @@ -188,7 +188,7 @@ public CompressionOutputStream createOutputStream(OutputStream out) throws IOExc @Override public ByteBuffDecompressor createByteBuffDecompressor() { - return new GzipByteBuffDecompressor(); + return new GzipByteBuffDecompressor(ZlibFactory.isNativeZlibLoaded(getConf())); } @Override diff --git a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java index 8fe93120a302..2df2ecd63b3c 100644 --- a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java +++ b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java @@ -21,6 +21,7 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.junit.Assume.assumeTrue; import java.io.IOException; import java.nio.ByteBuffer; @@ -32,6 +33,7 @@ import org.apache.hadoop.hbase.nio.SingleByteBuff; import org.apache.hadoop.hbase.testclassification.SmallTests; import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.util.NativeCodeLoader; import org.junit.ClassRule; import org.junit.Test; import org.junit.experimental.categories.Category; @@ -45,63 +47,66 @@ public class TestGzipByteBuffDecompressor { /* * "HBase is fun to use and very fast" compressed as a single gzip member via GZIPOutputStream, - * i.e. exactly the framing GzipByteBuffDecompressor expects: a fixed 10-byte header (no - * FEXTRA/FNAME/FCOMMENT/FHCRC), a raw DEFLATE stream, and an 8-byte CRC32/ISIZE trailer. + * matching the framing that ReusableStreamGzipCodec produces on the compression side. */ private static final byte[] COMPRESSED_PAYLOAD = Bytes.fromHex( "1f8b08000000000000fff3704a2c4e55c82c56482bcd5328c9572805f212f35214ca528b2a15d2128b4b006edf170321000000"); + /** + * GzipByteBuffDecompressor is backed by Hadoop's native zlib binding, so actually decompressing + * anything requires that native library to be loaded on this JVM. + */ + private static void assumeNativeZlibLoaded() { + assumeTrue("Hadoop's native code is not loaded on this JVM, skipping", + NativeCodeLoader.isNativeCodeLoaded()); + } + + @Test + public void testCapabilitiesWithoutNativeZlibLoaded() { + // Deliberately constructed as if native zlib is unavailable, regardless of this JVM's actual + // environment, so this test is deterministic everywhere. + ByteBuff emptySingleDirectBuff = new SingleByteBuff(ByteBuffer.allocateDirect(0)); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(false)) { + assertFalse("Without native zlib there is no way to decompress via ByteBuffs", + decompressor.canDecompress(emptySingleDirectBuff, emptySingleDirectBuff)); + } + } + @Test - public void testCapabilities() { + public void testCapabilitiesWithNativeZlibLoaded() { + assumeNativeZlibLoaded(); ByteBuff emptySingleHeapBuff = new SingleByteBuff(ByteBuffer.allocate(0)); ByteBuff emptyMultiHeapBuff = new MultiByteBuff(ByteBuffer.allocate(0), ByteBuffer.allocate(0)); ByteBuff emptySingleDirectBuff = new SingleByteBuff(ByteBuffer.allocateDirect(0)); ByteBuff emptyMultiDirectBuff = new MultiByteBuff(ByteBuffer.allocateDirect(0), ByteBuffer.allocateDirect(0)); - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { - assertTrue(decompressor.canDecompress(emptySingleHeapBuff, emptySingleHeapBuff)); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { assertTrue(decompressor.canDecompress(emptySingleDirectBuff, emptySingleDirectBuff)); - assertTrue(decompressor.canDecompress(emptySingleHeapBuff, emptySingleDirectBuff)); - assertTrue(decompressor.canDecompress(emptySingleDirectBuff, emptySingleHeapBuff)); + // The native zlib binding reads/writes buffer memory directly, so only direct buffers are + // supported; heap buffers must fall back to stream-based decompression instead. + assertFalse(decompressor.canDecompress(emptySingleHeapBuff, emptySingleHeapBuff)); + assertFalse(decompressor.canDecompress(emptySingleHeapBuff, emptySingleDirectBuff)); + assertFalse(decompressor.canDecompress(emptySingleDirectBuff, emptySingleHeapBuff)); assertFalse(decompressor.canDecompress(emptyMultiHeapBuff, emptyMultiHeapBuff)); assertFalse(decompressor.canDecompress(emptyMultiDirectBuff, emptyMultiDirectBuff)); - assertFalse(decompressor.canDecompress(emptySingleHeapBuff, emptyMultiHeapBuff)); assertFalse(decompressor.canDecompress(emptySingleDirectBuff, emptyMultiDirectBuff)); } } - @Test - public void testDecompressHeapToHeap() throws IOException { - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); - int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); - assertEquals("HBase is fun to use and very fast", - Bytes.toString(output.toBytes(0, decompressedSize))); - } + private static ByteBuff directBuffWith(byte[] data) { + ByteBuffer buffer = ByteBuffer.allocateDirect(data.length); + buffer.put(data); + buffer.rewind(); + return new SingleByteBuff(buffer); } @Test public void testDecompressDirectToDirect() throws IOException { - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); - ByteBuff input = new SingleByteBuff(ByteBuffer.allocateDirect(COMPRESSED_PAYLOAD.length)); - input.put(COMPRESSED_PAYLOAD); - input.rewind(); - int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); - assertEquals("HBase is fun to use and very fast", - Bytes.toString(output.toBytes(0, decompressedSize))); - } - } - - @Test - public void testDecompressDirectToHeap() throws IOException { - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = new SingleByteBuff(ByteBuffer.allocateDirect(COMPRESSED_PAYLOAD.length)); - input.put(COMPRESSED_PAYLOAD); - input.rewind(); + ByteBuff input = directBuffWith(COMPRESSED_PAYLOAD); int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); assertEquals("HBase is fun to use and very fast", Bytes.toString(output.toBytes(0, decompressedSize))); @@ -109,21 +114,11 @@ public void testDecompressDirectToHeap() throws IOException { } @Test - public void testDecompressHeapToDirect() throws IOException { - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + public void testDecompressFailsOnTooShortInput() throws IOException { + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); - ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); - int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); - assertEquals("HBase is fun to use and very fast", - Bytes.toString(output.toBytes(0, decompressedSize))); - } - } - - @Test - public void testDecompressFailsOnTooShortInput() { - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = new SingleByteBuff(ByteBuffer.allocate(10)); + ByteBuff input = new SingleByteBuff(ByteBuffer.allocateDirect(10)); decompressor.decompress(output, input, 10); fail("Expected an IOException because the input is too short to be a gzip member"); } catch (IOException e) { @@ -132,24 +127,26 @@ public void testDecompressFailsOnTooShortInput() { } @Test - public void testDecompressFailsOnBadMagicBytes() { + public void testDecompressFailsOnBadMagicBytes() throws IOException { + assumeNativeZlibLoaded(); byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); corrupted[0] ^= (byte) 0xff; - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(corrupted)); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); + ByteBuff input = directBuffWith(corrupted); decompressor.decompress(output, input, corrupted.length); fail("Expected an IOException because the magic bytes are wrong"); } catch (IOException e) { - assertTrue(e.getMessage().contains("bad magic bytes")); + assertTrue(e.getMessage().contains("Invalid gzip stream")); } } @Test - public void testDecompressFailsWhenOutputBufferTooSmall() { - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(10)); - ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); + public void testDecompressFailsWhenOutputBufferTooSmall() throws IOException { + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(10)); + ByteBuff input = directBuffWith(COMPRESSED_PAYLOAD); decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); fail("Expected an IOException because the output buffer is too small"); } catch (IOException e) { @@ -158,44 +155,46 @@ public void testDecompressFailsWhenOutputBufferTooSmall() { } @Test - public void testDecompressFailsOnCorruptedCrc32() { + public void testDecompressFailsOnCorruptedCrc32() throws IOException { + assumeNativeZlibLoaded(); byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); // First 4 bytes of the 8-byte trailer are the CRC32, leave ISIZE (the last 4 bytes) alone. corrupted[corrupted.length - 8] ^= (byte) 0xff; - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(corrupted)); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); + ByteBuff input = directBuffWith(corrupted); decompressor.decompress(output, input, corrupted.length); fail("Expected an IOException because the trailer's CRC32 no longer matches"); } catch (IOException e) { - assertTrue(e.getMessage().contains("CRC32")); + assertTrue(e.getMessage().contains("Invalid gzip stream")); } } @Test - public void testDecompressFailsOnCorruptedIsize() { + public void testDecompressFailsOnCorruptedIsize() throws IOException { + assumeNativeZlibLoaded(); byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); // Last 4 bytes of the 8-byte trailer are the ISIZE. corrupted[corrupted.length - 4] ^= (byte) 0xff; - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(corrupted)); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); + ByteBuff input = directBuffWith(corrupted); decompressor.decompress(output, input, corrupted.length); fail("Expected an IOException because the trailer's ISIZE no longer matches"); } catch (IOException e) { - assertTrue(e.getMessage().contains("ISIZE")); + assertTrue(e.getMessage().contains("Invalid gzip stream")); } } @Test public void testDecompressSucceedsRepeatedlyOnTheSameDecompressor() throws IOException { + assumeNativeZlibLoaded(); // Mirrors how CodecPool actually uses these: one instance is reused across many blocks, so the - // trailer (CRC32/ISIZE) verification must produce a correct result on every call, not just the - // first. - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + // native decompressor must produce a correct result on every call, not just the first. + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { for (int i = 0; i < 3; i++) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); + ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); + ByteBuff input = directBuffWith(COMPRESSED_PAYLOAD); int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); assertEquals("HBase is fun to use and very fast", Bytes.toString(output.toBytes(0, decompressedSize))); @@ -205,23 +204,24 @@ public void testDecompressSucceedsRepeatedlyOnTheSameDecompressor() throws IOExc @Test public void testDecompressorIsStillUsableAfterAPreviousCallThrows() throws IOException { + assumeNativeZlibLoaded(); byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); // First 4 bytes of the 8-byte trailer are the CRC32. corrupted[corrupted.length - 8] ^= (byte) 0xff; - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { - ByteBuff badOutput = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff badInput = new SingleByteBuff(ByteBuffer.wrap(corrupted)); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff badOutput = new SingleByteBuff(ByteBuffer.allocateDirect(64)); + ByteBuff badInput = directBuffWith(corrupted); try { decompressor.decompress(badOutput, badInput, corrupted.length); fail("Expected an IOException because the trailer's CRC32 no longer matches"); } catch (IOException e) { - assertTrue(e.getMessage().contains("CRC32")); + assertTrue(e.getMessage().contains("Invalid gzip stream")); } - // A prior failure must not leave the shared Inflater/CRC32 state corrupted for the next, - // valid call. - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); + // A prior failure must not leave the shared native decompressor state corrupted for the + // next, valid call. + ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); + ByteBuff input = directBuffWith(COMPRESSED_PAYLOAD); int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); assertEquals("HBase is fun to use and very fast", Bytes.toString(output.toBytes(0, decompressedSize))); @@ -235,9 +235,10 @@ public void testDecompressorIsStillUsableAfterAPreviousCallThrows() throws IOExc */ @Test public void testReinitControlsByteBuffDecompressionViaConfigFlag() { - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); + ByteBuff input = directBuffWith(COMPRESSED_PAYLOAD); Configuration conf = new Configuration(false); conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", false); @@ -259,9 +260,10 @@ public void testReinitControlsByteBuffDecompressionViaConfigFlag() { @Test public void testReinitWithNullContextIsNoOp() { - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = new SingleByteBuff(ByteBuffer.wrap(COMPRESSED_PAYLOAD)); + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); + ByteBuff input = directBuffWith(COMPRESSED_PAYLOAD); Configuration conf = new Configuration(false); conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", false); @@ -287,7 +289,7 @@ public long heapSize() { return 0; } }; - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor()) { + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(false)) { decompressor.reinit(wrongContext); fail("Expected an IllegalArgumentException because the context was not a " + "GzipHFileDecompressionContext"); diff --git a/pom.xml b/pom.xml index 1937b5e564e9..dcd7400835bc 100644 --- a/pom.xml +++ b/pom.xml @@ -540,8 +540,8 @@ ${project.build.finalName}.tar.gz yyyy-MM-dd'T'HH:mm ${maven.build.timestamp} - 11 - 11 + 1.8 + 8 3.5.0 From 922c683b68507032e5c941a057fd29abfdae6995 Mon Sep 17 00:00:00 2001 From: sahmadsabri Date: Tue, 4 Aug 2026 23:53:00 -0500 Subject: [PATCH 06/12] Fix typo in InterfaceAudience annotation in GzipByteBuffDecompressor --- .../hadoop/hbase/io/compress/GzipByteBuffDecompressor.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java index 6c216827f9a9..dcbcd954a270 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java @@ -23,12 +23,13 @@ import org.apache.hadoop.hbase.nio.ByteBuff; import org.apache.hadoop.hbase.nio.SingleByteBuff; import org.apache.hadoop.io.compress.zlib.ZlibDecompressor; +import org.apache.yetus.audience.InterfaceAudience; /** * Glue for ByteBuffDecompressor on top of Hadoop's native * {@link ZlibDecompressor.ZlibDirectDecompressor}. */ -@InterfaceAudienc.Private +@InterfaceAudience.Private public class GzipByteBuffDecompressor implements ByteBuffDecompressor { private static final int GZIP_HEADER_LENGTH = 10; From 26c931e78fd399a7070ce01db5b11337bef927c4 Mon Sep 17 00:00:00 2001 From: sahmadsabri Date: Fri, 7 Aug 2026 09:12:07 -0500 Subject: [PATCH 07/12] Enhance GzipByteBuffDecompressor to support heap ByteBuffers and improve error handling --- .../io/compress/GzipByteBuffDecompressor.java | 116 +++++++++++++++--- 1 file changed, 99 insertions(+), 17 deletions(-) diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java index dcbcd954a270..62cff4bb039b 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java @@ -20,6 +20,9 @@ import edu.umd.cs.findbugs.annotations.Nullable; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.zip.CRC32; +import java.util.zip.DataFormatException; +import java.util.zip.Inflater; import org.apache.hadoop.hbase.nio.ByteBuff; import org.apache.hadoop.hbase.nio.SingleByteBuff; import org.apache.hadoop.io.compress.zlib.ZlibDecompressor; @@ -37,31 +40,37 @@ public class GzipByteBuffDecompressor implements ByteBuffDecompressor { @Nullable private final ZlibDecompressor.ZlibDirectDecompressor decompressor; - // Intended to be set to false by some unit tests + + private final Inflater inflater = new Inflater(true); + private boolean allowByteBuffDecompression; GzipByteBuffDecompressor(boolean nativeZlibLoaded) { decompressor = nativeZlibLoaded ? new ZlibDecompressor.ZlibDirectDecompressor(ZlibDecompressor.CompressionHeader.GZIP_FORMAT, - 0) + 0) : null; allowByteBuffDecompression = true; } @Override public boolean canDecompress(ByteBuff output, ByteBuff input) { - return decompressor != null && allowByteBuffDecompression && output instanceof SingleByteBuff - && input instanceof SingleByteBuff && output.nioByteBuffers()[0].isDirect() - && input.nioByteBuffers()[0].isDirect(); + if (!allowByteBuffDecompression) { + return false; + } + if (!(output instanceof SingleByteBuff) || !(input instanceof SingleByteBuff)) { + return false; + } + boolean inputDirect = input.nioByteBuffers()[0].isDirect(); + boolean outputDirect = output.nioByteBuffers()[0].isDirect(); + if (inputDirect && outputDirect) { + return decompressor != null; + } + return true; } @Override public int decompress(ByteBuff output, ByteBuff input, int inputLen) throws IOException { - if (decompressor == null) { - throw new IllegalStateException( - "GzipByteBuffDecompressor#decompress() was called but Hadoop's native zlib library is " - + "not loaded, this should never happen since canDecompress() would have returned false"); - } if (!(output instanceof SingleByteBuff) || !(input instanceof SingleByteBuff)) { throw new IllegalStateException( "At least one buffer is not a SingleByteBuff, this is not supported"); @@ -72,17 +81,26 @@ public int decompress(ByteBuff output, ByteBuff input, int inputLen) throws IOEx ByteBuffer nioInput = input.nioByteBuffers()[0]; ByteBuffer nioOutput = output.nioByteBuffers()[0]; - if (!nioInput.isDirect() || !nioOutput.isDirect()) { - throw new IllegalStateException( - "At least one buffer is not direct, this is not supported by the native zlib decompressor"); + boolean inputDirect = nioInput.isDirect(); + boolean outputDirect = nioOutput.isDirect(); + + if (inputDirect && outputDirect) { + if (decompressor == null) { + throw new IllegalStateException( + "GzipByteBuffDecompressor#decompress() was called with direct buffers but Hadoop's " + + "native zlib library is not loaded, this should never happen since " + + "canDecompress() would have returned false"); + } + return decompressOffHeap(nioInput, nioOutput, inputLen); } + return decompressOnHeap(nioInput, nioOutput, inputLen); + } + private int decompressOffHeap(ByteBuffer nioInput, ByteBuffer nioOutput, int inputLen) + throws IOException { int inputStart = nioInput.position(); int outputStart = nioOutput.position(); - // Duplicate so the decompressor can advance its own position without disturbing nioInput. - // The native decompressor consumes the whole gzip member, including the header and the - // CRC32/ISIZE trailer, and validates the trailer itself. ByteBuffer gzipMember = nioInput.duplicate(); gzipMember.limit(inputStart + inputLen); @@ -94,7 +112,6 @@ public int decompress(ByteBuff output, ByteBuff input, int inputLen) throws IOEx } catch (IOException e) { throw new IOException("Invalid gzip stream: " + e.getMessage(), e); } - // No progress means either the output buffer is full or the gzip member is truncated. if (nioOutput.remaining() == outputRemainingBefore && !decompressor.finished()) { if (!nioOutput.hasRemaining()) { throw new IOException("Output buffer is too small for the decompressed gzip stream"); @@ -111,6 +128,70 @@ public int decompress(ByteBuff output, ByteBuff input, int inputLen) throws IOEx return nioOutput.position() - outputStart; } + // ZlibDirectDecompressor requires direct buffers — heap ByteBuffers have no stable native + // address, so we fall back to Java's Inflater for the heap case. + private int decompressOnHeap(ByteBuffer nioInput, ByteBuffer nioOutput, int inputLen) + throws IOException { + if (!nioInput.hasArray() || !nioOutput.hasArray()) { + throw new IllegalStateException( + "decompressOnHeap() requires heap ByteBuffers with backing arrays"); + } + int inputStart = nioInput.position(); + int outputStart = nioOutput.position(); + + inflater.reset(); + inflater.setInput(nioInput.array(), nioInput.arrayOffset() + inputStart + GZIP_HEADER_LENGTH, + inputLen - GZIP_HEADER_LENGTH - GZIP_TRAILER_LENGTH); + int totalDecompressed = 0; + while (!inflater.finished()) { + int remaining = nioOutput.remaining() - totalDecompressed; + if (remaining == 0) { + throw new IOException("Output buffer is too small for the decompressed gzip stream"); + } + int n; + try { + n = inflater.inflate(nioOutput.array(), + nioOutput.arrayOffset() + outputStart + totalDecompressed, remaining); + } catch (DataFormatException e) { + throw new IOException("Invalid gzip stream: " + e.getMessage(), e); + } + if (n == 0 && !inflater.finished()) { + if (inflater.needsInput()) { + throw new IOException("Unexpected end of gzip stream"); + } + throw new IOException("Unexpected state in gzip stream"); + } + totalDecompressed += n; + } + verifyGzipTrailer(nioInput.array(), nioInput.arrayOffset() + inputStart, inputLen, + nioOutput.array(), nioOutput.arrayOffset() + outputStart, totalDecompressed); + nioOutput.position(outputStart + totalDecompressed); + nioInput.position(inputStart + inputLen); + return totalDecompressed; + } + + // Inflater runs in nowrap (raw DEFLATE) mode and is unaware of the gzip envelope, so it never + // checks the trailer. ZlibDirectDecompressor handles this automatically via GZIP_FORMAT, but + // for heap buffers we must verify the CRC32 and ISIZE fields ourselves. + private static void verifyGzipTrailer(byte[] inputData, int inputDataOffset, int inputLen, + byte[] outputData, int outputDataOffset, int decompressedLen) throws IOException { + long expectedCrc = readLittleEndianUInt32(inputData, inputDataOffset + inputLen - 8); + long expectedSize = readLittleEndianUInt32(inputData, inputDataOffset + inputLen - 4); + CRC32 crc32 = new CRC32(); + crc32.update(outputData, outputDataOffset, decompressedLen); + if (crc32.getValue() != expectedCrc) { + throw new IOException("Gzip CRC32 mismatch"); + } + if ((decompressedLen & 0xFFFFFFFFL) != expectedSize) { + throw new IOException("Gzip size mismatch"); + } + } + + private static long readLittleEndianUInt32(byte[] data, int offset) { + return (data[offset] & 0xFFL) | ((data[offset + 1] & 0xFFL) << 8) + | ((data[offset + 2] & 0xFFL) << 16) | ((data[offset + 3] & 0xFFL) << 24); + } + @Override public void reinit(@Nullable Compression.HFileDecompressionContext newHFileDecompressionContext) { if (newHFileDecompressionContext == null) { @@ -128,6 +209,7 @@ public void reinit(@Nullable Compression.HFileDecompressionContext newHFileDecom @Override public void close() { + inflater.end(); if (decompressor != null) { decompressor.end(); } From 0c6de7ddf45409c9755b648929e5f14559ba457a Mon Sep 17 00:00:00 2001 From: sahmadsabri Date: Fri, 7 Aug 2026 09:57:21 -0500 Subject: [PATCH 08/12] Update GzipByteBuffDecompressor to refine direct buffer handling logic --- .../hadoop/hbase/io/compress/GzipByteBuffDecompressor.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java index 62cff4bb039b..d80f060e2c12 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java @@ -66,7 +66,7 @@ public boolean canDecompress(ByteBuff output, ByteBuff input) { if (inputDirect && outputDirect) { return decompressor != null; } - return true; + return !inputDirect && !outputDirect; } @Override From 4dfb91eb5aef65797affb5d66c1a2344ec70e157 Mon Sep 17 00:00:00 2001 From: sahmadsabri Date: Mon, 10 Aug 2026 00:31:25 -0400 Subject: [PATCH 09/12] Copy updated GzipByteBuffDecompressor files from HBASE-30321-gzipbytebuff Brings in the latest versions of GzipByteBuffDecompressor, GzipHFileDecompressionContext, TestGzipByteBuffDecompressor, and TestHFileCompressionGzip from the HBASE-30321-gzipbytebuff branch. Co-Authored-By: Claude Sonnet 4.6 --- .../io/compress/GzipByteBuffDecompressor.java | 110 +++-- .../GzipHFileDecompressionContext.java | 6 +- .../TestGzipByteBuffDecompressor.java | 381 +++++++++++++++--- .../io/compress/TestHFileCompressionGzip.java | 31 +- 4 files changed, 392 insertions(+), 136 deletions(-) diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java index d80f060e2c12..26bb49c5b0df 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java @@ -20,9 +20,6 @@ import edu.umd.cs.findbugs.annotations.Nullable; import java.io.IOException; import java.nio.ByteBuffer; -import java.util.zip.CRC32; -import java.util.zip.DataFormatException; -import java.util.zip.Inflater; import org.apache.hadoop.hbase.nio.ByteBuff; import org.apache.hadoop.hbase.nio.SingleByteBuff; import org.apache.hadoop.io.compress.zlib.ZlibDecompressor; @@ -38,17 +35,25 @@ public class GzipByteBuffDecompressor implements ByteBuffDecompressor { private static final int GZIP_HEADER_LENGTH = 10; private static final int GZIP_TRAILER_LENGTH = 8; + // Size of the on-heap ZlibDecompressor's internal staging buffers + private static final int HEAP_DECOMPRESSOR_BUFFER_SIZE = 64 * 1024; + @Nullable private final ZlibDecompressor.ZlibDirectDecompressor decompressor; - private final Inflater inflater = new Inflater(true); + @Nullable + private final ZlibDecompressor heapDecompressor; private boolean allowByteBuffDecompression; GzipByteBuffDecompressor(boolean nativeZlibLoaded) { decompressor = nativeZlibLoaded ? new ZlibDecompressor.ZlibDirectDecompressor(ZlibDecompressor.CompressionHeader.GZIP_FORMAT, - 0) + 0) + : null; + heapDecompressor = nativeZlibLoaded + ? new ZlibDecompressor(ZlibDecompressor.CompressionHeader.GZIP_FORMAT, + HEAP_DECOMPRESSOR_BUFFER_SIZE) : null; allowByteBuffDecompression = true; } @@ -66,7 +71,10 @@ public boolean canDecompress(ByteBuff output, ByteBuff input) { if (inputDirect && outputDirect) { return decompressor != null; } - return !inputDirect && !outputDirect; + if (inputDirect != outputDirect) { + return false; + } + return !inputDirect && !outputDirect && heapDecompressor != null; } @Override @@ -93,6 +101,12 @@ public int decompress(ByteBuff output, ByteBuff input, int inputLen) throws IOEx } return decompressOffHeap(nioInput, nioOutput, inputLen); } + if (heapDecompressor == null) { + throw new IllegalStateException( + "GzipByteBuffDecompressor#decompress() was called with heap buffers but Hadoop's " + + "native zlib library is not loaded, this should never happen since " + + "canDecompress() would have returned false"); + } return decompressOnHeap(nioInput, nioOutput, inputLen); } @@ -105,31 +119,26 @@ private int decompressOffHeap(ByteBuffer nioInput, ByteBuffer nioOutput, int inp gzipMember.limit(inputStart + inputLen); decompressor.reset(); - while (!decompressor.finished()) { - int outputRemainingBefore = nioOutput.remaining(); - try { - decompressor.decompress(gzipMember, nioOutput); - } catch (IOException e) { - throw new IOException("Invalid gzip stream: " + e.getMessage(), e); - } - if (nioOutput.remaining() == outputRemainingBefore && !decompressor.finished()) { - if (!nioOutput.hasRemaining()) { - throw new IOException("Output buffer is too small for the decompressed gzip stream"); - } - throw new IOException("Unexpected end of gzip stream"); + try { + decompressor.decompress(gzipMember, nioOutput); + } catch (IOException e) { + throw new IOException("Invalid gzip stream: " + e.getMessage(), e); + } + if (!decompressor.finished()) { + if (!nioOutput.hasRemaining()) { + throw new IOException("Output buffer is too small for the decompressed gzip stream"); } + throw new IOException("Unexpected end of gzip stream"); } - if (gzipMember.hasRemaining()) { throw new IOException("Unexpected trailing bytes after decompressing gzip stream"); } nioInput.position(inputStart + inputLen); + return nioOutput.position() - outputStart; } - // ZlibDirectDecompressor requires direct buffers — heap ByteBuffers have no stable native - // address, so we fall back to Java's Inflater for the heap case. private int decompressOnHeap(ByteBuffer nioInput, ByteBuffer nioOutput, int inputLen) throws IOException { if (!nioInput.hasArray() || !nioOutput.hasArray()) { @@ -138,78 +147,51 @@ private int decompressOnHeap(ByteBuffer nioInput, ByteBuffer nioOutput, int inpu } int inputStart = nioInput.position(); int outputStart = nioOutput.position(); + int outputCapacity = nioOutput.remaining(); - inflater.reset(); - inflater.setInput(nioInput.array(), nioInput.arrayOffset() + inputStart + GZIP_HEADER_LENGTH, - inputLen - GZIP_HEADER_LENGTH - GZIP_TRAILER_LENGTH); + heapDecompressor.reset(); + heapDecompressor.setInput(nioInput.array(), nioInput.arrayOffset() + inputStart, inputLen); int totalDecompressed = 0; - while (!inflater.finished()) { - int remaining = nioOutput.remaining() - totalDecompressed; + while (!heapDecompressor.finished()) { + int remaining = outputCapacity - totalDecompressed; if (remaining == 0) { throw new IOException("Output buffer is too small for the decompressed gzip stream"); } - int n; - try { - n = inflater.inflate(nioOutput.array(), - nioOutput.arrayOffset() + outputStart + totalDecompressed, remaining); - } catch (DataFormatException e) { - throw new IOException("Invalid gzip stream: " + e.getMessage(), e); - } - if (n == 0 && !inflater.finished()) { - if (inflater.needsInput()) { + int n = heapDecompressor.decompress(nioOutput.array(), + nioOutput.arrayOffset() + outputStart + totalDecompressed, remaining); + if (n == 0 && !heapDecompressor.finished()) { + if (heapDecompressor.needsInput()) { throw new IOException("Unexpected end of gzip stream"); } - throw new IOException("Unexpected state in gzip stream"); + throw new IOException( + "Gzip decompressor made no progress and is not finished; aborting to avoid an " + + "infinite loop"); } totalDecompressed += n; } - verifyGzipTrailer(nioInput.array(), nioInput.arrayOffset() + inputStart, inputLen, - nioOutput.array(), nioOutput.arrayOffset() + outputStart, totalDecompressed); nioOutput.position(outputStart + totalDecompressed); nioInput.position(inputStart + inputLen); return totalDecompressed; } - // Inflater runs in nowrap (raw DEFLATE) mode and is unaware of the gzip envelope, so it never - // checks the trailer. ZlibDirectDecompressor handles this automatically via GZIP_FORMAT, but - // for heap buffers we must verify the CRC32 and ISIZE fields ourselves. - private static void verifyGzipTrailer(byte[] inputData, int inputDataOffset, int inputLen, - byte[] outputData, int outputDataOffset, int decompressedLen) throws IOException { - long expectedCrc = readLittleEndianUInt32(inputData, inputDataOffset + inputLen - 8); - long expectedSize = readLittleEndianUInt32(inputData, inputDataOffset + inputLen - 4); - CRC32 crc32 = new CRC32(); - crc32.update(outputData, outputDataOffset, decompressedLen); - if (crc32.getValue() != expectedCrc) { - throw new IOException("Gzip CRC32 mismatch"); - } - if ((decompressedLen & 0xFFFFFFFFL) != expectedSize) { - throw new IOException("Gzip size mismatch"); - } - } - - private static long readLittleEndianUInt32(byte[] data, int offset) { - return (data[offset] & 0xFFL) | ((data[offset + 1] & 0xFFL) << 8) - | ((data[offset + 2] & 0xFFL) << 16) | ((data[offset + 3] & 0xFFL) << 24); - } - @Override public void reinit(@Nullable Compression.HFileDecompressionContext newHFileDecompressionContext) { if (newHFileDecompressionContext == null) { return; } - if (!(newHFileDecompressionContext instanceof GzipHFileDecompressionContext)) { + if (!(newHFileDecompressionContext instanceof GzipHFileDecompressionContext gzipContext)) { throw new IllegalArgumentException( "GzipByteBuffDecompressor#reinit() was given an HFileDecompressionContext that was not " + "a GzipHFileDecompressionContext, this should never happen"); } - GzipHFileDecompressionContext gzipContext = - (GzipHFileDecompressionContext) newHFileDecompressionContext; allowByteBuffDecompression = gzipContext.isAllowByteBuffDecompression(); } @Override public void close() { - inflater.end(); + if (heapDecompressor != null) { + heapDecompressor.end(); + } if (decompressor != null) { decompressor.end(); } diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipHFileDecompressionContext.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipHFileDecompressionContext.java index fc94bda2ceea..69bdc7ed10ba 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipHFileDecompressionContext.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipHFileDecompressionContext.java @@ -33,7 +33,9 @@ public final class GzipHFileDecompressionContext extends Compression.HFileDecomp public static final long FIXED_OVERHEAD = ClassSize.estimateBase(GzipHFileDecompressionContext.class, false); - // Intended to be set to false by some unit tests + public static final String ALLOW_BYTE_BUFF_DECOMPRESSION_KEY = + "hbase.io.compress.gz.allowByteBuffDecompression"; + private final boolean allowByteBuffDecompression; private GzipHFileDecompressionContext(boolean allowByteBuffDecompression) { @@ -46,7 +48,7 @@ public boolean isAllowByteBuffDecompression() { public static GzipHFileDecompressionContext fromConfiguration(Configuration conf) { return new GzipHFileDecompressionContext( - conf.getBoolean("hbase.io.compress.gz.allowByteBuffDecompression", true)); + conf.getBoolean(ALLOW_BYTE_BUFF_DECOMPRESSION_KEY, true)); } @Override diff --git a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java index 2df2ecd63b3c..52db16cec994 100644 --- a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java +++ b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java @@ -17,63 +17,57 @@ */ package org.apache.hadoop.hbase.io.compress; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; -import static org.junit.Assume.assumeTrue; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; +import static org.junit.jupiter.api.Assumptions.assumeTrue; +import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.UncheckedIOException; import java.nio.ByteBuffer; import java.util.Arrays; +import java.util.zip.GZIPOutputStream; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.HBaseClassTestRule; import org.apache.hadoop.hbase.nio.ByteBuff; import org.apache.hadoop.hbase.nio.MultiByteBuff; import org.apache.hadoop.hbase.nio.SingleByteBuff; import org.apache.hadoop.hbase.testclassification.SmallTests; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.util.NativeCodeLoader; -import org.junit.ClassRule; -import org.junit.Test; -import org.junit.experimental.categories.Category; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; -@Category(SmallTests.class) +@Tag(SmallTests.TAG) public class TestGzipByteBuffDecompressor { - @ClassRule - public static final HBaseClassTestRule CLASS_RULE = - HBaseClassTestRule.forClass(TestGzipByteBuffDecompressor.class); - - /* - * "HBase is fun to use and very fast" compressed as a single gzip member via GZIPOutputStream, - * matching the framing that ReusableStreamGzipCodec produces on the compression side. - */ - private static final byte[] COMPRESSED_PAYLOAD = Bytes.fromHex( - "1f8b08000000000000fff3704a2c4e55c82c56482bcd5328c9572805f212f35214ca528b2a15d2128b4b006edf170321000000"); + // A single gzip member, reused as decompressor input across the tests. + private static final byte[] COMPRESSED_PAYLOAD = gzip("HBase is fun to use and very fast"); /** * GzipByteBuffDecompressor is backed by Hadoop's native zlib binding, so actually decompressing * anything requires that native library to be loaded on this JVM. */ private static void assumeNativeZlibLoaded() { - assumeTrue("Hadoop's native code is not loaded on this JVM, skipping", - NativeCodeLoader.isNativeCodeLoaded()); + assumeTrue(NativeCodeLoader.isNativeCodeLoaded(), + "Hadoop's native code is not loaded on this JVM, skipping"); } @Test - public void testCapabilitiesWithoutNativeZlibLoaded() { - // Deliberately constructed as if native zlib is unavailable, regardless of this JVM's actual - // environment, so this test is deterministic everywhere. + public void itReportsCorrectCapabilitiesWithoutNativeZlib() { ByteBuff emptySingleDirectBuff = new SingleByteBuff(ByteBuffer.allocateDirect(0)); + ByteBuff emptySingleHeapBuff = new SingleByteBuff(ByteBuffer.allocate(0)); try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(false)) { - assertFalse("Without native zlib there is no way to decompress via ByteBuffs", - decompressor.canDecompress(emptySingleDirectBuff, emptySingleDirectBuff)); + assertFalse(decompressor.canDecompress(emptySingleDirectBuff, emptySingleDirectBuff), + "Without native zlib, direct-to-direct decompression is not available"); + assertFalse(decompressor.canDecompress(emptySingleHeapBuff, emptySingleHeapBuff), + "Without native zlib, heap decompression is not available"); } } @Test - public void testCapabilitiesWithNativeZlibLoaded() { + public void itReportsCorrectCapabilitiesWithNativeZlib() { assumeNativeZlibLoaded(); ByteBuff emptySingleHeapBuff = new SingleByteBuff(ByteBuffer.allocate(0)); ByteBuff emptyMultiHeapBuff = new MultiByteBuff(ByteBuffer.allocate(0), ByteBuffer.allocate(0)); @@ -83,9 +77,10 @@ public void testCapabilitiesWithNativeZlibLoaded() { try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { assertTrue(decompressor.canDecompress(emptySingleDirectBuff, emptySingleDirectBuff)); - // The native zlib binding reads/writes buffer memory directly, so only direct buffers are - // supported; heap buffers must fall back to stream-based decompression instead. - assertFalse(decompressor.canDecompress(emptySingleHeapBuff, emptySingleHeapBuff)); + assertTrue(decompressor.canDecompress(emptySingleHeapBuff, emptySingleHeapBuff), + "On-heap decompression is supported when both buffers are heap SingleByteBuffs"); + // Mixed (one direct, one heap) is not supported: decompressOnHeap() requires both to have + // backing arrays, and decompressOffHeap() requires both to be direct. assertFalse(decompressor.canDecompress(emptySingleHeapBuff, emptySingleDirectBuff)); assertFalse(decompressor.canDecompress(emptySingleDirectBuff, emptySingleHeapBuff)); assertFalse(decompressor.canDecompress(emptyMultiHeapBuff, emptyMultiHeapBuff)); @@ -101,8 +96,32 @@ private static ByteBuff directBuffWith(byte[] data) { return new SingleByteBuff(buffer); } + private static ByteBuff heapBuffWith(byte[] data) { + ByteBuffer buffer = ByteBuffer.allocate(data.length); + buffer.put(data); + buffer.rewind(); + return new SingleByteBuff(buffer); + } + + private static byte[] gzip(String text) { + ByteArrayOutputStream compressed = new ByteArrayOutputStream(); + try (GZIPOutputStream out = new GZIPOutputStream(compressed)) { + out.write(Bytes.toBytes(text)); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + return compressed.toByteArray(); + } + + private static byte[] concat(byte[] first, byte[] second) { + byte[] combined = new byte[first.length + second.length]; + System.arraycopy(first, 0, combined, 0, first.length); + System.arraycopy(second, 0, combined, first.length, second.length); + return combined; + } + @Test - public void testDecompressDirectToDirect() throws IOException { + public void itDecompressesDirectToDirectSuccessfully() throws IOException { assumeNativeZlibLoaded(); try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); @@ -114,7 +133,7 @@ public void testDecompressDirectToDirect() throws IOException { } @Test - public void testDecompressFailsOnTooShortInput() throws IOException { + public void itDecompressDirectFailsOnTooShortInput() throws IOException { assumeNativeZlibLoaded(); try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); @@ -127,7 +146,7 @@ public void testDecompressFailsOnTooShortInput() throws IOException { } @Test - public void testDecompressFailsOnBadMagicBytes() throws IOException { + public void itDecompressDirectFailsOnBadMagicBytes() throws IOException { assumeNativeZlibLoaded(); byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); corrupted[0] ^= (byte) 0xff; @@ -142,7 +161,7 @@ public void testDecompressFailsOnBadMagicBytes() throws IOException { } @Test - public void testDecompressFailsWhenOutputBufferTooSmall() throws IOException { + public void itDecompressDirectFailsWhenOutputBufferTooSmall() throws IOException { assumeNativeZlibLoaded(); try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(10)); @@ -155,7 +174,7 @@ public void testDecompressFailsWhenOutputBufferTooSmall() throws IOException { } @Test - public void testDecompressFailsOnCorruptedCrc32() throws IOException { + public void itDecompressDirectFailsOnCorruptedCrc32() throws IOException { assumeNativeZlibLoaded(); byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); // First 4 bytes of the 8-byte trailer are the CRC32, leave ISIZE (the last 4 bytes) alone. @@ -171,7 +190,7 @@ public void testDecompressFailsOnCorruptedCrc32() throws IOException { } @Test - public void testDecompressFailsOnCorruptedIsize() throws IOException { + public void itDecompressDirectFailsOnCorruptedIsize() throws IOException { assumeNativeZlibLoaded(); byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); // Last 4 bytes of the 8-byte trailer are the ISIZE. @@ -187,7 +206,7 @@ public void testDecompressFailsOnCorruptedIsize() throws IOException { } @Test - public void testDecompressSucceedsRepeatedlyOnTheSameDecompressor() throws IOException { + public void itDecompressesDirectSuccessfullyOnRepeatedCalls() throws IOException { assumeNativeZlibLoaded(); // Mirrors how CodecPool actually uses these: one instance is reused across many blocks, so the // native decompressor must produce a correct result on every call, not just the first. @@ -203,7 +222,7 @@ public void testDecompressSucceedsRepeatedlyOnTheSameDecompressor() throws IOExc } @Test - public void testDecompressorIsStillUsableAfterAPreviousCallThrows() throws IOException { + public void itDecompressDirectIsStillUsableAfterAPreviousCallThrows() throws IOException { assumeNativeZlibLoaded(); byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); // First 4 bytes of the 8-byte trailer are the CRC32. @@ -228,28 +247,82 @@ public void testDecompressorIsStillUsableAfterAPreviousCallThrows() throws IOExc } } + @Test + public void itDecompressesDirectToDirectWithNonZeroBufferPosition() throws IOException { + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuffer rawOutput = ByteBuffer.allocateDirect(128); + rawOutput.position(32); + + ByteBuffer rawInput = ByteBuffer.allocateDirect(16 + COMPRESSED_PAYLOAD.length); + for (int i = 0; i < 16; i++) { + rawInput.put((byte) 0); + } + rawInput.put(COMPRESSED_PAYLOAD); + rawInput.position(16); + + ByteBuff output = new SingleByteBuff(rawOutput); + ByteBuff input = new SingleByteBuff(rawInput); + int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + + byte[] result = new byte[decompressedSize]; + rawOutput.position(32); + rawOutput.get(result); + assertEquals("HBase is fun to use and very fast", Bytes.toString(result)); + } + } + + @Test + public void itDecompressDirectFailsOnTruncatedGzipStream() throws IOException { + assumeNativeZlibLoaded(); + byte[] truncated = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length - 4); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); + ByteBuff input = directBuffWith(truncated); + decompressor.decompress(output, input, truncated.length); + fail("Expected an IOException because the gzip stream is truncated"); + } catch (IOException e) { + // Expected: the decompressor must not report finished() on an incomplete stream + } + } + + @Test + public void itDecompressesOnlyTheDelimitedMemberFromAMultiMemberPayload() throws IOException { + assumeNativeZlibLoaded(); + // Two distinct members concatenated: we must decode only the one delimited by inputLen. + byte[] firstMember = gzip("first member"); + byte[] secondMember = gzip("second member"); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); + ByteBuff input = directBuffWith(concat(firstMember, secondMember)); + int decompressedSize = decompressor.decompress(output, input, firstMember.length); + assertEquals("first member", Bytes.toString(output.toBytes(0, decompressedSize))); + } + } + /** * This is the exact gate {@code HFileBlockDefaultDecodingContext#canDecompressViaByteBuff} relies * on to decide between ByteBuff decompression and the stream path, driven end-to-end from the - * {@code hbase.io.compress.gz.allowByteBuffDecompression} config flag. + * {@code GzipHFileDecompressionContext#ALLOW_BYTE_BUFF_DECOMPRESSION_KEY} config flag. */ @Test - public void testReinitControlsByteBuffDecompressionViaConfigFlag() { + public void itReinitControlsByteBuffDecompressionViaConfigFlag() { assumeNativeZlibLoaded(); try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); ByteBuff input = directBuffWith(COMPRESSED_PAYLOAD); Configuration conf = new Configuration(false); - conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", false); + conf.setBoolean(GzipHFileDecompressionContext.ALLOW_BYTE_BUFF_DECOMPRESSION_KEY, false); decompressor.reinit(GzipHFileDecompressionContext.fromConfiguration(conf)); - assertFalse("Block reader must fall back to stream decompression when the config flag " - + "disables ByteBuff decompression", decompressor.canDecompress(output, input)); + assertFalse(decompressor.canDecompress(output, input), + "Block reader must fall back to stream decompression when the config flag " + + "disables ByteBuff decompression"); - conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", true); + conf.setBoolean(GzipHFileDecompressionContext.ALLOW_BYTE_BUFF_DECOMPRESSION_KEY, true); decompressor.reinit(GzipHFileDecompressionContext.fromConfiguration(conf)); - assertTrue("Block reader must use ByteBuff decompression when the config flag is enabled", - decompressor.canDecompress(output, input)); + assertTrue(decompressor.canDecompress(output, input), + "Block reader must use ByteBuff decompression when the config flag is enabled"); // The default, with no config value set, must also allow ByteBuff decompression. decompressor @@ -259,25 +332,25 @@ public void testReinitControlsByteBuffDecompressionViaConfigFlag() { } @Test - public void testReinitWithNullContextIsNoOp() { + public void itReinitWithNullContextIsNoOp() { assumeNativeZlibLoaded(); try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64)); ByteBuff input = directBuffWith(COMPRESSED_PAYLOAD); Configuration conf = new Configuration(false); - conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", false); + conf.setBoolean(GzipHFileDecompressionContext.ALLOW_BYTE_BUFF_DECOMPRESSION_KEY, false); decompressor.reinit(GzipHFileDecompressionContext.fromConfiguration(conf)); assertFalse(decompressor.canDecompress(output, input)); decompressor.reinit(null); - assertFalse("reinit(null) must not reset allowByteBuffDecompression back to the default", - decompressor.canDecompress(output, input)); + assertFalse(decompressor.canDecompress(output, input), + "reinit(null) must not reset allowByteBuffDecompression back to the default"); } } @Test - public void testReinitFailsOnWrongContextType() { + public void itReinitFailsOnWrongContextType() { Compression.HFileDecompressionContext wrongContext = new Compression.HFileDecompressionContext() { @Override @@ -298,4 +371,208 @@ public long heapSize() { } } + @Test + public void itDecompressThrowsWhenPassedAMultiByteBuff() throws IOException { + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(false)) { + ByteBuff multiOutput = new MultiByteBuff(ByteBuffer.allocate(64), ByteBuffer.allocate(64)); + ByteBuff input = heapBuffWith(COMPRESSED_PAYLOAD); + decompressor.decompress(multiOutput, input, COMPRESSED_PAYLOAD.length); + fail("Expected an IllegalStateException when output is a MultiByteBuff"); + } catch (IllegalStateException e) { + assertTrue(e.getMessage().contains("not a SingleByteBuff")); + } + } + + // On-heap (heap -> heap) decompression tests; these also require native zlib. + + @Test + public void itDecompressesHeapToHeapSuccessfully() throws IOException { + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = heapBuffWith(COMPRESSED_PAYLOAD); + int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + assertEquals("HBase is fun to use and very fast", + Bytes.toString(output.toBytes(0, decompressedSize))); + } + } + + @Test + public void itDecompressesHeapToHeapSuccessfullyWhenNativeZlibIsAlsoAvailable() + throws IOException { + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = heapBuffWith(COMPRESSED_PAYLOAD); + int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + assertEquals("HBase is fun to use and very fast", + Bytes.toString(output.toBytes(0, decompressedSize))); + } + } + + @Test + public void itDecompressesHeapToHeapFailsWhenOutputBufferTooSmall() throws IOException { + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(10)); + ByteBuff input = heapBuffWith(COMPRESSED_PAYLOAD); + decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + fail("Expected an IOException because the output buffer is too small"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("Output buffer is too small")); + } + } + + @Test + public void itDecompressesHeapToHeapFailsOnTooShortInput() throws IOException { + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(false)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = new SingleByteBuff(ByteBuffer.allocate(10)); + decompressor.decompress(output, input, 10); + fail("Expected an IOException because the input is too short to be a gzip member"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("too short to be a gzip member")); + } + } + + @Test + public void itDecompressesHeapToHeapFailsOnCorruptedDeflateBody() throws IOException { + // Corrupt a byte inside the DEFLATE payload (bytes 10 through len-9). ZlibDecompressor with + // GZIP_FORMAT will reject the corrupted data via native zlib's CRC/format checks. + byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); + corrupted[15] ^= (byte) 0xff; + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = heapBuffWith(corrupted); + decompressor.decompress(output, input, corrupted.length); + fail("Expected an IOException because the DEFLATE payload is corrupted"); + } catch (IOException e) { + // Expected; the exact message depends on native zlib internals + } + } + + @Test + public void itDecompressesHeapToHeapFailsOnCrc32Mismatch() throws IOException { + // Corrupt the first 4 bytes of the 8-byte trailer (CRC32), leaving ISIZE intact. + // ZlibDecompressor with GZIP_FORMAT verifies the CRC32 field via native zlib. + byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); + corrupted[corrupted.length - 8] ^= (byte) 0xff; + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = heapBuffWith(corrupted); + decompressor.decompress(output, input, corrupted.length); + fail("Expected an IOException due to a CRC32 mismatch in the gzip trailer"); + } catch (IOException e) { + // Expected; ZlibDecompressor delegates CRC32 verification to native zlib + } + } + + @Test + public void itDecompressesHeapToHeapFailsOnIsizeMismatch() throws IOException { + // Corrupt the last 4 bytes of the 8-byte trailer (ISIZE), leaving CRC32 intact. + // ZlibDecompressor with GZIP_FORMAT verifies the ISIZE field via native zlib. + byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); + corrupted[corrupted.length - 4] ^= (byte) 0xff; + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = heapBuffWith(corrupted); + decompressor.decompress(output, input, corrupted.length); + fail("Expected an IOException due to an ISIZE mismatch in the gzip trailer"); + } catch (IOException e) { + // Expected; ZlibDecompressor delegates ISIZE verification to native zlib + } + } + + @Test + public void itDecompressesHeapToHeapSucceedsRepeatedly() throws IOException { + // ZlibDecompressor must be reset between calls; verify multiple sequential decompressions on + // the same instance all produce correct output. + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + for (int i = 0; i < 3; i++) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = heapBuffWith(COMPRESSED_PAYLOAD); + int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + assertEquals("HBase is fun to use and very fast", + Bytes.toString(output.toBytes(0, decompressedSize))); + } + } + } + + @Test + public void itDecompressesHeapToHeapIsStillUsableAfterAPreviousCallThrows() throws IOException { + byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); + corrupted[corrupted.length - 8] ^= (byte) 0xff; + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff badOutput = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff badInput = heapBuffWith(corrupted); + try { + decompressor.decompress(badOutput, badInput, corrupted.length); + fail("Expected an IOException because the CRC32 is corrupted"); + } catch (IOException e) { + // Expected + } + + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = heapBuffWith(COMPRESSED_PAYLOAD); + int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + assertEquals("HBase is fun to use and very fast", + Bytes.toString(output.toBytes(0, decompressedSize))); + } + } + + @Test + public void itDecompressesHeapToHeapSuccessfullyWithNonZeroBufferPosition() throws IOException { + assumeNativeZlibLoaded(); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuffer rawOutput = ByteBuffer.allocate(128); + rawOutput.position(32); + + ByteBuffer rawInput = ByteBuffer.allocate(16 + COMPRESSED_PAYLOAD.length); + rawInput.put(new byte[16]); + rawInput.put(COMPRESSED_PAYLOAD); + rawInput.position(16); + + ByteBuff output = new SingleByteBuff(rawOutput); + ByteBuff input = new SingleByteBuff(rawInput); + int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); + + assertEquals("HBase is fun to use and very fast", + Bytes.toString(rawOutput.array(), rawOutput.arrayOffset() + 32, decompressedSize)); + } + } + + @Test + public void itDecompressesHeapToHeapFailsOnTruncatedGzipStream() throws IOException { + assumeNativeZlibLoaded(); + byte[] truncated = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length - 4); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = heapBuffWith(truncated); + decompressor.decompress(output, input, truncated.length); + fail("Expected an IOException because the gzip stream is truncated"); + } catch (IOException e) { + // Expected: the decompressor must not report finished() on an incomplete stream + } + } + + @Test + public void itDecompressesOnlyTheDelimitedMemberFromAMultiMemberPayloadOnHeap() + throws IOException { + assumeNativeZlibLoaded(); + // Two distinct members concatenated, on the on-heap path: decode only the delimited one. + byte[] firstMember = gzip("first member"); + byte[] secondMember = gzip("second member"); + try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { + ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); + ByteBuff input = heapBuffWith(concat(firstMember, secondMember)); + int decompressedSize = decompressor.decompress(output, input, firstMember.length); + assertEquals("first member", Bytes.toString(output.toBytes(0, decompressedSize))); + } + } + } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java index 1f689235b5a0..b7d5e31ef921 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java @@ -19,31 +19,26 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; -import org.apache.hadoop.hbase.HBaseClassTestRule; -import org.apache.hadoop.hbase.HBaseTestingUtility; +import org.apache.hadoop.hbase.HBaseTestingUtil; import org.apache.hadoop.hbase.testclassification.IOTests; import org.apache.hadoop.hbase.testclassification.SmallTests; -import org.junit.Before; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Test; -import org.junit.experimental.categories.Category; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; -@Category({ IOTests.class, SmallTests.class }) +@Tag(IOTests.TAG) +@Tag(SmallTests.TAG) public class TestHFileCompressionGzip extends HFileTestBase { - @ClassRule - public static final HBaseClassTestRule CLASS_RULE = - HBaseClassTestRule.forClass(TestHFileCompressionGzip.class); - private static Configuration conf; - @BeforeClass + @BeforeAll public static void setUpBeforeClass() throws Exception { HFileTestBase.setUpBeforeClass(); } - @Before + @BeforeEach public void setUp() throws Exception { conf = TEST_UTIL.getConfiguration(); HFileTestBase.setUpBeforeClass(); @@ -54,8 +49,8 @@ public void testWithStreamDecompression() throws Exception { conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", false); Compression.Algorithm.GZ.reload(conf); - Path path = new Path(TEST_UTIL.getDataTestDir(), - HBaseTestingUtility.getRandomUUID().toString() + ".hfile"); + Path path = + new Path(TEST_UTIL.getDataTestDir(), HBaseTestingUtil.getRandomUUID().toString() + ".hfile"); doTest(conf, path, Compression.Algorithm.GZ); } @@ -64,8 +59,8 @@ public void testWithByteBuffDecompression() throws Exception { conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", true); Compression.Algorithm.GZ.reload(conf); - Path path = new Path(TEST_UTIL.getDataTestDir(), - HBaseTestingUtility.getRandomUUID().toString() + ".hfile"); + Path path = + new Path(TEST_UTIL.getDataTestDir(), HBaseTestingUtil.getRandomUUID().toString() + ".hfile"); doTest(conf, path, Compression.Algorithm.GZ); } From 927e7115e61536b4758a06c33dd5d569298cf555 Mon Sep 17 00:00:00 2001 From: sahmadsabri Date: Mon, 10 Aug 2026 12:15:52 -0400 Subject: [PATCH 10/12] Sync GzipByteBuffDecompressor changes from HBASE-30321-gzipbytebuff Pulls latest refactor that removes heap decompression logic and supports only direct-to-direct decompression via ZlibDirectDecompressor (zero-copy path). Heap/mixed buffer callers fall back to the stream path. Co-Authored-By: Claude Sonnet 4.6 --- .../io/compress/GzipByteBuffDecompressor.java | 88 ++------ .../TestGzipByteBuffDecompressor.java | 200 +----------------- 2 files changed, 15 insertions(+), 273 deletions(-) diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java index 26bb49c5b0df..e484141f9b37 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java @@ -27,7 +27,8 @@ /** * Glue for ByteBuffDecompressor on top of Hadoop's native - * {@link ZlibDecompressor.ZlibDirectDecompressor}. + * {@link ZlibDecompressor.ZlibDirectDecompressor}. Only direct-to-direct decompression is + * supported, which is zero-copy; callers with on-heap buffers fall back to the stream path. */ @InterfaceAudience.Private public class GzipByteBuffDecompressor implements ByteBuffDecompressor { @@ -35,15 +36,9 @@ public class GzipByteBuffDecompressor implements ByteBuffDecompressor { private static final int GZIP_HEADER_LENGTH = 10; private static final int GZIP_TRAILER_LENGTH = 8; - // Size of the on-heap ZlibDecompressor's internal staging buffers - private static final int HEAP_DECOMPRESSOR_BUFFER_SIZE = 64 * 1024; - @Nullable private final ZlibDecompressor.ZlibDirectDecompressor decompressor; - @Nullable - private final ZlibDecompressor heapDecompressor; - private boolean allowByteBuffDecompression; GzipByteBuffDecompressor(boolean nativeZlibLoaded) { @@ -51,10 +46,6 @@ public class GzipByteBuffDecompressor implements ByteBuffDecompressor { ? new ZlibDecompressor.ZlibDirectDecompressor(ZlibDecompressor.CompressionHeader.GZIP_FORMAT, 0) : null; - heapDecompressor = nativeZlibLoaded - ? new ZlibDecompressor(ZlibDecompressor.CompressionHeader.GZIP_FORMAT, - HEAP_DECOMPRESSOR_BUFFER_SIZE) - : null; allowByteBuffDecompression = true; } @@ -66,15 +57,9 @@ public boolean canDecompress(ByteBuff output, ByteBuff input) { if (!(output instanceof SingleByteBuff) || !(input instanceof SingleByteBuff)) { return false; } - boolean inputDirect = input.nioByteBuffers()[0].isDirect(); - boolean outputDirect = output.nioByteBuffers()[0].isDirect(); - if (inputDirect && outputDirect) { - return decompressor != null; - } - if (inputDirect != outputDirect) { - return false; - } - return !inputDirect && !outputDirect && heapDecompressor != null; + // Only direct-to-direct decompression is supported. + return input.nioByteBuffers()[0].isDirect() && output.nioByteBuffers()[0].isDirect() + && decompressor != null; } @Override @@ -89,25 +74,12 @@ public int decompress(ByteBuff output, ByteBuff input, int inputLen) throws IOEx ByteBuffer nioInput = input.nioByteBuffers()[0]; ByteBuffer nioOutput = output.nioByteBuffers()[0]; - boolean inputDirect = nioInput.isDirect(); - boolean outputDirect = nioOutput.isDirect(); - - if (inputDirect && outputDirect) { - if (decompressor == null) { - throw new IllegalStateException( - "GzipByteBuffDecompressor#decompress() was called with direct buffers but Hadoop's " - + "native zlib library is not loaded, this should never happen since " - + "canDecompress() would have returned false"); - } - return decompressOffHeap(nioInput, nioOutput, inputLen); - } - if (heapDecompressor == null) { + if (!nioInput.isDirect() || !nioOutput.isDirect() || decompressor == null) { throw new IllegalStateException( - "GzipByteBuffDecompressor#decompress() was called with heap buffers but Hadoop's " - + "native zlib library is not loaded, this should never happen since " - + "canDecompress() would have returned false"); + "GzipByteBuffDecompressor only supports direct-to-direct decompression with native zlib " + + "loaded, this should never happen since canDecompress() would have returned false"); } - return decompressOnHeap(nioInput, nioOutput, inputLen); + return decompressOffHeap(nioInput, nioOutput, inputLen); } private int decompressOffHeap(ByteBuffer nioInput, ByteBuffer nioOutput, int inputLen) @@ -139,59 +111,23 @@ private int decompressOffHeap(ByteBuffer nioInput, ByteBuffer nioOutput, int inp return nioOutput.position() - outputStart; } - private int decompressOnHeap(ByteBuffer nioInput, ByteBuffer nioOutput, int inputLen) - throws IOException { - if (!nioInput.hasArray() || !nioOutput.hasArray()) { - throw new IllegalStateException( - "decompressOnHeap() requires heap ByteBuffers with backing arrays"); - } - int inputStart = nioInput.position(); - int outputStart = nioOutput.position(); - int outputCapacity = nioOutput.remaining(); - - heapDecompressor.reset(); - heapDecompressor.setInput(nioInput.array(), nioInput.arrayOffset() + inputStart, inputLen); - int totalDecompressed = 0; - while (!heapDecompressor.finished()) { - int remaining = outputCapacity - totalDecompressed; - if (remaining == 0) { - throw new IOException("Output buffer is too small for the decompressed gzip stream"); - } - int n = heapDecompressor.decompress(nioOutput.array(), - nioOutput.arrayOffset() + outputStart + totalDecompressed, remaining); - if (n == 0 && !heapDecompressor.finished()) { - if (heapDecompressor.needsInput()) { - throw new IOException("Unexpected end of gzip stream"); - } - throw new IOException( - "Gzip decompressor made no progress and is not finished; aborting to avoid an " - + "infinite loop"); - } - totalDecompressed += n; - } - nioOutput.position(outputStart + totalDecompressed); - nioInput.position(inputStart + inputLen); - return totalDecompressed; - } - @Override public void reinit(@Nullable Compression.HFileDecompressionContext newHFileDecompressionContext) { if (newHFileDecompressionContext == null) { return; } - if (!(newHFileDecompressionContext instanceof GzipHFileDecompressionContext gzipContext)) { + if (!(newHFileDecompressionContext instanceof GzipHFileDecompressionContext)) { throw new IllegalArgumentException( "GzipByteBuffDecompressor#reinit() was given an HFileDecompressionContext that was not " + "a GzipHFileDecompressionContext, this should never happen"); } + GzipHFileDecompressionContext gzipContext = + (GzipHFileDecompressionContext) newHFileDecompressionContext; allowByteBuffDecompression = gzipContext.isAllowByteBuffDecompression(); } @Override public void close() { - if (heapDecompressor != null) { - heapDecompressor.end(); - } if (decompressor != null) { decompressor.end(); } diff --git a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java index 52db16cec994..2b86c0ab042d 100644 --- a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java +++ b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java @@ -62,7 +62,7 @@ public void itReportsCorrectCapabilitiesWithoutNativeZlib() { assertFalse(decompressor.canDecompress(emptySingleDirectBuff, emptySingleDirectBuff), "Without native zlib, direct-to-direct decompression is not available"); assertFalse(decompressor.canDecompress(emptySingleHeapBuff, emptySingleHeapBuff), - "Without native zlib, heap decompression is not available"); + "Heap decompression is not supported; only direct-to-direct"); } } @@ -77,10 +77,8 @@ public void itReportsCorrectCapabilitiesWithNativeZlib() { try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { assertTrue(decompressor.canDecompress(emptySingleDirectBuff, emptySingleDirectBuff)); - assertTrue(decompressor.canDecompress(emptySingleHeapBuff, emptySingleHeapBuff), - "On-heap decompression is supported when both buffers are heap SingleByteBuffs"); - // Mixed (one direct, one heap) is not supported: decompressOnHeap() requires both to have - // backing arrays, and decompressOffHeap() requires both to be direct. + // Only direct-to-direct is supported; heap and mixed buffers return false. + assertFalse(decompressor.canDecompress(emptySingleHeapBuff, emptySingleHeapBuff)); assertFalse(decompressor.canDecompress(emptySingleHeapBuff, emptySingleDirectBuff)); assertFalse(decompressor.canDecompress(emptySingleDirectBuff, emptySingleHeapBuff)); assertFalse(decompressor.canDecompress(emptyMultiHeapBuff, emptyMultiHeapBuff)); @@ -383,196 +381,4 @@ public void itDecompressThrowsWhenPassedAMultiByteBuff() throws IOException { } } - // On-heap (heap -> heap) decompression tests; these also require native zlib. - - @Test - public void itDecompressesHeapToHeapSuccessfully() throws IOException { - assumeNativeZlibLoaded(); - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = heapBuffWith(COMPRESSED_PAYLOAD); - int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); - assertEquals("HBase is fun to use and very fast", - Bytes.toString(output.toBytes(0, decompressedSize))); - } - } - - @Test - public void itDecompressesHeapToHeapSuccessfullyWhenNativeZlibIsAlsoAvailable() - throws IOException { - assumeNativeZlibLoaded(); - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = heapBuffWith(COMPRESSED_PAYLOAD); - int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); - assertEquals("HBase is fun to use and very fast", - Bytes.toString(output.toBytes(0, decompressedSize))); - } - } - - @Test - public void itDecompressesHeapToHeapFailsWhenOutputBufferTooSmall() throws IOException { - assumeNativeZlibLoaded(); - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(10)); - ByteBuff input = heapBuffWith(COMPRESSED_PAYLOAD); - decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); - fail("Expected an IOException because the output buffer is too small"); - } catch (IOException e) { - assertTrue(e.getMessage().contains("Output buffer is too small")); - } - } - - @Test - public void itDecompressesHeapToHeapFailsOnTooShortInput() throws IOException { - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(false)) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = new SingleByteBuff(ByteBuffer.allocate(10)); - decompressor.decompress(output, input, 10); - fail("Expected an IOException because the input is too short to be a gzip member"); - } catch (IOException e) { - assertTrue(e.getMessage().contains("too short to be a gzip member")); - } - } - - @Test - public void itDecompressesHeapToHeapFailsOnCorruptedDeflateBody() throws IOException { - // Corrupt a byte inside the DEFLATE payload (bytes 10 through len-9). ZlibDecompressor with - // GZIP_FORMAT will reject the corrupted data via native zlib's CRC/format checks. - byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); - corrupted[15] ^= (byte) 0xff; - assumeNativeZlibLoaded(); - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = heapBuffWith(corrupted); - decompressor.decompress(output, input, corrupted.length); - fail("Expected an IOException because the DEFLATE payload is corrupted"); - } catch (IOException e) { - // Expected; the exact message depends on native zlib internals - } - } - - @Test - public void itDecompressesHeapToHeapFailsOnCrc32Mismatch() throws IOException { - // Corrupt the first 4 bytes of the 8-byte trailer (CRC32), leaving ISIZE intact. - // ZlibDecompressor with GZIP_FORMAT verifies the CRC32 field via native zlib. - byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); - corrupted[corrupted.length - 8] ^= (byte) 0xff; - assumeNativeZlibLoaded(); - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = heapBuffWith(corrupted); - decompressor.decompress(output, input, corrupted.length); - fail("Expected an IOException due to a CRC32 mismatch in the gzip trailer"); - } catch (IOException e) { - // Expected; ZlibDecompressor delegates CRC32 verification to native zlib - } - } - - @Test - public void itDecompressesHeapToHeapFailsOnIsizeMismatch() throws IOException { - // Corrupt the last 4 bytes of the 8-byte trailer (ISIZE), leaving CRC32 intact. - // ZlibDecompressor with GZIP_FORMAT verifies the ISIZE field via native zlib. - byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); - corrupted[corrupted.length - 4] ^= (byte) 0xff; - assumeNativeZlibLoaded(); - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = heapBuffWith(corrupted); - decompressor.decompress(output, input, corrupted.length); - fail("Expected an IOException due to an ISIZE mismatch in the gzip trailer"); - } catch (IOException e) { - // Expected; ZlibDecompressor delegates ISIZE verification to native zlib - } - } - - @Test - public void itDecompressesHeapToHeapSucceedsRepeatedly() throws IOException { - // ZlibDecompressor must be reset between calls; verify multiple sequential decompressions on - // the same instance all produce correct output. - assumeNativeZlibLoaded(); - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { - for (int i = 0; i < 3; i++) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = heapBuffWith(COMPRESSED_PAYLOAD); - int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); - assertEquals("HBase is fun to use and very fast", - Bytes.toString(output.toBytes(0, decompressedSize))); - } - } - } - - @Test - public void itDecompressesHeapToHeapIsStillUsableAfterAPreviousCallThrows() throws IOException { - byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length); - corrupted[corrupted.length - 8] ^= (byte) 0xff; - assumeNativeZlibLoaded(); - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { - ByteBuff badOutput = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff badInput = heapBuffWith(corrupted); - try { - decompressor.decompress(badOutput, badInput, corrupted.length); - fail("Expected an IOException because the CRC32 is corrupted"); - } catch (IOException e) { - // Expected - } - - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = heapBuffWith(COMPRESSED_PAYLOAD); - int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); - assertEquals("HBase is fun to use and very fast", - Bytes.toString(output.toBytes(0, decompressedSize))); - } - } - - @Test - public void itDecompressesHeapToHeapSuccessfullyWithNonZeroBufferPosition() throws IOException { - assumeNativeZlibLoaded(); - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { - ByteBuffer rawOutput = ByteBuffer.allocate(128); - rawOutput.position(32); - - ByteBuffer rawInput = ByteBuffer.allocate(16 + COMPRESSED_PAYLOAD.length); - rawInput.put(new byte[16]); - rawInput.put(COMPRESSED_PAYLOAD); - rawInput.position(16); - - ByteBuff output = new SingleByteBuff(rawOutput); - ByteBuff input = new SingleByteBuff(rawInput); - int decompressedSize = decompressor.decompress(output, input, COMPRESSED_PAYLOAD.length); - - assertEquals("HBase is fun to use and very fast", - Bytes.toString(rawOutput.array(), rawOutput.arrayOffset() + 32, decompressedSize)); - } - } - - @Test - public void itDecompressesHeapToHeapFailsOnTruncatedGzipStream() throws IOException { - assumeNativeZlibLoaded(); - byte[] truncated = Arrays.copyOf(COMPRESSED_PAYLOAD, COMPRESSED_PAYLOAD.length - 4); - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = heapBuffWith(truncated); - decompressor.decompress(output, input, truncated.length); - fail("Expected an IOException because the gzip stream is truncated"); - } catch (IOException e) { - // Expected: the decompressor must not report finished() on an incomplete stream - } - } - - @Test - public void itDecompressesOnlyTheDelimitedMemberFromAMultiMemberPayloadOnHeap() - throws IOException { - assumeNativeZlibLoaded(); - // Two distinct members concatenated, on the on-heap path: decode only the delimited one. - byte[] firstMember = gzip("first member"); - byte[] secondMember = gzip("second member"); - try (GzipByteBuffDecompressor decompressor = new GzipByteBuffDecompressor(true)) { - ByteBuff output = new SingleByteBuff(ByteBuffer.allocate(64)); - ByteBuff input = heapBuffWith(concat(firstMember, secondMember)); - int decompressedSize = decompressor.decompress(output, input, firstMember.length); - assertEquals("first member", Bytes.toString(output.toBytes(0, decompressedSize))); - } - } - } From 60ebea06f531217a364b201ef92448e5cad1da7f Mon Sep 17 00:00:00 2001 From: sahmadsabri Date: Mon, 10 Aug 2026 12:31:16 -0400 Subject: [PATCH 11/12] Fix import typo in TestHFileCompressionGzip for HBaseTestingUtility --- .../hbase/io/compress/TestHFileCompressionGzip.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java index b7d5e31ef921..8bcb0e5361cd 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java @@ -19,7 +19,7 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; -import org.apache.hadoop.hbase.HBaseTestingUtil; +import org.apache.hadoop.hbase.HBaseTestingUtility; import org.apache.hadoop.hbase.testclassification.IOTests; import org.apache.hadoop.hbase.testclassification.SmallTests; import org.junit.jupiter.api.BeforeAll; @@ -49,8 +49,8 @@ public void testWithStreamDecompression() throws Exception { conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", false); Compression.Algorithm.GZ.reload(conf); - Path path = - new Path(TEST_UTIL.getDataTestDir(), HBaseTestingUtil.getRandomUUID().toString() + ".hfile"); + Path path = new Path(TEST_UTIL.getDataTestDir(), + HBaseTestingUtility.getRandomUUID().toString() + ".hfile"); doTest(conf, path, Compression.Algorithm.GZ); } @@ -59,8 +59,8 @@ public void testWithByteBuffDecompression() throws Exception { conf.setBoolean("hbase.io.compress.gz.allowByteBuffDecompression", true); Compression.Algorithm.GZ.reload(conf); - Path path = - new Path(TEST_UTIL.getDataTestDir(), HBaseTestingUtil.getRandomUUID().toString() + ".hfile"); + Path path = new Path(TEST_UTIL.getDataTestDir(), + HBaseTestingUtility.getRandomUUID().toString() + ".hfile"); doTest(conf, path, Compression.Algorithm.GZ); } From 422e10c6964d4effaef0ad9c8b7623b1b46be0a8 Mon Sep 17 00:00:00 2001 From: sahmadsabri Date: Tue, 18 Aug 2026 10:15:48 -0500 Subject: [PATCH 12/12] Update GzipByteBuffDecompressor to use AUTODETECT_GZIP_ZLIB for improved format detection --- .../hadoop/hbase/io/compress/GzipByteBuffDecompressor.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java index e484141f9b37..92ff19a183d0 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java @@ -43,7 +43,7 @@ public class GzipByteBuffDecompressor implements ByteBuffDecompressor { GzipByteBuffDecompressor(boolean nativeZlibLoaded) { decompressor = nativeZlibLoaded - ? new ZlibDecompressor.ZlibDirectDecompressor(ZlibDecompressor.CompressionHeader.GZIP_FORMAT, + ? new ZlibDecompressor.ZlibDirectDecompressor(ZlibDecompressor.CompressionHeader.AUTODETECT_GZIP_ZLIB, 0) : null; allowByteBuffDecompression = true;