This is an automated email from the ASF dual-hosted git repository.
charlesconnell pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/master by this push:
new 1f44b91eb5e HBASE-30321: Implement GzipByteBuffDecompressor with
on-heap and off-heap decompression paths (#8541)
1f44b91eb5e is described below
commit 1f44b91eb5e08510e3c043aba430819bdf5a0433
Author: Saad Ahmad Sabri <[email protected]>
AuthorDate: Tue Aug 18 18:50:23 2026 -0400
HBASE-30321: Implement GzipByteBuffDecompressor with on-heap and off-heap
decompression paths (#8541)
Signed-off by: Charles Connell <[email protected]>
---
.../io/compress/GzipByteBuffDecompressor.java | 136 ++++++++
.../io/compress/GzipHFileDecompressionContext.java | 68 ++++
.../hbase/io/compress/ReusableStreamGzipCodec.java | 21 +-
.../io/compress/TestGzipByteBuffDecompressor.java | 384 +++++++++++++++++++++
.../io/compress/TestHFileCompressionGzip.java | 67 ++++
5 files changed, 674 insertions(+), 2 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
new file mode 100644
index 00000000000..b11af0ec3c5
--- /dev/null
+++
b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipByteBuffDecompressor.java
@@ -0,0 +1,136 @@
+/*
+ * 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 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}. Only direct-to-direct
decompression is
+ * supported, which is zero-copy; callers with on-heap buffers fall back to
the stream path.
+ */
[email protected]
+public class GzipByteBuffDecompressor implements ByteBuffDecompressor {
+
+ private static final int GZIP_HEADER_LENGTH = 10;
+ private static final int GZIP_TRAILER_LENGTH = 8;
+
+ @Nullable
+ private final ZlibDecompressor.ZlibDirectDecompressor decompressor;
+
+ private boolean allowByteBuffDecompression;
+
+ GzipByteBuffDecompressor(boolean nativeZlibLoaded) {
+ decompressor = nativeZlibLoaded
+ ? new ZlibDecompressor.ZlibDirectDecompressor(
+ ZlibDecompressor.CompressionHeader.AUTODETECT_GZIP_ZLIB, 0)
+ : null;
+ allowByteBuffDecompression = true;
+ }
+
+ @Override
+ public boolean canDecompress(ByteBuff output, ByteBuff input) {
+ if (!allowByteBuffDecompression) {
+ return false;
+ }
+ if (!(output instanceof SingleByteBuff) || !(input instanceof
SingleByteBuff)) {
+ return false;
+ }
+ // Only direct-to-direct decompression is supported.
+ return input.nioByteBuffers()[0].isDirect() &&
output.nioByteBuffers()[0].isDirect()
+ && decompressor != null;
+ }
+
+ @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];
+ ByteBuffer nioOutput = output.nioByteBuffers()[0];
+ if (!nioInput.isDirect() || !nioOutput.isDirect() || decompressor == null)
{
+ throw new IllegalStateException(
+ "GzipByteBuffDecompressor only supports direct-to-direct decompression
with native zlib "
+ + "loaded, this should never happen since canDecompress() would have
returned false");
+ }
+ return decompressOffHeap(nioInput, nioOutput, inputLen);
+ }
+
+ private int decompressOffHeap(ByteBuffer nioInput, ByteBuffer nioOutput, int
inputLen)
+ throws IOException {
+ int inputStart = nioInput.position();
+ int outputStart = nioOutput.position();
+
+ ByteBuffer gzipMember = nioInput.duplicate();
+ gzipMember.limit(inputStart + inputLen);
+
+ decompressor.reset();
+ 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;
+ }
+
+ @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() {
+ 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
new file mode 100644
index 00000000000..69bdc7ed10b
--- /dev/null
+++
b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/GzipHFileDecompressionContext.java
@@ -0,0 +1,68 @@
+/*
+ * 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.
+ */
[email protected]
+public final class GzipHFileDecompressionContext extends
Compression.HFileDecompressionContext {
+
+ public static final long FIXED_OVERHEAD =
+ ClassSize.estimateBase(GzipHFileDecompressionContext.class, false);
+
+ public static final String ALLOW_BYTE_BUFF_DECOMPRESSION_KEY =
+ "hbase.io.compress.gz.allowByteBuffDecompression";
+
+ 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(ALLOW_BYTE_BUFF_DECOMPRESSION_KEY, 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 23aac29981c..08276d7f073 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.IOException;
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,9 +36,9 @@ import org.slf4j.LoggerFactory;
* 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(ReusableStreamGzipCodec.class);
+ private static final Logger LOG = LoggerFactory.getLogger(Compression.class);
/**
* A bridge that wraps around a DeflaterOutputStream to make it a
CompressionOutputStream.
@@ -185,4 +186,20 @@ public class ReusableStreamGzipCodec extends GzipCodec {
return new ReusableGzipOutputStream(out);
}
+ @Override
+ public ByteBuffDecompressor createByteBuffDecompressor() {
+ return new
GzipByteBuffDecompressor(ZlibFactory.isNativeZlibLoaded(getConf()));
+ }
+
+ @Override
+ public Class<? extends ByteBuffDecompressor> 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 00000000000..2b86c0ab042
--- /dev/null
+++
b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/compress/TestGzipByteBuffDecompressor.java
@@ -0,0 +1,384 @@
+/*
+ * 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.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.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.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+@Tag(SmallTests.TAG)
+public class TestGzipByteBuffDecompressor {
+
+ // 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(NativeCodeLoader.isNativeCodeLoaded(),
+ "Hadoop's native code is not loaded on this JVM, skipping");
+ }
+
+ @Test
+ public void itReportsCorrectCapabilitiesWithoutNativeZlib() {
+ ByteBuff emptySingleDirectBuff = new
SingleByteBuff(ByteBuffer.allocateDirect(0));
+ ByteBuff emptySingleHeapBuff = new SingleByteBuff(ByteBuffer.allocate(0));
+ try (GzipByteBuffDecompressor decompressor = new
GzipByteBuffDecompressor(false)) {
+ assertFalse(decompressor.canDecompress(emptySingleDirectBuff,
emptySingleDirectBuff),
+ "Without native zlib, direct-to-direct decompression is not
available");
+ assertFalse(decompressor.canDecompress(emptySingleHeapBuff,
emptySingleHeapBuff),
+ "Heap decompression is not supported; only direct-to-direct");
+ }
+ }
+
+ @Test
+ public void itReportsCorrectCapabilitiesWithNativeZlib() {
+ 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(true)) {
+ assertTrue(decompressor.canDecompress(emptySingleDirectBuff,
emptySingleDirectBuff));
+ // 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));
+ assertFalse(decompressor.canDecompress(emptyMultiDirectBuff,
emptyMultiDirectBuff));
+ assertFalse(decompressor.canDecompress(emptySingleDirectBuff,
emptyMultiDirectBuff));
+ }
+ }
+
+ private static ByteBuff directBuffWith(byte[] data) {
+ ByteBuffer buffer = ByteBuffer.allocateDirect(data.length);
+ buffer.put(data);
+ buffer.rewind();
+ 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 itDecompressesDirectToDirectSuccessfully() throws IOException {
+ assumeNativeZlibLoaded();
+ try (GzipByteBuffDecompressor decompressor = new
GzipByteBuffDecompressor(true)) {
+ 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)));
+ }
+ }
+
+ @Test
+ public void itDecompressDirectFailsOnTooShortInput() throws IOException {
+ assumeNativeZlibLoaded();
+ try (GzipByteBuffDecompressor decompressor = new
GzipByteBuffDecompressor(true)) {
+ ByteBuff output = new SingleByteBuff(ByteBuffer.allocateDirect(64));
+ 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) {
+ assertTrue(e.getMessage().contains("too short to be a gzip member"));
+ }
+ }
+
+ @Test
+ public void itDecompressDirectFailsOnBadMagicBytes() throws IOException {
+ assumeNativeZlibLoaded();
+ byte[] corrupted = Arrays.copyOf(COMPRESSED_PAYLOAD,
COMPRESSED_PAYLOAD.length);
+ corrupted[0] ^= (byte) 0xff;
+ 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("Invalid gzip stream"));
+ }
+ }
+
+ @Test
+ public void itDecompressDirectFailsWhenOutputBufferTooSmall() 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) {
+ assertTrue(e.getMessage().contains("Output buffer is too small"));
+ }
+ }
+
+ @Test
+ 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.
+ corrupted[corrupted.length - 8] ^= (byte) 0xff;
+ 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("Invalid gzip stream"));
+ }
+ }
+
+ @Test
+ 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.
+ corrupted[corrupted.length - 4] ^= (byte) 0xff;
+ 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("Invalid gzip stream"));
+ }
+ }
+
+ @Test
+ 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.
+ try (GzipByteBuffDecompressor decompressor = new
GzipByteBuffDecompressor(true)) {
+ for (int i = 0; i < 3; i++) {
+ 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)));
+ }
+ }
+ }
+
+ @Test
+ 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.
+ corrupted[corrupted.length - 8] ^= (byte) 0xff;
+ 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("Invalid gzip stream"));
+ }
+
+ // 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)));
+ }
+ }
+
+ @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 GzipHFileDecompressionContext#ALLOW_BYTE_BUFF_DECOMPRESSION_KEY}
config flag.
+ */
+ @Test
+ 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(GzipHFileDecompressionContext.ALLOW_BYTE_BUFF_DECOMPRESSION_KEY,
false);
+
decompressor.reinit(GzipHFileDecompressionContext.fromConfiguration(conf));
+ assertFalse(decompressor.canDecompress(output, input),
+ "Block reader must fall back to stream decompression when the config
flag "
+ + "disables ByteBuff decompression");
+
+
conf.setBoolean(GzipHFileDecompressionContext.ALLOW_BYTE_BUFF_DECOMPRESSION_KEY,
true);
+
decompressor.reinit(GzipHFileDecompressionContext.fromConfiguration(conf));
+ 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
+ .reinit(GzipHFileDecompressionContext.fromConfiguration(new
Configuration(false)));
+ assertTrue(decompressor.canDecompress(output, input));
+ }
+ }
+
+ @Test
+ 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(GzipHFileDecompressionContext.ALLOW_BYTE_BUFF_DECOMPRESSION_KEY,
false);
+
decompressor.reinit(GzipHFileDecompressionContext.fromConfiguration(conf));
+ assertFalse(decompressor.canDecompress(output, input));
+
+ decompressor.reinit(null);
+ assertFalse(decompressor.canDecompress(output, input),
+ "reinit(null) must not reset allowByteBuffDecompression back to the
default");
+ }
+ }
+
+ @Test
+ public void itReinitFailsOnWrongContextType() {
+ Compression.HFileDecompressionContext wrongContext =
+ new Compression.HFileDecompressionContext() {
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public long heapSize() {
+ return 0;
+ }
+ };
+ try (GzipByteBuffDecompressor decompressor = new
GzipByteBuffDecompressor(false)) {
+ decompressor.reinit(wrongContext);
+ fail("Expected an IllegalArgumentException because the context was not a
"
+ + "GzipHFileDecompressionContext");
+ } catch (IllegalArgumentException e) {
+ assertTrue(e.getMessage().contains("GzipHFileDecompressionContext"));
+ }
+ }
+
+ @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"));
+ }
+ }
+
+}
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 00000000000..b7d5e31ef92
--- /dev/null
+++
b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/compress/TestHFileCompressionGzip.java
@@ -0,0 +1,67 @@
+/*
+ * 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.HBaseTestingUtil;
+import org.apache.hadoop.hbase.testclassification.IOTests;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+@Tag(IOTests.TAG)
+@Tag(SmallTests.TAG)
+public class TestHFileCompressionGzip extends HFileTestBase {
+
+ private static Configuration conf;
+
+ @BeforeAll
+ public static void setUpBeforeClass() throws Exception {
+ HFileTestBase.setUpBeforeClass();
+ }
+
+ @BeforeEach
+ 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);
+ }
+
+}
