View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.regionserver.wal;
20  
21  import java.io.DataInput;
22  import java.io.DataOutput;
23  import java.io.IOException;
24  import java.util.ArrayList;
25  import java.util.NavigableMap;
26  import java.util.TreeMap;
27  
28  import org.apache.commons.logging.Log;
29  import org.apache.commons.logging.LogFactory;
30  import org.apache.hadoop.hbase.classification.InterfaceAudience;
31  import org.apache.hadoop.hbase.Cell;
32  import org.apache.hadoop.hbase.CellUtil;
33  import org.apache.hadoop.hbase.HRegionInfo;
34  import org.apache.hadoop.hbase.HBaseInterfaceAudience;
35  import org.apache.hadoop.hbase.KeyValue;
36  import org.apache.hadoop.hbase.KeyValueUtil;
37  import org.apache.hadoop.hbase.codec.Codec;
38  import org.apache.hadoop.hbase.io.HeapSize;
39  import org.apache.hadoop.hbase.protobuf.generated.WALProtos.CompactionDescriptor;
40  import org.apache.hadoop.hbase.util.Bytes;
41  import org.apache.hadoop.hbase.util.ClassSize;
42  import org.apache.hadoop.io.Writable;
43  
44  
45  /**
46   * WALEdit: Used in HBase's transaction log (WAL) to represent
47   * the collection of edits (KeyValue objects) corresponding to a
48   * single transaction. The class implements "Writable" interface
49   * for serializing/deserializing a set of KeyValue items.
50   *
51   * Previously, if a transaction contains 3 edits to c1, c2, c3 for a row R,
52   * the HLog would have three log entries as follows:
53   *
54   *    <logseq1-for-edit1>:<KeyValue-for-edit-c1>
55   *    <logseq2-for-edit2>:<KeyValue-for-edit-c2>
56   *    <logseq3-for-edit3>:<KeyValue-for-edit-c3>
57   *
58   * This presents problems because row level atomicity of transactions
59   * was not guaranteed. If we crash after few of the above appends make
60   * it, then recovery will restore a partial transaction.
61   *
62   * In the new world, all the edits for a given transaction are written
63   * out as a single record, for example:
64   *
65   *   <logseq#-for-entire-txn>:<WALEdit-for-entire-txn>
66   *
67   * where, the WALEdit is serialized as:
68   *   <-1, # of edits, <KeyValue>, <KeyValue>, ... >
69   * For example:
70   *   <-1, 3, <Keyvalue-for-edit-c1>, <KeyValue-for-edit-c2>, <KeyValue-for-edit-c3>>
71   *
72   * The -1 marker is just a special way of being backward compatible with
73   * an old HLog which would have contained a single <KeyValue>.
74   *
75   * The deserializer for WALEdit backward compatibly detects if the record
76   * is an old style KeyValue or the new style WALEdit.
77   *
78   */
79  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.REPLICATION)
80  public class WALEdit implements Writable, HeapSize {
81    public static final Log LOG = LogFactory.getLog(WALEdit.class);
82  
83    // TODO: Get rid of this; see HBASE-8457
84    public static final byte [] METAFAMILY = Bytes.toBytes("METAFAMILY");
85    static final byte [] METAROW = Bytes.toBytes("METAROW");
86    static final byte[] COMPLETE_CACHE_FLUSH = Bytes.toBytes("HBASE::CACHEFLUSH");
87    static final byte[] COMPACTION = Bytes.toBytes("HBASE::COMPACTION");
88    private final int VERSION_2 = -1;
89    private final boolean isReplay;
90  
91    private final ArrayList<KeyValue> kvs = new ArrayList<KeyValue>(1);
92  
93    // Only here for legacy writable deserialization
94    @Deprecated
95    private NavigableMap<byte[], Integer> scopes;
96  
97    private CompressionContext compressionContext;
98  
99    public WALEdit() {
100     this(false);
101   }
102 
103   public WALEdit(boolean isReplay) {
104     this.isReplay = isReplay;
105   }
106 
107   /**
108    * @param f
109    * @return True is <code>f</code> is {@link #METAFAMILY}
110    */
111   public static boolean isMetaEditFamily(final byte [] f) {
112     return Bytes.equals(METAFAMILY, f);
113   }
114 
115   /**
116    * @return True when current WALEdit is created by log replay. Replication skips WALEdits from
117    *         replay.
118    */
119   public boolean isReplay() {
120     return this.isReplay;
121   }
122 
123   public void setCompressionContext(final CompressionContext compressionContext) {
124     this.compressionContext = compressionContext;
125   }
126 
127   /**
128    * Adds a KeyValue to this edit
129    * @param kv
130    * @return this for chained action
131    * @deprecated Use {@link #add(Cell)} instead
132    */
133   @Deprecated
134   public WALEdit add(KeyValue kv) {
135     this.kvs.add(kv);
136     return this;
137   }
138 
139   /**
140    * Adds a Cell to this edit
141    * @param cell
142    * @return this for chained action
143    */
144   public WALEdit add(Cell cell) {
145     return add(KeyValueUtil.ensureKeyValue(cell));
146   }
147 
148   public boolean isEmpty() {
149     return kvs.isEmpty();
150   }
151 
152   public int size() {
153     return kvs.size();
154   }
155 
156   /**
157    * @return The KeyValues associated with this edit
158    * @deprecated Use {@link #getCells()} instead
159    */
160   @Deprecated
161   public ArrayList<KeyValue> getKeyValues() {
162     return kvs;
163   }
164 
165   /**
166    * @return The Cells associated with this edit
167    */
168   public ArrayList<Cell> getCells() {
169     ArrayList<Cell> cells = new ArrayList<Cell>(kvs.size());
170     cells.addAll(kvs);
171     return cells;
172   }
173 
174   public NavigableMap<byte[], Integer> getAndRemoveScopes() {
175     NavigableMap<byte[], Integer> result = scopes;
176     scopes = null;
177     return result;
178   }
179 
180   @Override
181   public void readFields(DataInput in) throws IOException {
182     kvs.clear();
183     if (scopes != null) {
184       scopes.clear();
185     }
186     int versionOrLength = in.readInt();
187     // TODO: Change version when we protobuf.  Also, change way we serialize KV!  Pb it too.
188     if (versionOrLength == VERSION_2) {
189       // this is new style HLog entry containing multiple KeyValues.
190       int numEdits = in.readInt();
191       for (int idx = 0; idx < numEdits; idx++) {
192         if (compressionContext != null) {
193           this.add(KeyValueCompression.readKV(in, compressionContext));
194         } else {
195           this.add(KeyValue.create(in));
196         }
197       }
198       int numFamilies = in.readInt();
199       if (numFamilies > 0) {
200         if (scopes == null) {
201           scopes = new TreeMap<byte[], Integer>(Bytes.BYTES_COMPARATOR);
202         }
203         for (int i = 0; i < numFamilies; i++) {
204           byte[] fam = Bytes.readByteArray(in);
205           int scope = in.readInt();
206           scopes.put(fam, scope);
207         }
208       }
209     } else {
210       // this is an old style HLog entry. The int that we just
211       // read is actually the length of a single KeyValue
212       this.add(KeyValue.create(versionOrLength, in));
213     }
214   }
215 
216   @Override
217   public void write(DataOutput out) throws IOException {
218     LOG.warn("WALEdit is being serialized to writable - only expected in test code");
219     out.writeInt(VERSION_2);
220     out.writeInt(kvs.size());
221     // We interleave the two lists for code simplicity
222     for (KeyValue kv : kvs) {
223       if (compressionContext != null) {
224         KeyValueCompression.writeKV(out, kv, compressionContext);
225       } else{
226         KeyValue.write(kv, out);
227       }
228     }
229     if (scopes == null) {
230       out.writeInt(0);
231     } else {
232       out.writeInt(scopes.size());
233       for (byte[] key : scopes.keySet()) {
234         Bytes.writeByteArray(out, key);
235         out.writeInt(scopes.get(key));
236       }
237     }
238   }
239 
240   /**
241    * Reads WALEdit from cells.
242    * @param cellDecoder Cell decoder.
243    * @param expectedCount Expected cell count.
244    * @return Number of KVs read.
245    */
246   public int readFromCells(Codec.Decoder cellDecoder, int expectedCount) throws IOException {
247     kvs.clear();
248     kvs.ensureCapacity(expectedCount);
249     while (kvs.size() < expectedCount && cellDecoder.advance()) {
250       Cell cell = cellDecoder.current();
251       if (!(cell instanceof KeyValue)) {
252         throw new IOException("WAL edit only supports KVs as cells");
253       }
254       kvs.add((KeyValue)cell);
255     }
256     return kvs.size();
257   }
258 
259   @Override
260   public long heapSize() {
261     long ret = ClassSize.ARRAYLIST;
262     for (KeyValue kv : kvs) {
263       ret += kv.heapSize();
264     }
265     if (scopes != null) {
266       ret += ClassSize.TREEMAP;
267       ret += ClassSize.align(scopes.size() * ClassSize.MAP_ENTRY);
268       // TODO this isn't quite right, need help here
269     }
270     return ret;
271   }
272 
273   @Override
274   public String toString() {
275     StringBuilder sb = new StringBuilder();
276 
277     sb.append("[#edits: " + kvs.size() + " = <");
278     for (KeyValue kv : kvs) {
279       sb.append(kv.toString());
280       sb.append("; ");
281     }
282     if (scopes != null) {
283       sb.append(" scopes: " + scopes.toString());
284     }
285     sb.append(">]");
286     return sb.toString();
287   }
288 
289   /**
290    * Create a compacion WALEdit
291    * @param c
292    * @return A WALEdit that has <code>c</code> serialized as its value
293    */
294   public static WALEdit createCompaction(final HRegionInfo hri, final CompactionDescriptor c) {
295     byte [] pbbytes = c.toByteArray();
296     KeyValue kv = new KeyValue(getRowForRegion(hri), METAFAMILY, COMPACTION, System.currentTimeMillis(), pbbytes);
297     return new WALEdit().add(kv); //replication scope null so that this won't be replicated
298   }
299 
300   private static byte[] getRowForRegion(HRegionInfo hri) {
301     byte[] startKey = hri.getStartKey();
302     if (startKey.length == 0) {
303       // empty row key is not allowed in mutations because it is both the start key and the end key
304       // we return the smallest byte[] that is bigger (in lex comparison) than byte[0].
305       return new byte[] {0};
306     }
307     return startKey;
308   }
309 
310   /**
311    * Deserialized and returns a CompactionDescriptor is the KeyValue contains one.
312    * @param kv the key value
313    * @return deserialized CompactionDescriptor or null.
314    */
315   public static CompactionDescriptor getCompaction(Cell kv) throws IOException {
316     if (CellUtil.matchingColumn(kv, METAFAMILY, COMPACTION)) {
317       return CompactionDescriptor.parseFrom(kv.getValue());
318     }
319     return null;
320   }
321 }
322