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.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  // imports for things that haven't moved from regionserver.wal yet.
58  import org.apache.hadoop.hbase.regionserver.wal.CompressionContext;
59  import org.apache.hadoop.hbase.regionserver.wal.WALCellCodec;
60  
61  /**
62   * A Key for an entry in the change log.
63   *
64   * The log intermingles edits to many tables and rows, so each log entry
65   * identifies the appropriate table and row.  Within a table and row, they're
66   * also sorted.
67   *
68   * <p>Some Transactional edits (START, COMMIT, ABORT) will not have an
69   * associated row.
70   *
71   * Note that protected members marked @InterfaceAudience.Private are only protected
72   * to support the legacy HLogKey class, which is in a different package.
73   * 
74   * <p>
75   */
76  // TODO: Key and WALEdit are never used separately, or in one-to-many relation, for practical
77  //       purposes. They need to be merged into WALEntry.
78  // TODO: Cleanup. We have logSeqNum and then WriteEntry, both are sequence id'ing. Fix.
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 // For internal use only.
84    public MultiVersionConcurrencyControl getMvcc() {
85      return mvcc;
86    }
87  
88    /**
89     * Will block until a write entry has been assigned by they WAL subsystem.
90     * @return A WriteEntry gotten from local WAL subsystem. Must be completed by calling
91     * mvcc#complete or mvcc#completeAndWait.
92     * @throws InterruptedIOException
93     * @see
94     * #setWriteEntry(org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl.WriteEntry)
95     */
96    @InterfaceAudience.Private // For internal use only.
97    public MultiVersionConcurrencyControl.WriteEntry getWriteEntry() throws InterruptedIOException {
98      assert this.writeEntry != null;
99      return this.writeEntry;
100   }
101 
102   @InterfaceAudience.Private // For internal use only.
103   public void setWriteEntry(MultiVersionConcurrencyControl.WriteEntry writeEntry) {
104     assert this.writeEntry == null;
105     this.writeEntry = writeEntry;
106     // Set our sequenceid now using WriteEntry.
107     this.logSeqNum = writeEntry.getWriteNumber();
108   }
109 
110   // should be < 0 (@see HLogKey#readFields(DataInput))
111   // version 2 supports WAL compression
112   // public members here are only public because of HLogKey
113   @InterfaceAudience.Private
114   protected enum Version {
115     UNVERSIONED(0),
116     // Initial number we put on WALKey when we introduced versioning.
117     INITIAL(-1),
118     // Version -2 introduced a dictionary compression facility.  Only this
119     // dictionary-based compression is available in version -2.
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    * This is used for reading the log entries created by the previous releases
148    * (0.94.11) which write the clusters information to the scopes of WALEdit.
149    */
150   private static final String PREFIX_CLUSTER_KEY = ".";
151 
152 
153   // visible for deprecated HLogKey
154   @InterfaceAudience.Private
155   protected static final Version VERSION = Version.COMPRESSED;
156 
157   /** Used to represent when a particular wal key doesn't know/care about the sequence ordering. */
158   public static final long NO_SEQUENCE_ID = -1;
159 
160 
161   // visible for deprecated HLogKey
162   @InterfaceAudience.Private
163   protected byte [] encodedRegionName;
164   // visible for deprecated HLogKey
165   @InterfaceAudience.Private
166   protected TableName tablename;
167   // visible for deprecated HLogKey
168   @InterfaceAudience.Private
169   protected long logSeqNum;
170   private long origLogSeqNum = 0;
171   // Time at which this edit was written.
172   // visible for deprecated HLogKey
173   @InterfaceAudience.Private
174   protected long writeTime;
175 
176   // The first element in the list is the cluster id on which the change has originated
177   // visible for deprecated HLogKey
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   // visible for deprecated HLogKey
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    * Create the log key for writing to somewhere.
239    * We maintain the tablename mainly for debugging purposes.
240    * A regionName is always a sub-table object.
241    * <p>Used by log splitting and snapshots.
242    *
243    * @param encodedRegionName Encoded name of the region as returned by
244    *                          <code>HRegionInfo#getEncodedNameAsBytes()</code>.
245    * @param tablename         - name of table
246    * @param logSeqNum         - log sequence number
247    * @param now               Time at which this edit was written.
248    * @param clusterIds        the clusters that have consumed the change(used in Replication)
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    * Create the log key for writing to somewhere.
263    * We maintain the tablename mainly for debugging purposes.
264    * A regionName is always a sub-table object.
265    *
266    * @param encodedRegionName Encoded name of the region as returned by
267    *                          <code>HRegionInfo#getEncodedNameAsBytes()</code>.
268    * @param tablename
269    * @param now               Time at which this edit was written.
270    * @param clusterIds        the clusters that have consumed the change(used in Replication)
271    * @param nonceGroup
272    * @param nonce
273    * @param mvcc mvcc control used to generate sequence numbers and control read/write points
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    * Create the log key for writing to somewhere.
283    * We maintain the tablename mainly for debugging purposes.
284    * A regionName is always a sub-table object.
285    *
286    * @param encodedRegionName Encoded name of the region as returned by
287    *                          <code>HRegionInfo#getEncodedNameAsBytes()</code>.
288    * @param tablename
289    * @param logSeqNum
290    * @param nonceGroup
291    * @param nonce
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    * @param compressionContext Compression context to use
330    */
331   public void setCompressionContext(CompressionContext compressionContext) {
332     this.compressionContext = compressionContext;
333   }
334 
335   /** @return encoded region name */
336   public byte [] getEncodedRegionName() {
337     return encodedRegionName;
338   }
339 
340   /** @return table name */
341   public TableName getTablename() {
342     return tablename;
343   }
344 
345   /** @return log sequence number */
346   public long getLogSeqNum() {
347     return this.logSeqNum;
348   }
349 
350   /**
351    * Used to set original seq Id for WALKey during wal replay
352    * @param seqId
353    */
354   public void setOrigLogSeqNum(final long seqId) {
355     this.origLogSeqNum = seqId;
356   }
357   
358   /**
359    * Return a positive long if current WALKey is created from a replay edit
360    * @return original sequence number of the WALEdit
361    */
362   public long getOrigLogSeqNum() {
363     return this.origLogSeqNum;
364   }
365   
366   /**
367    * Wait for sequence number to be assigned &amp; return the assigned value
368    * @return long the new assigned sequence number
369    * @throws IOException
370    */
371   @Override
372   public long getSequenceId() throws IOException {
373     return this.logSeqNum;
374   }
375 
376   /**
377    * @return the write time
378    */
379   public long getWriteTime() {
380     return this.writeTime;
381   }
382 
383   public NavigableMap<byte[], Integer> getScopes() {
384     return scopes;
385   }
386 
387   /** @return The nonce group */
388   public long getNonceGroup() {
389     return nonceGroup;
390   }
391 
392   /** @return The nonce */
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    * Marks that the cluster with the given clusterId has consumed the change
422    */
423   public void addClusterId(UUID clusterId) {
424     if (!clusterIds.contains(clusterId)) {
425       clusterIds.add(clusterId);
426     }
427   }
428 
429   /**
430    * @return the set of cluster Ids that have consumed the change
431    */
432   public List<UUID> getClusterIds() {
433     return clusterIds;
434   }
435 
436   /**
437    * @return the cluster id on which the change has originated. It there is no such cluster, it
438    *         returns DEFAULT_CLUSTER_ID (cases where replication is not enabled)
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    * Produces a string map for this key. Useful for programmatic use and
452    * manipulation of the data stored in an WALKey, for example, printing
453    * as JSON.
454    *
455    * @return a Map containing data from this key
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     // why isn't cluster id accounted for?
502     return result;
503   }
504 
505   /**
506    * Drop this instance's tablename byte array and instead
507    * hold a reference to the provided tablename. This is not
508    * meant to be a general purpose setter - it's only used
509    * to collapse references to conserve memory.
510    */
511   void internTableName(TableName tablename) {
512     // We should not use this as a setter - only to swap
513     // in a new reference to the same table name.
514     assert tablename.equals(this.tablename);
515     this.tablename = tablename;
516   }
517 
518   /**
519    * Drop this instance's region name byte array and instead
520    * hold a reference to the provided region name. This is not
521    * meant to be a general purpose setter - it's only used
522    * to collapse references to conserve memory.
523    */
524   void internEncodedRegionName(byte []encodedRegionName) {
525     // We should not use this as a setter - only to swap
526     // in a new reference to the same table name.
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       //When we are reading the older log (0.95.1 release)
588       //This is definitely the originating cluster
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; // nonce group
625     }
626     if (nonce != HConstants.NO_NONCE) {
627       size += Bytes.SIZEOF_LONG; // nonce
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; // sequence number
636     size += Bytes.SIZEOF_LONG; // write time
637     if (origLogSeqNum > 0) {
638       size += Bytes.SIZEOF_LONG; // original sequence number
639     }
640     return size;
641   }
642 }