From 281fe5c5363ce946792a619c4a05c6d51e43df8c Mon Sep 17 00:00:00 2001 From: sahmadsabri Date: Mon, 3 Aug 2026 21:53:15 -0500 Subject: [PATCH] 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); + } + +}