View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  
19  package org.apache.hadoop.hbase.regionserver.wal;
20  
21  import java.io.DataInput;
22  import java.io.DataInputStream;
23  import java.io.DataOutput;
24  import java.io.DataOutputStream;
25  import java.io.IOException;
26  
27  import org.apache.hadoop.hbase.KeyValue;
28  import org.apache.hadoop.hbase.codec.BaseDecoder;
29  import org.apache.hadoop.hbase.codec.BaseEncoder;
30  import org.apache.hadoop.hbase.util.Bytes;
31  import org.apache.hadoop.io.WritableUtils;
32  
33  /**
34   * Compression class for {@link KeyValue}s written to the WAL. This is not
35   * synchronized, so synchronization should be handled outside.
36   * 
37   * Class only compresses and uncompresses row keys, family names, and the
38   * qualifier. More may be added depending on use patterns.
39   */
40  class KeyValueCompression {
41    /**
42     * Uncompresses a KeyValue from a DataInput and returns it.
43     * 
44     * @param in the DataInput
45     * @param readContext the compressionContext to use.
46     * @return an uncompressed KeyValue
47     * @throws IOException
48     */
49  
50    public static KeyValue readKV(DataInput in, CompressionContext readContext)
51        throws IOException {
52      int keylength = WritableUtils.readVInt(in);
53      int vlength = WritableUtils.readVInt(in);
54      int length = KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE + keylength + vlength;
55  
56      byte[] backingArray = new byte[length];
57      int pos = 0;
58      pos = Bytes.putInt(backingArray, pos, keylength);
59      pos = Bytes.putInt(backingArray, pos, vlength);
60  
61      // the row
62      int elemLen = Compressor.uncompressIntoArray(backingArray,
63          pos + Bytes.SIZEOF_SHORT, in, readContext.rowDict);
64      checkLength(elemLen, Short.MAX_VALUE);
65      pos = Bytes.putShort(backingArray, pos, (short)elemLen);
66      pos += elemLen;
67  
68      // family
69      elemLen = Compressor.uncompressIntoArray(backingArray,
70          pos + Bytes.SIZEOF_BYTE, in, readContext.familyDict);
71      checkLength(elemLen, Byte.MAX_VALUE);
72      pos = Bytes.putByte(backingArray, pos, (byte)elemLen);
73      pos += elemLen;
74  
75      // qualifier
76      elemLen = Compressor.uncompressIntoArray(backingArray, pos, in,
77          readContext.qualifierDict);
78      pos += elemLen;
79  
80      // the rest
81      in.readFully(backingArray, pos, length - pos);
82  
83      return new KeyValue(backingArray);
84    }
85  
86    private static void checkLength(int len, int max) throws IOException {
87      if (len < 0 || len > max) {
88        throw new IOException(
89            "Invalid length for compresesed portion of keyvalue: " + len);
90      }
91    }
92  
93    /**
94     * Compresses and writes ourKV to out, a DataOutput.
95     * 
96     * @param out the DataOutput
97     * @param keyVal the KV to compress and write
98     * @param writeContext the compressionContext to use.
99     * @throws IOException
100    */
101   public static void writeKV(final DataOutput out, KeyValue keyVal,
102       CompressionContext writeContext) throws IOException {
103     byte[] backingArray = keyVal.getBuffer();
104     int offset = keyVal.getOffset();
105 
106     // we first write the KeyValue infrastructure as VInts.
107     WritableUtils.writeVInt(out, keyVal.getKeyLength());
108     WritableUtils.writeVInt(out, keyVal.getValueLength());
109 
110     // now we write the row key, as the row key is likely to be repeated
111     // We save space only if we attempt to compress elements with duplicates
112     Compressor.writeCompressed(keyVal.getBuffer(), keyVal.getRowOffset(),
113         keyVal.getRowLength(), out, writeContext.rowDict);
114 
115   
116     // now family, if it exists. if it doesn't, we write a 0 length array.
117     Compressor.writeCompressed(keyVal.getBuffer(), keyVal.getFamilyOffset(),
118         keyVal.getFamilyLength(), out, writeContext.familyDict);
119 
120     // qualifier next
121     Compressor.writeCompressed(keyVal.getBuffer(), keyVal.getQualifierOffset(),
122         keyVal.getQualifierLength(), out,
123         writeContext.qualifierDict);
124 
125     // now we write the rest uncompressed
126     int pos = keyVal.getTimestampOffset();
127     int remainingLength = keyVal.getLength() + offset - (pos);
128     out.write(backingArray, pos, remainingLength);
129   }
130 
131   static class CompressedKvEncoder extends BaseEncoder {
132     private final CompressionContext compression;
133 
134     public CompressedKvEncoder(DataOutputStream out, CompressionContext compression) {
135       super(out);
136       this.compression = compression;
137     }
138 
139     @Override
140     public void write(KeyValue kv) throws IOException {
141       KeyValueCompression.writeKV((DataOutputStream) out, kv, compression);
142     }
143   }
144 
145   static class CompressedKvDecoder extends BaseDecoder {
146     private final CompressionContext compression;
147 
148     public CompressedKvDecoder(DataInputStream in, CompressionContext compression) {
149       super(in);
150       this.compression = compression;
151     }
152 
153     @Override
154     protected KeyValue parseCell() throws IOException {
155       return KeyValueCompression.readKV((DataInputStream) in, compression);
156     }
157   }
158 }