001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.io.compress; 019 020import edu.umd.cs.findbugs.annotations.Nullable; 021import java.io.IOException; 022import java.nio.ByteBuffer; 023import org.apache.hadoop.hbase.nio.ByteBuff; 024import org.apache.hadoop.hbase.nio.SingleByteBuff; 025import org.apache.hadoop.io.compress.zlib.ZlibDecompressor; 026import org.apache.yetus.audience.InterfaceAudience; 027 028/** 029 * Glue for ByteBuffDecompressor on top of Hadoop's native 030 * {@link ZlibDecompressor.ZlibDirectDecompressor}. Only direct-to-direct decompression is 031 * supported, which is zero-copy; callers with on-heap buffers fall back to the stream path. 032 */ 033@InterfaceAudience.Private 034public class GzipByteBuffDecompressor implements ByteBuffDecompressor { 035 036 private static final int GZIP_HEADER_LENGTH = 10; 037 private static final int GZIP_TRAILER_LENGTH = 8; 038 039 @Nullable 040 private final ZlibDecompressor.ZlibDirectDecompressor decompressor; 041 042 private boolean allowByteBuffDecompression; 043 044 GzipByteBuffDecompressor(boolean nativeZlibLoaded) { 045 decompressor = nativeZlibLoaded 046 ? new ZlibDecompressor.ZlibDirectDecompressor( 047 ZlibDecompressor.CompressionHeader.AUTODETECT_GZIP_ZLIB, 0) 048 : null; 049 allowByteBuffDecompression = true; 050 } 051 052 @Override 053 public boolean canDecompress(ByteBuff output, ByteBuff input) { 054 if (!allowByteBuffDecompression) { 055 return false; 056 } 057 if (!(output instanceof SingleByteBuff) || !(input instanceof SingleByteBuff)) { 058 return false; 059 } 060 // Only direct-to-direct decompression is supported. 061 return input.nioByteBuffers()[0].isDirect() && output.nioByteBuffers()[0].isDirect() 062 && decompressor != null; 063 } 064 065 @Override 066 public int decompress(ByteBuff output, ByteBuff input, int inputLen) throws IOException { 067 if (!(output instanceof SingleByteBuff) || !(input instanceof SingleByteBuff)) { 068 throw new IllegalStateException( 069 "At least one buffer is not a SingleByteBuff, this is not supported"); 070 } 071 if (inputLen < GZIP_HEADER_LENGTH + GZIP_TRAILER_LENGTH) { 072 throw new IOException("Input of length " + inputLen + " is too short to be a gzip member"); 073 } 074 075 ByteBuffer nioInput = input.nioByteBuffers()[0]; 076 ByteBuffer nioOutput = output.nioByteBuffers()[0]; 077 if (!nioInput.isDirect() || !nioOutput.isDirect() || decompressor == null) { 078 throw new IllegalStateException( 079 "GzipByteBuffDecompressor only supports direct-to-direct decompression with native zlib " 080 + "loaded, this should never happen since canDecompress() would have returned false"); 081 } 082 return decompressOffHeap(nioInput, nioOutput, inputLen); 083 } 084 085 private int decompressOffHeap(ByteBuffer nioInput, ByteBuffer nioOutput, int inputLen) 086 throws IOException { 087 int inputStart = nioInput.position(); 088 int outputStart = nioOutput.position(); 089 090 ByteBuffer gzipMember = nioInput.duplicate(); 091 gzipMember.limit(inputStart + inputLen); 092 093 decompressor.reset(); 094 try { 095 decompressor.decompress(gzipMember, nioOutput); 096 } catch (IOException e) { 097 throw new IOException("Invalid gzip stream: " + e.getMessage(), e); 098 } 099 if (!decompressor.finished()) { 100 if (!nioOutput.hasRemaining()) { 101 throw new IOException("Output buffer is too small for the decompressed gzip stream"); 102 } 103 throw new IOException("Unexpected end of gzip stream"); 104 } 105 if (gzipMember.hasRemaining()) { 106 throw new IOException("Unexpected trailing bytes after decompressing gzip stream"); 107 } 108 109 nioInput.position(inputStart + inputLen); 110 111 return nioOutput.position() - outputStart; 112 } 113 114 @Override 115 public void reinit(@Nullable Compression.HFileDecompressionContext newHFileDecompressionContext) { 116 if (newHFileDecompressionContext == null) { 117 return; 118 } 119 if (!(newHFileDecompressionContext instanceof GzipHFileDecompressionContext)) { 120 throw new IllegalArgumentException( 121 "GzipByteBuffDecompressor#reinit() was given an HFileDecompressionContext that was not " 122 + "a GzipHFileDecompressionContext, this should never happen"); 123 } 124 GzipHFileDecompressionContext gzipContext = 125 (GzipHFileDecompressionContext) newHFileDecompressionContext; 126 allowByteBuffDecompression = gzipContext.isAllowByteBuffDecompression(); 127 } 128 129 @Override 130 public void close() { 131 if (decompressor != null) { 132 decompressor.end(); 133 } 134 } 135 136}