View Javadoc

1   /*
2    * Licensed to the Apache Software Foundation (ASF) under one or more
3    * contributor license agreements. See the NOTICE file distributed with this
4    * work for additional information regarding copyright ownership. The ASF
5    * licenses this file to you under the Apache License, Version 2.0 (the
6    * "License"); you may not use this file except in compliance with the License.
7    * You may obtain a copy of the License at
8    *
9    * http://www.apache.org/licenses/LICENSE-2.0
10   *
11   * Unless required by applicable law or agreed to in writing, software
12   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14   * License for the specific language governing permissions and limitations
15   * under the License.
16   */
17  package org.apache.hadoop.hbase.io.hfile;
18  
19  import java.io.ByteArrayOutputStream;
20  import java.io.IOException;
21  import java.io.OutputStream;
22  import java.util.Arrays;
23  import java.util.zip.GZIPOutputStream;
24  
25  import org.apache.commons.logging.Log;
26  import org.apache.commons.logging.LogFactory;
27  import org.apache.hadoop.io.compress.CompressionOutputStream;
28  import org.apache.hadoop.io.compress.CompressorStream;
29  import org.apache.hadoop.io.compress.GzipCodec;
30  import org.apache.hadoop.io.compress.zlib.ZlibFactory;
31  
32  /**
33   * Fixes an inefficiency in Hadoop's Gzip codec, allowing to reuse compression
34   * streams.
35   */
36  public class ReusableStreamGzipCodec extends GzipCodec {
37  
38    private static final Log LOG = LogFactory.getLog(Compression.class);
39  
40    /**
41     * A bridge that wraps around a DeflaterOutputStream to make it a
42     * CompressionOutputStream.
43     */
44    protected static class ReusableGzipOutputStream extends CompressorStream {
45  
46      private static final int GZIP_HEADER_LENGTH = 10;
47  
48      /**
49       * Fixed ten-byte gzip header. See {@link GZIPOutputStream}'s source for
50       * details.
51       */
52      private static final byte[] GZIP_HEADER;
53  
54      static {
55        // Capture the fixed ten-byte header hard-coded in GZIPOutputStream.
56        ByteArrayOutputStream baos = new ByteArrayOutputStream();
57        byte[] header = null;
58        GZIPOutputStream gzipStream = null;
59        try {
60          gzipStream  = new GZIPOutputStream(baos);
61          gzipStream.finish();
62          header = Arrays.copyOfRange(baos.toByteArray(), 0, GZIP_HEADER_LENGTH);
63        } catch (IOException e) {
64          throw new RuntimeException("Could not create gzip stream", e);
65        } finally {
66          if (gzipStream != null) {
67            try {
68              gzipStream.close();
69            } catch (IOException e) {
70              LOG.error(e);
71            }
72          }
73        }
74        GZIP_HEADER = header;
75      }
76  
77      private static class ResetableGZIPOutputStream extends GZIPOutputStream {
78        public ResetableGZIPOutputStream(OutputStream out) throws IOException {
79          super(out);
80        }
81  
82        public void resetState() throws IOException {
83          def.reset();
84          crc.reset();
85          out.write(GZIP_HEADER);
86        }
87      }
88  
89      public ReusableGzipOutputStream(OutputStream out) throws IOException {
90        super(new ResetableGZIPOutputStream(out));
91      }
92  
93      @Override
94      public void close() throws IOException {
95        out.close();
96      }
97  
98      @Override
99      public void flush() throws IOException {
100       out.flush();
101     }
102 
103     @Override
104     public void write(int b) throws IOException {
105       out.write(b);
106     }
107 
108     @Override
109     public void write(byte[] data, int offset, int length) throws IOException {
110       out.write(data, offset, length);
111     }
112 
113     @Override
114     public void finish() throws IOException {
115       ((GZIPOutputStream) out).finish();
116     }
117 
118     @Override
119     public void resetState() throws IOException {
120       ((ResetableGZIPOutputStream) out).resetState();
121     }
122   }
123 
124   @Override
125   public CompressionOutputStream createOutputStream(OutputStream out)
126       throws IOException {
127     if (ZlibFactory.isNativeZlibLoaded(getConf())) {
128       return super.createOutputStream(out);
129     }
130     return new ReusableGzipOutputStream(out);
131   }
132 
133 }