1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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
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
111
112
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
138
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
159
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
188
189
190
191
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
212 if (versionOrLength == VERSION_2) {
213
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
235
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
246 for (Cell cell : cells) {
247
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
268
269
270
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
291 }
292 return ret;
293 }
294
295 @Override
296 public String toString() {
297 StringBuilder sb = new StringBuilder();
298
299 sb.append("[#edits: " + cells.size() + " = <");
300 for (Cell cell : cells) {
301 sb.append(cell);
302 sb.append("; ");
303 }
304 if (scopes != null) {
305 sb.append(" scopes: " + scopes.toString());
306 }
307 sb.append(">]");
308 return sb.toString();
309 }
310
311 public static WALEdit createFlushWALEdit(HRegionInfo hri, FlushDescriptor f) {
312 KeyValue kv = new KeyValue(getRowForRegion(hri), METAFAMILY, FLUSH,
313 EnvironmentEdgeManager.currentTime(), f.toByteArray());
314 return new WALEdit().add(kv);
315 }
316
317 public static FlushDescriptor getFlushDescriptor(Cell cell) throws IOException {
318 if (CellUtil.matchingColumn(cell, METAFAMILY, FLUSH)) {
319 return FlushDescriptor.parseFrom(cell.getValue());
320 }
321 return null;
322 }
323
324 public static WALEdit createRegionEventWALEdit(HRegionInfo hri,
325 RegionEventDescriptor regionEventDesc) {
326 KeyValue kv = new KeyValue(getRowForRegion(hri), METAFAMILY, REGION_EVENT,
327 EnvironmentEdgeManager.currentTime(), regionEventDesc.toByteArray());
328 return new WALEdit().add(kv);
329 }
330
331 public static RegionEventDescriptor getRegionEventDescriptor(Cell cell) throws IOException {
332 if (CellUtil.matchingColumn(cell, METAFAMILY, REGION_EVENT)) {
333 return RegionEventDescriptor.parseFrom(cell.getValue());
334 }
335 return null;
336 }
337
338
339
340
341
342
343 public static WALEdit createCompaction(final HRegionInfo hri, final CompactionDescriptor c) {
344 byte [] pbbytes = c.toByteArray();
345 KeyValue kv = new KeyValue(getRowForRegion(hri), METAFAMILY, COMPACTION,
346 EnvironmentEdgeManager.currentTime(), pbbytes);
347 return new WALEdit().add(kv);
348 }
349
350 public static byte[] getRowForRegion(HRegionInfo hri) {
351 byte[] startKey = hri.getStartKey();
352 if (startKey.length == 0) {
353
354
355 return new byte[] {0};
356 }
357 return startKey;
358 }
359
360
361
362
363
364
365 public static CompactionDescriptor getCompaction(Cell kv) throws IOException {
366 if (isCompactionMarker(kv)) {
367 return CompactionDescriptor.parseFrom(CellUtil.cloneValue(kv));
368 }
369 return null;
370 }
371
372
373
374
375
376
377 public static boolean isCompactionMarker(Cell cell) {
378 return CellUtil.matchingColumn(cell, METAFAMILY, COMPACTION);
379 }
380
381
382
383
384
385
386
387
388 public static WALEdit createBulkLoadEvent(HRegionInfo hri,
389 WALProtos.BulkLoadDescriptor bulkLoadDescriptor) {
390 KeyValue kv = new KeyValue(getRowForRegion(hri),
391 METAFAMILY,
392 BULK_LOAD,
393 EnvironmentEdgeManager.currentTime(),
394 bulkLoadDescriptor.toByteArray());
395 return new WALEdit().add(kv);
396 }
397
398
399
400
401
402
403 public static WALProtos.BulkLoadDescriptor getBulkLoadDescriptor(Cell cell) throws IOException {
404 if (CellUtil.matchingColumn(cell, METAFAMILY, BULK_LOAD)) {
405 return WALProtos.BulkLoadDescriptor.parseFrom(cell.getValue());
406 }
407 return null;
408 }
409 }