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}