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;
40  import org.apache.hadoop.hbase.protobuf.generated.WALProtos.CompactionDescriptor;
41  import org.apache.hadoop.hbase.protobuf.generated.WALProtos.FlushDescriptor;
42  import org.apache.hadoop.hbase.protobuf.generated.WALProtos.RegionEventDescriptor;
43  import org.apache.hadoop.hbase.util.Bytes;
44  import org.apache.hadoop.hbase.util.ClassSize;
45  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
46  import org.apache.hadoop.io.Writable;
47  
48  import com.google.common.annotations.VisibleForTesting;
49  
50  
51  /**
52   * WALEdit: Used in HBase's transaction log (WAL) to represent
53   * the collection of edits (KeyValue objects) corresponding to a
54   * single transaction. The class implements "Writable" interface
55   * for serializing/deserializing a set of KeyValue items.
56   *
57   * Previously, if a transaction contains 3 edits to c1, c2, c3 for a row R,
58   * the WAL would have three log entries as follows:
59   *
60   *    <logseq1-for-edit1>:<eyValue-for-edit-c1>
61   *    <logseq2-for-edit2>:<KeyValue-for-edit-c2>
62   *    <logseq3-for-edit3>:<KeyValue-for-edit-c3>
63   *
64   * This presents problems because row level atomicity of transactions
65   * was not guaranteed. If we crash after few of the above appends make
66   * it, then recovery will restore a partial transaction.
67   *
68   * In the new world, all the edits for a given transaction are written
69   * out as a single record, for example:
70   *
71   *   <logseq#-for-entire-txn>:<WALEdit-for-entire-txn>
72   *
73   * where, the WALEdit is serialized as:
74   *   <-1, # of edits, <KeyValue>, <KeyValue>, ... >
75   * For example:
76   *   <-1, 3, <KV-for-edit-c1>, <KV-for-edit-c2>, <KV-for-edit-c3>>
77   *
78   * The -1 marker is just a special way of being backward compatible with
79   * an old WAL which would have contained a single <KeyValue>.
80   *
81   * The deserializer for WALEdit backward compatibly detects if the record
82   * is an old style KeyValue or the new style WALEdit.
83   *
84   */
85  @InterfaceAudience.LimitedPrivate({ HBaseInterfaceAudience.REPLICATION,
86      HBaseInterfaceAudience.COPROC })
87  public class WALEdit implements Writable, HeapSize {
88    private static final Log LOG = LogFactory.getLog(WALEdit.class);
89  
90    // TODO: Get rid of this; see HBASE-8457
91    public static final byte [] METAFAMILY = Bytes.toBytes("METAFAMILY");
92    @VisibleForTesting
93    public static final byte [] METAROW = Bytes.toBytes("METAROW");
94    @VisibleForTesting
95    public static final byte[] COMPACTION = Bytes.toBytes("HBASE::COMPACTION");
96    @VisibleForTesting
97    public static final byte [] FLUSH = Bytes.toBytes("HBASE::FLUSH");
98    @VisibleForTesting
99    public static final byte [] REGION_EVENT = Bytes.toBytes("HBASE::REGION_EVENT");
100   @VisibleForTesting
101   public static final byte [] BULK_LOAD = Bytes.toBytes("HBASE::BULK_LOAD");
102 
103   private final int VERSION_2 = -1;
104   private final boolean isReplay;
105 
106   private ArrayList<Cell> cells = null;
107 
108   public static final WALEdit EMPTY_WALEDIT = new WALEdit();
109 
110   // Only here for legacy writable deserialization
111   /**
112    * @deprecated Legacy
113    */
114   @Deprecated
115   private NavigableMap<byte[], Integer> scopes;
116 
117   private CompressionContext compressionContext;
118 
119   public WALEdit() {
120     this(false);
121   }
122 
123   public WALEdit(boolean isReplay) {
124     this(1, isReplay);
125   }
126 
127   public WALEdit(int cellCount) {
128     this(cellCount, false);
129   }
130 
131   public WALEdit(int cellCount, boolean isReplay) {
132     this.isReplay = isReplay;
133     cells = new ArrayList<Cell>(cellCount);
134   }
135 
136   /**
137    * @param f
138    * @return True is <code>f</code> is {@link #METAFAMILY}
139    */
140   public static boolean isMetaEditFamily(final byte [] f) {
141     return Bytes.equals(METAFAMILY, f);
142   }
143 
144   public static boolean isMetaEditFamily(Cell cell) {
145     return CellUtil.matchingFamily(cell, METAFAMILY);
146   }
147 
148   public boolean isMetaEdit() {
149     for (Cell cell: cells) {
150       if (!isMetaEditFamily(cell)) {
151         return false;
152       }
153     }
154     return true;
155   }
156 
157   /**
158    * @return True when current WALEdit is created by log replay. Replication skips WALEdits from
159    *         replay.
160    */
161   public boolean isReplay() {
162     return this.isReplay;
163   }
164 
165   public void setCompressionContext(final CompressionContext compressionContext) {
166     this.compressionContext = compressionContext;
167   }
168 
169   public WALEdit add(Cell cell) {
170     this.cells.add(cell);
171     return this;
172   }
173 
174   public boolean isEmpty() {
175     return cells.isEmpty();
176   }
177 
178   public int size() {
179     return cells.size();
180   }
181 
182   public ArrayList<Cell> getCells() {
183     return cells;
184   }
185 
186   /**
187    * This is not thread safe.
188    * This will change the WALEdit and shouldn't be used unless you are sure that nothing
189    * else depends on the contents being immutable.
190    *
191    * @param cells the list of cells that this WALEdit now contains.
192    */
193   @InterfaceAudience.Private
194   public void setCells(ArrayList<Cell> cells) {
195     this.cells = cells;
196   }
197 
198   public NavigableMap<byte[], Integer> getAndRemoveScopes() {
199     NavigableMap<byte[], Integer> result = scopes;
200     scopes = null;
201     return result;
202   }
203 
204   @Override
205   public void readFields(DataInput in) throws IOException {
206     cells.clear();
207     if (scopes != null) {
208       scopes.clear();
209     }
210     int versionOrLength = in.readInt();
211     // TODO: Change version when we protobuf.  Also, change way we serialize KV!  Pb it too.
212     if (versionOrLength == VERSION_2) {
213       // this is new style WAL entry containing multiple KeyValues.
214       int numEdits = in.readInt();
215       for (int idx = 0; idx < numEdits; idx++) {
216         if (compressionContext != null) {
217           this.add(KeyValueCompression.readKV(in, compressionContext));
218         } else {
219           this.add(KeyValue.create(in));
220         }
221       }
222       int numFamilies = in.readInt();
223       if (numFamilies > 0) {
224         if (scopes == null) {
225           scopes = new TreeMap<byte[], Integer>(Bytes.BYTES_COMPARATOR);
226         }
227         for (int i = 0; i < numFamilies; i++) {
228           byte[] fam = Bytes.readByteArray(in);
229           int scope = in.readInt();
230           scopes.put(fam, scope);
231         }
232       }
233     } else {
234       // this is an old style WAL entry. The int that we just
235       // read is actually the length of a single KeyValue
236       this.add(KeyValue.create(versionOrLength, in));
237     }
238   }
239 
240   @Override
241   public void write(DataOutput out) throws IOException {
242     LOG.warn("WALEdit is being serialized to writable - only expected in test code");
243     out.writeInt(VERSION_2);
244     out.writeInt(cells.size());
245     // We interleave the two lists for code simplicity
246     for (Cell cell : cells) {
247       // This is not used in any of the core code flows so it is just fine to convert to KV
248       KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
249       if (compressionContext != null) {
250         KeyValueCompression.writeKV(out, kv, compressionContext);
251       } else{
252         KeyValue.write(kv, out);
253       }
254     }
255     if (scopes == null) {
256       out.writeInt(0);
257     } else {
258       out.writeInt(scopes.size());
259       for (byte[] key : scopes.keySet()) {
260         Bytes.writeByteArray(out, key);
261         out.writeInt(scopes.get(key));
262       }
263     }
264   }
265 
266   /**
267    * Reads WALEdit from cells.
268    * @param cellDecoder Cell decoder.
269    * @param expectedCount Expected cell count.
270    * @return Number of KVs read.
271    */
272   public int readFromCells(Codec.Decoder cellDecoder, int expectedCount) throws IOException {
273     cells.clear();
274     cells.ensureCapacity(expectedCount);
275     while (cells.size() < expectedCount && cellDecoder.advance()) {
276       cells.add(cellDecoder.current());
277     }
278     return cells.size();
279   }
280 
281   @Override
282   public long heapSize() {
283     long ret = ClassSize.ARRAYLIST;
284     for (Cell cell : cells) {
285       ret += CellUtil.estimatedHeapSizeOf(cell);
286     }
287     if (scopes != null) {
288       ret += ClassSize.TREEMAP;
289       ret += ClassSize.align(scopes.size() * ClassSize.MAP_ENTRY);
290       // TODO this isn't quite right, need help here
291     }
292     return ret;
293   }
294 
295   public long estimatedSerializedSizeOf() {
296     long ret = 0;
297     for (Cell cell: cells) {
298       ret += CellUtil.estimatedSerializedSizeOf(cell);
299     }
300     return ret;
301   }
302 
303   @Override
304   public String toString() {
305     StringBuilder sb = new StringBuilder();
306 
307     sb.append("[#edits: " + cells.size() + " = <");
308     for (Cell cell : cells) {
309       sb.append(cell);
310       sb.append("; ");
311     }
312     if (scopes != null) {
313       sb.append(" scopes: " + scopes.toString());
314     }
315     sb.append(">]");
316     return sb.toString();
317   }
318 
319   public static WALEdit createFlushWALEdit(HRegionInfo hri, FlushDescriptor f) {
320     KeyValue kv = new KeyValue(getRowForRegion(hri), METAFAMILY, FLUSH,
321       EnvironmentEdgeManager.currentTime(), f.toByteArray());
322     return new WALEdit().add(kv);
323   }
324 
325   public static FlushDescriptor getFlushDescriptor(Cell cell) throws IOException {
326     if (CellUtil.matchingColumn(cell, METAFAMILY, FLUSH)) {
327       return FlushDescriptor.parseFrom(cell.getValue());
328     }
329     return null;
330   }
331 
332   public static WALEdit createRegionEventWALEdit(HRegionInfo hri,
333       RegionEventDescriptor regionEventDesc) {
334     KeyValue kv = new KeyValue(getRowForRegion(hri), METAFAMILY, REGION_EVENT,
335       EnvironmentEdgeManager.currentTime(), regionEventDesc.toByteArray());
336     return new WALEdit().add(kv);
337   }
338 
339   public static RegionEventDescriptor getRegionEventDescriptor(Cell cell) throws IOException {
340     if (CellUtil.matchingColumn(cell, METAFAMILY, REGION_EVENT)) {
341       return RegionEventDescriptor.parseFrom(cell.getValue());
342     }
343     return null;
344   }
345 
346   /**
347    * Create a compaction WALEdit
348    * @param c
349    * @return A WALEdit that has <code>c</code> serialized as its value
350    */
351   public static WALEdit createCompaction(final HRegionInfo hri, final CompactionDescriptor c) {
352     byte [] pbbytes = c.toByteArray();
353     KeyValue kv = new KeyValue(getRowForRegion(hri), METAFAMILY, COMPACTION,
354       EnvironmentEdgeManager.currentTime(), pbbytes);
355     return new WALEdit().add(kv); //replication scope null so that this won't be replicated
356   }
357 
358   public static byte[] getRowForRegion(HRegionInfo hri) {
359     byte[] startKey = hri.getStartKey();
360     if (startKey.length == 0) {
361       // empty row key is not allowed in mutations because it is both the start key and the end key
362       // we return the smallest byte[] that is bigger (in lex comparison) than byte[0].
363       return new byte[] {0};
364     }
365     return startKey;
366   }
367 
368   /**
369    * Deserialized and returns a CompactionDescriptor is the KeyValue contains one.
370    * @param kv the key value
371    * @return deserialized CompactionDescriptor or null.
372    */
373   public static CompactionDescriptor getCompaction(Cell kv) throws IOException {
374     if (isCompactionMarker(kv)) {
375       return CompactionDescriptor.parseFrom(CellUtil.cloneValue(kv));
376     }
377     return null;
378   }
379 
380   /**
381    * Returns true if the given cell is a serialized {@link CompactionDescriptor}
382    *
383    * @see #getCompaction(Cell)
384    */
385   public static boolean isCompactionMarker(Cell cell) {
386     return CellUtil.matchingColumn(cell, METAFAMILY, COMPACTION);
387   }
388 
389   /**
390    * Create a bulk loader WALEdit
391    *
392    * @param hri                The HRegionInfo for the region in which we are bulk loading
393    * @param bulkLoadDescriptor The descriptor for the Bulk Loader
394    * @return The WALEdit for the BulkLoad
395    */
396   public static WALEdit createBulkLoadEvent(HRegionInfo hri,
397                                             WALProtos.BulkLoadDescriptor bulkLoadDescriptor) {
398     KeyValue kv = new KeyValue(getRowForRegion(hri),
399         METAFAMILY,
400         BULK_LOAD,
401         EnvironmentEdgeManager.currentTime(),
402         bulkLoadDescriptor.toByteArray());
403     return new WALEdit().add(kv);
404   }
405 
406   /**
407    * Deserialized and returns a BulkLoadDescriptor from the passed in Cell
408    * @param cell the key value
409    * @return deserialized BulkLoadDescriptor or null.
410    */
411   public static WALProtos.BulkLoadDescriptor getBulkLoadDescriptor(Cell cell) throws IOException {
412     if (CellUtil.matchingColumn(cell, METAFAMILY, BULK_LOAD)) {
413       return WALProtos.BulkLoadDescriptor.parseFrom(cell.getValue());
414     }
415     return null;
416   }
417 }