1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.wal;
20
21 import java.io.IOException;
22 import java.io.InterruptedIOException;
23 import java.util.ArrayList;
24 import java.util.Collections;
25 import java.util.HashMap;
26 import java.util.Iterator;
27 import java.util.List;
28 import java.util.Map;
29 import java.util.NavigableMap;
30 import java.util.TreeMap;
31 import java.util.UUID;
32 import java.util.concurrent.CountDownLatch;
33 import java.util.concurrent.TimeUnit;
34
35 import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl;
36 import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl.WriteEntry;
37 import org.apache.hadoop.hbase.util.ByteStringer;
38 import org.apache.commons.logging.Log;
39 import org.apache.commons.logging.LogFactory;
40 import org.apache.hadoop.hbase.classification.InterfaceAudience;
41 import org.apache.hadoop.hbase.exceptions.TimeoutIOException;
42 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
43 import org.apache.hadoop.hbase.HConstants;
44 import org.apache.hadoop.hbase.TableName;
45 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
46 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.FamilyScope;
47 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.ScopeType;
48 import org.apache.hadoop.hbase.regionserver.SequenceId;
49 import org.apache.hadoop.hbase.util.Bytes;
50 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
51
52 import com.google.common.annotations.VisibleForTesting;
53 import com.google.protobuf.ByteString;
54
55
56
57
58 import org.apache.hadoop.hbase.regionserver.wal.CompressionContext;
59 import org.apache.hadoop.hbase.regionserver.wal.WALCellCodec;
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.REPLICATION)
80 public class WALKey implements SequenceId, Comparable<WALKey> {
81 private static final Log LOG = LogFactory.getLog(WALKey.class);
82
83 @InterfaceAudience.Private
84 public MultiVersionConcurrencyControl getMvcc() {
85 return mvcc;
86 }
87
88
89
90
91
92
93
94
95
96 @InterfaceAudience.Private
97 public MultiVersionConcurrencyControl.WriteEntry getWriteEntry() throws InterruptedIOException {
98 assert this.writeEntry != null;
99 return this.writeEntry;
100 }
101
102 @InterfaceAudience.Private
103 public void setWriteEntry(MultiVersionConcurrencyControl.WriteEntry writeEntry) {
104 assert this.writeEntry == null;
105 this.writeEntry = writeEntry;
106
107 this.logSeqNum = writeEntry.getWriteNumber();
108 }
109
110
111
112
113 @InterfaceAudience.Private
114 protected enum Version {
115 UNVERSIONED(0),
116
117 INITIAL(-1),
118
119
120 COMPRESSED(-2);
121
122 public final int code;
123 static final Version[] byCode;
124 static {
125 byCode = Version.values();
126 for (int i = 0; i < byCode.length; i++) {
127 if (byCode[i].code != -1 * i) {
128 throw new AssertionError("Values in this enum should be descending by one");
129 }
130 }
131 }
132
133 Version(int code) {
134 this.code = code;
135 }
136
137 public boolean atLeast(Version other) {
138 return code <= other.code;
139 }
140
141 public static Version fromCode(int code) {
142 return byCode[code * -1];
143 }
144 }
145
146
147
148
149
150 private static final String PREFIX_CLUSTER_KEY = ".";
151
152
153
154 @InterfaceAudience.Private
155 protected static final Version VERSION = Version.COMPRESSED;
156
157
158 public static final long NO_SEQUENCE_ID = -1;
159
160
161
162 @InterfaceAudience.Private
163 protected byte [] encodedRegionName;
164
165 @InterfaceAudience.Private
166 protected TableName tablename;
167
168 @InterfaceAudience.Private
169 protected long logSeqNum;
170 private long origLogSeqNum = 0;
171
172
173 @InterfaceAudience.Private
174 protected long writeTime;
175
176
177
178 @InterfaceAudience.Private
179 protected List<UUID> clusterIds;
180
181 private NavigableMap<byte[], Integer> scopes;
182
183 private long nonceGroup = HConstants.NO_NONCE;
184 private long nonce = HConstants.NO_NONCE;
185 private MultiVersionConcurrencyControl mvcc;
186 private MultiVersionConcurrencyControl.WriteEntry writeEntry;
187 public static final List<UUID> EMPTY_UUIDS = Collections.unmodifiableList(new ArrayList<UUID>());
188
189
190 @InterfaceAudience.Private
191 protected CompressionContext compressionContext;
192
193 public WALKey() {
194 init(null, null, 0L, HConstants.LATEST_TIMESTAMP,
195 new ArrayList<UUID>(), HConstants.NO_NONCE, HConstants.NO_NONCE, null);
196 }
197
198 @VisibleForTesting
199 public WALKey(final byte[] encodedRegionName, final TableName tablename,
200 long logSeqNum,
201 final long now, UUID clusterId) {
202 List<UUID> clusterIds = new ArrayList<UUID>();
203 clusterIds.add(clusterId);
204 init(encodedRegionName, tablename, logSeqNum, now, clusterIds,
205 HConstants.NO_NONCE, HConstants.NO_NONCE, null);
206 }
207
208 public WALKey(final byte[] encodedRegionName, final TableName tablename) {
209 this(encodedRegionName, tablename, System.currentTimeMillis());
210 }
211
212 public WALKey(final byte[] encodedRegionName, final TableName tablename, final long now) {
213 init(encodedRegionName,
214 tablename,
215 NO_SEQUENCE_ID,
216 now,
217 EMPTY_UUIDS,
218 HConstants.NO_NONCE,
219 HConstants.NO_NONCE,
220 null);
221 }
222
223 public WALKey(final byte[] encodedRegionName,
224 final TableName tablename,
225 final long now,
226 MultiVersionConcurrencyControl mvcc) {
227 init(encodedRegionName,
228 tablename,
229 NO_SEQUENCE_ID,
230 now,
231 EMPTY_UUIDS,
232 HConstants.NO_NONCE,
233 HConstants.NO_NONCE,
234 mvcc);
235 }
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250 public WALKey(final byte[] encodedRegionName,
251 final TableName tablename,
252 long logSeqNum,
253 final long now,
254 List<UUID> clusterIds,
255 long nonceGroup,
256 long nonce,
257 MultiVersionConcurrencyControl mvcc) {
258 init(encodedRegionName, tablename, logSeqNum, now, clusterIds, nonceGroup, nonce, mvcc);
259 }
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275 public WALKey(final byte[] encodedRegionName, final TableName tablename,
276 final long now, List<UUID> clusterIds, long nonceGroup,
277 final long nonce, final MultiVersionConcurrencyControl mvcc) {
278 init(encodedRegionName, tablename, NO_SEQUENCE_ID, now, clusterIds, nonceGroup, nonce, mvcc);
279 }
280
281
282
283
284
285
286
287
288
289
290
291
292
293 public WALKey(final byte[] encodedRegionName,
294 final TableName tablename,
295 long logSeqNum,
296 long nonceGroup,
297 long nonce,
298 final MultiVersionConcurrencyControl mvcc) {
299 init(encodedRegionName,
300 tablename,
301 logSeqNum,
302 EnvironmentEdgeManager.currentTime(),
303 EMPTY_UUIDS,
304 nonceGroup,
305 nonce,
306 mvcc);
307 }
308
309 @InterfaceAudience.Private
310 protected void init(final byte[] encodedRegionName,
311 final TableName tablename,
312 long logSeqNum,
313 final long now,
314 List<UUID> clusterIds,
315 long nonceGroup,
316 long nonce,
317 MultiVersionConcurrencyControl mvcc) {
318 this.logSeqNum = logSeqNum;
319 this.writeTime = now;
320 this.clusterIds = clusterIds;
321 this.encodedRegionName = encodedRegionName;
322 this.tablename = tablename;
323 this.nonceGroup = nonceGroup;
324 this.nonce = nonce;
325 this.mvcc = mvcc;
326 }
327
328
329
330
331 public void setCompressionContext(CompressionContext compressionContext) {
332 this.compressionContext = compressionContext;
333 }
334
335
336 public byte [] getEncodedRegionName() {
337 return encodedRegionName;
338 }
339
340
341 public TableName getTablename() {
342 return tablename;
343 }
344
345
346 public long getLogSeqNum() {
347 return this.logSeqNum;
348 }
349
350
351
352
353
354 public void setOrigLogSeqNum(final long seqId) {
355 this.origLogSeqNum = seqId;
356 }
357
358
359
360
361
362 public long getOrigLogSeqNum() {
363 return this.origLogSeqNum;
364 }
365
366
367
368
369
370
371 @Override
372 public long getSequenceId() throws IOException {
373 return this.logSeqNum;
374 }
375
376
377
378
379 public long getWriteTime() {
380 return this.writeTime;
381 }
382
383 public NavigableMap<byte[], Integer> getScopes() {
384 return scopes;
385 }
386
387
388 public long getNonceGroup() {
389 return nonceGroup;
390 }
391
392
393 public long getNonce() {
394 return nonce;
395 }
396
397 public void setScopes(NavigableMap<byte[], Integer> scopes) {
398 this.scopes = scopes;
399 }
400
401 public void readOlderScopes(NavigableMap<byte[], Integer> scopes) {
402 if (scopes != null) {
403 Iterator<Map.Entry<byte[], Integer>> iterator = scopes.entrySet()
404 .iterator();
405 while (iterator.hasNext()) {
406 Map.Entry<byte[], Integer> scope = iterator.next();
407 String key = Bytes.toString(scope.getKey());
408 if (key.startsWith(PREFIX_CLUSTER_KEY)) {
409 addClusterId(UUID.fromString(key.substring(PREFIX_CLUSTER_KEY
410 .length())));
411 iterator.remove();
412 }
413 }
414 if (scopes.size() > 0) {
415 this.scopes = scopes;
416 }
417 }
418 }
419
420
421
422
423 public void addClusterId(UUID clusterId) {
424 if (!clusterIds.contains(clusterId)) {
425 clusterIds.add(clusterId);
426 }
427 }
428
429
430
431
432 public List<UUID> getClusterIds() {
433 return clusterIds;
434 }
435
436
437
438
439
440 public UUID getOriginatingClusterId(){
441 return clusterIds.isEmpty() ? HConstants.DEFAULT_CLUSTER_ID : clusterIds.get(0);
442 }
443
444 @Override
445 public String toString() {
446 return tablename + "/" + Bytes.toString(encodedRegionName) + "/" +
447 logSeqNum;
448 }
449
450
451
452
453
454
455
456
457 public Map<String, Object> toStringMap() {
458 Map<String, Object> stringMap = new HashMap<String, Object>();
459 stringMap.put("table", tablename);
460 stringMap.put("region", Bytes.toStringBinary(encodedRegionName));
461 stringMap.put("sequence", logSeqNum);
462 return stringMap;
463 }
464
465 @Override
466 public boolean equals(Object obj) {
467 if (this == obj) {
468 return true;
469 }
470 if (obj == null || getClass() != obj.getClass()) {
471 return false;
472 }
473 return compareTo((WALKey)obj) == 0;
474 }
475
476 @Override
477 public int hashCode() {
478 int result = Bytes.hashCode(this.encodedRegionName);
479 result = (int) (result ^ this.logSeqNum);
480 result = (int) (result ^ this.writeTime);
481 return result;
482 }
483
484 @Override
485 public int compareTo(WALKey o) {
486 int result = Bytes.compareTo(this.encodedRegionName, o.encodedRegionName);
487 if (result == 0) {
488 if (this.logSeqNum < o.logSeqNum) {
489 result = -1;
490 } else if (this.logSeqNum > o.logSeqNum) {
491 result = 1;
492 }
493 if (result == 0) {
494 if (this.writeTime < o.writeTime) {
495 result = -1;
496 } else if (this.writeTime > o.writeTime) {
497 return 1;
498 }
499 }
500 }
501
502 return result;
503 }
504
505
506
507
508
509
510
511 void internTableName(TableName tablename) {
512
513
514 assert tablename.equals(this.tablename);
515 this.tablename = tablename;
516 }
517
518
519
520
521
522
523
524 void internEncodedRegionName(byte []encodedRegionName) {
525
526
527 assert Bytes.equals(this.encodedRegionName, encodedRegionName);
528 this.encodedRegionName = encodedRegionName;
529 }
530
531 public org.apache.hadoop.hbase.protobuf.generated.WALProtos.WALKey.Builder getBuilder(
532 WALCellCodec.ByteStringCompressor compressor) throws IOException {
533 org.apache.hadoop.hbase.protobuf.generated.WALProtos.WALKey.Builder builder =
534 org.apache.hadoop.hbase.protobuf.generated.WALProtos.WALKey.newBuilder();
535 if (compressionContext == null) {
536 builder.setEncodedRegionName(ByteStringer.wrap(this.encodedRegionName));
537 builder.setTableName(ByteStringer.wrap(this.tablename.getName()));
538 } else {
539 builder.setEncodedRegionName(compressor.compress(this.encodedRegionName,
540 compressionContext.regionDict));
541 builder.setTableName(compressor.compress(this.tablename.getName(),
542 compressionContext.tableDict));
543 }
544 builder.setLogSequenceNumber(this.logSeqNum);
545 builder.setWriteTime(writeTime);
546 if (this.origLogSeqNum > 0) {
547 builder.setOrigSequenceNumber(this.origLogSeqNum);
548 }
549 if (this.nonce != HConstants.NO_NONCE) {
550 builder.setNonce(nonce);
551 }
552 if (this.nonceGroup != HConstants.NO_NONCE) {
553 builder.setNonceGroup(nonceGroup);
554 }
555 HBaseProtos.UUID.Builder uuidBuilder = HBaseProtos.UUID.newBuilder();
556 for (UUID clusterId : clusterIds) {
557 uuidBuilder.setLeastSigBits(clusterId.getLeastSignificantBits());
558 uuidBuilder.setMostSigBits(clusterId.getMostSignificantBits());
559 builder.addClusterIds(uuidBuilder.build());
560 }
561 if (scopes != null) {
562 for (Map.Entry<byte[], Integer> e : scopes.entrySet()) {
563 ByteString family = (compressionContext == null) ? ByteStringer.wrap(e.getKey())
564 : compressor.compress(e.getKey(), compressionContext.familyDict);
565 builder.addScopes(FamilyScope.newBuilder()
566 .setFamily(family).setScopeType(ScopeType.valueOf(e.getValue())));
567 }
568 }
569 return builder;
570 }
571
572 public void readFieldsFromPb(org.apache.hadoop.hbase.protobuf.generated.WALProtos.WALKey walKey,
573 WALCellCodec.ByteStringUncompressor uncompressor)
574 throws IOException {
575 if (this.compressionContext != null) {
576 this.encodedRegionName = uncompressor.uncompress(
577 walKey.getEncodedRegionName(), compressionContext.regionDict);
578 byte[] tablenameBytes = uncompressor.uncompress(
579 walKey.getTableName(), compressionContext.tableDict);
580 this.tablename = TableName.valueOf(tablenameBytes);
581 } else {
582 this.encodedRegionName = walKey.getEncodedRegionName().toByteArray();
583 this.tablename = TableName.valueOf(walKey.getTableName().toByteArray());
584 }
585 clusterIds.clear();
586 if (walKey.hasClusterId()) {
587
588
589 clusterIds.add(new UUID(walKey.getClusterId().getMostSigBits(), walKey.getClusterId()
590 .getLeastSigBits()));
591 }
592 for (HBaseProtos.UUID clusterId : walKey.getClusterIdsList()) {
593 clusterIds.add(new UUID(clusterId.getMostSigBits(), clusterId.getLeastSigBits()));
594 }
595 if (walKey.hasNonceGroup()) {
596 this.nonceGroup = walKey.getNonceGroup();
597 }
598 if (walKey.hasNonce()) {
599 this.nonce = walKey.getNonce();
600 }
601 this.scopes = null;
602 if (walKey.getScopesCount() > 0) {
603 this.scopes = new TreeMap<byte[], Integer>(Bytes.BYTES_COMPARATOR);
604 for (FamilyScope scope : walKey.getScopesList()) {
605 byte[] family = (compressionContext == null) ? scope.getFamily().toByteArray() :
606 uncompressor.uncompress(scope.getFamily(), compressionContext.familyDict);
607 this.scopes.put(family, scope.getScopeType().getNumber());
608 }
609 }
610 this.logSeqNum = walKey.getLogSequenceNumber();
611 this.writeTime = walKey.getWriteTime();
612 if(walKey.hasOrigSequenceNumber()) {
613 this.origLogSeqNum = walKey.getOrigSequenceNumber();
614 }
615 }
616
617 public long estimatedSerializedSizeOf() {
618 long size = encodedRegionName != null ? encodedRegionName.length : 0;
619 size += tablename != null ? tablename.toBytes().length : 0;
620 if (clusterIds != null) {
621 size += 16 * clusterIds.size();
622 }
623 if (nonceGroup != HConstants.NO_NONCE) {
624 size += Bytes.SIZEOF_LONG;
625 }
626 if (nonce != HConstants.NO_NONCE) {
627 size += Bytes.SIZEOF_LONG;
628 }
629 if (scopes != null) {
630 for (Map.Entry<byte[], Integer> scope: scopes.entrySet()) {
631 size += scope.getKey().length;
632 size += Bytes.SIZEOF_INT;
633 }
634 }
635 size += Bytes.SIZEOF_LONG;
636 size += Bytes.SIZEOF_LONG;
637 if (origLogSeqNum > 0) {
638 size += Bytes.SIZEOF_LONG;
639 }
640 return size;
641 }
642 }