View Javadoc

1   /**
2    * Copyright The Apache Software Foundation
3    *
4    * Licensed to the Apache Software Foundation (ASF) under one
5    * or more contributor license agreements.  See the NOTICE file
6    * distributed with this work for additional information
7    * regarding copyright ownership.  The ASF licenses this file
8    * to you under the Apache License, Version 2.0 (the
9    * "License"); you may not use this file except in compliance
10   * with the License.  You may obtain a copy of the License at
11   *
12   *     http://www.apache.org/licenses/LICENSE-2.0
13   *
14   * Unless required by applicable law or agreed to in writing, software
15   * distributed under the License is distributed on an "AS IS" BASIS,
16   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17   * See the License for the specific language governing permissions and
18   * limitations under the License.
19   */
20  package org.apache.hadoop.hbase;
21  
22  import static org.apache.hadoop.hbase.util.Bytes.len;
23  
24  import java.io.DataInput;
25  import java.io.DataOutput;
26  import java.io.IOException;
27  import java.io.InputStream;
28  import java.io.OutputStream;
29  import java.nio.ByteBuffer;
30  import java.util.ArrayList;
31  import java.util.Arrays;
32  import java.util.Comparator;
33  import java.util.HashMap;
34  import java.util.List;
35  import java.util.Map;
36  
37  import org.apache.commons.logging.Log;
38  import org.apache.commons.logging.LogFactory;
39  import org.apache.hadoop.classification.InterfaceAudience;
40  import org.apache.hadoop.hbase.io.HeapSize;
41  import org.apache.hadoop.hbase.io.util.StreamUtils;
42  import org.apache.hadoop.hbase.util.Bytes;
43  import org.apache.hadoop.hbase.util.ClassSize;
44  import org.apache.hadoop.io.IOUtils;
45  import org.apache.hadoop.io.RawComparator;
46  
47  import com.google.common.annotations.VisibleForTesting;
48  
49  /**
50   * An HBase Key/Value. This is the fundamental HBase Type.  
51   * <p>
52   * HBase applications and users should use the Cell interface and avoid directly using KeyValue
53   * and member functions not defined in Cell.
54   * <p>
55   * If being used client-side, the primary methods to access individual fields are {@link #getRow()},
56   * {@link #getFamily()}, {@link #getQualifier()}, {@link #getTimestamp()}, and {@link #getValue()}.
57   * These methods allocate new byte arrays and return copies. Avoid their use server-side.
58   * <p>
59   * Instances of this class are immutable. They do not implement Comparable but Comparators are
60   * provided. Comparators change with context, whether user table or a catalog table comparison. Its
61   * critical you use the appropriate comparator. There are Comparators for normal HFiles, Meta's
62   * Hfiles, and bloom filter keys.
63   * <p>
64   * KeyValue wraps a byte array and takes offsets and lengths into passed array at where to start
65   * interpreting the content as KeyValue. The KeyValue format inside a byte array is:
66   * <code>&lt;keylength> &lt;valuelength> &lt;key> &lt;value></code> Key is further decomposed as:
67   * <code>&lt;rowlength> &lt;row> &lt;columnfamilylength> &lt;columnfamily> &lt;columnqualifier>
68   * &lt;timestamp> &lt;keytype></code>
69   * The <code>rowlength</code> maximum is <code>Short.MAX_SIZE</code>, column family length maximum
70   * is <code>Byte.MAX_SIZE</code>, and column qualifier + key length must be <
71   * <code>Integer.MAX_SIZE</code>. The column does not contain the family/qualifier delimiter,
72   * {@link #COLUMN_FAMILY_DELIMITER}<br>
73   * KeyValue can optionally contain Tags. When it contains tags, it is added in the byte array after
74   * the value part. The format for this part is: <code>&lt;tagslength>&lt;tagsbytes></code>.
75   * <code>tagslength</code> maximum is <code>Short.MAX_SIZE</code>. The <code>tagsbytes</code>
76   * contain one or more tags where as each tag is of the form
77   * <code>&lt;taglength>&lt;tagtype>&lt;tagbytes></code>.  <code>tagtype</code> is one byte and
78   * <code>taglength</code> maximum is <code>Short.MAX_SIZE</code> and it includes 1 byte type length
79   * and actual tag bytes length.
80   */
81  @InterfaceAudience.Private
82  public class KeyValue implements Cell, HeapSize, Cloneable, SettableSequenceId {
83    private static final ArrayList<Tag> EMPTY_ARRAY_LIST = new ArrayList<Tag>();
84  
85    static final Log LOG = LogFactory.getLog(KeyValue.class);
86  
87    /**
88     * Colon character in UTF-8
89     */
90    public static final char COLUMN_FAMILY_DELIMITER = ':';
91  
92    public static final byte[] COLUMN_FAMILY_DELIM_ARRAY =
93      new byte[]{COLUMN_FAMILY_DELIMITER};
94  
95    /**
96     * Comparator for plain key/values; i.e. non-catalog table key/values. Works on Key portion
97     * of KeyValue only.
98     */
99    public static final KVComparator COMPARATOR = new KVComparator();
100   /**
101    * A {@link KVComparator} for <code>hbase:meta</code> catalog table
102    * {@link KeyValue}s.
103    */
104   public static final KVComparator META_COMPARATOR = new MetaComparator();
105 
106   /**
107    * Needed for Bloom Filters.
108    */
109   public static final KVComparator RAW_COMPARATOR = new RawBytesComparator();
110 
111   /** Size of the key length field in bytes*/
112   public static final int KEY_LENGTH_SIZE = Bytes.SIZEOF_INT;
113 
114   /** Size of the key type field in bytes */
115   public static final int TYPE_SIZE = Bytes.SIZEOF_BYTE;
116 
117   /** Size of the row length field in bytes */
118   public static final int ROW_LENGTH_SIZE = Bytes.SIZEOF_SHORT;
119 
120   /** Size of the family length field in bytes */
121   public static final int FAMILY_LENGTH_SIZE = Bytes.SIZEOF_BYTE;
122 
123   /** Size of the timestamp field in bytes */
124   public static final int TIMESTAMP_SIZE = Bytes.SIZEOF_LONG;
125 
126   // Size of the timestamp and type byte on end of a key -- a long + a byte.
127   public static final int TIMESTAMP_TYPE_SIZE = TIMESTAMP_SIZE + TYPE_SIZE;
128 
129   // Size of the length shorts and bytes in key.
130   public static final int KEY_INFRASTRUCTURE_SIZE = ROW_LENGTH_SIZE
131       + FAMILY_LENGTH_SIZE + TIMESTAMP_TYPE_SIZE;
132 
133   // How far into the key the row starts at. First thing to read is the short
134   // that says how long the row is.
135   public static final int ROW_OFFSET =
136     Bytes.SIZEOF_INT /*keylength*/ +
137     Bytes.SIZEOF_INT /*valuelength*/;
138 
139   // Size of the length ints in a KeyValue datastructure.
140   public static final int KEYVALUE_INFRASTRUCTURE_SIZE = ROW_OFFSET;
141 
142   /** Size of the tags length field in bytes */
143   public static final int TAGS_LENGTH_SIZE = Bytes.SIZEOF_SHORT;
144 
145   public static final int KEYVALUE_WITH_TAGS_INFRASTRUCTURE_SIZE = ROW_OFFSET + TAGS_LENGTH_SIZE;
146 
147   private static final int MAX_TAGS_LENGTH = (2 * Short.MAX_VALUE) + 1;
148 
149   /**
150    * Computes the number of bytes that a <code>KeyValue</code> instance with the provided
151    * characteristics would take up for its underlying data structure.
152    *
153    * @param rlength row length
154    * @param flength family length
155    * @param qlength qualifier length
156    * @param vlength value length
157    *
158    * @return the <code>KeyValue</code> data structure length
159    */
160   public static long getKeyValueDataStructureSize(int rlength,
161       int flength, int qlength, int vlength) {
162     return KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE
163         + getKeyDataStructureSize(rlength, flength, qlength) + vlength;
164   }
165 
166   /**
167    * Computes the number of bytes that a <code>KeyValue</code> instance with the provided
168    * characteristics would take up for its underlying data structure.
169    *
170    * @param rlength row length
171    * @param flength family length
172    * @param qlength qualifier length
173    * @param vlength value length
174    * @param tagsLength total length of the tags
175    *
176    * @return the <code>KeyValue</code> data structure length
177    */
178   public static long getKeyValueDataStructureSize(int rlength, int flength, int qlength,
179       int vlength, int tagsLength) {
180     if (tagsLength == 0) {
181       return getKeyValueDataStructureSize(rlength, flength, qlength, vlength);
182     }
183     return KeyValue.KEYVALUE_WITH_TAGS_INFRASTRUCTURE_SIZE
184         + getKeyDataStructureSize(rlength, flength, qlength) + vlength + tagsLength;
185   }
186 
187   /**
188    * Computes the number of bytes that a <code>KeyValue</code> instance with the provided
189    * characteristics would take up for its underlying data structure.
190    *
191    * @param klength key length
192    * @param vlength value length
193    * @param tagsLength total length of the tags
194    *
195    * @return the <code>KeyValue</code> data structure length
196    */
197   public static long getKeyValueDataStructureSize(int klength, int vlength, int tagsLength) {
198     if (tagsLength == 0) {
199       return KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE + klength + vlength;
200     }
201     return KeyValue.KEYVALUE_WITH_TAGS_INFRASTRUCTURE_SIZE + klength + vlength + tagsLength;
202   }
203 
204   /**
205    * Computes the number of bytes that a <code>KeyValue</code> instance with the provided
206    * characteristics would take up in its underlying data structure for the key.
207    *
208    * @param rlength row length
209    * @param flength family length
210    * @param qlength qualifier length
211    *
212    * @return the key data structure length
213    */
214   public static long getKeyDataStructureSize(int rlength, int flength, int qlength) {
215     return KeyValue.KEY_INFRASTRUCTURE_SIZE + rlength + flength + qlength;
216   }
217 
218   /**
219    * Key type.
220    * Has space for other key types to be added later.  Cannot rely on
221    * enum ordinals . They change if item is removed or moved.  Do our own codes.
222    */
223   public static enum Type {
224     Minimum((byte)0),
225     Put((byte)4),
226 
227     Delete((byte)8),
228     DeleteFamilyVersion((byte)10),
229     DeleteColumn((byte)12),
230     DeleteFamily((byte)14),
231 
232     // Maximum is used when searching; you look from maximum on down.
233     Maximum((byte)255);
234 
235     private final byte code;
236 
237     Type(final byte c) {
238       this.code = c;
239     }
240 
241     public byte getCode() {
242       return this.code;
243     }
244 
245     /**
246      * Cannot rely on enum ordinals . They change if item is removed or moved.
247      * Do our own codes.
248      * @param b
249      * @return Type associated with passed code.
250      */
251     public static Type codeToType(final byte b) {
252       for (Type t : Type.values()) {
253         if (t.getCode() == b) {
254           return t;
255         }
256       }
257       throw new RuntimeException("Unknown code " + b);
258     }
259   }
260 
261   /**
262    * Lowest possible key.
263    * Makes a Key with highest possible Timestamp, empty row and column.  No
264    * key can be equal or lower than this one in memstore or in store file.
265    */
266   public static final KeyValue LOWESTKEY =
267     new KeyValue(HConstants.EMPTY_BYTE_ARRAY, HConstants.LATEST_TIMESTAMP);
268 
269   ////
270   // KeyValue core instance fields.
271   private byte [] bytes = null;  // an immutable byte array that contains the KV
272   private int offset = 0;  // offset into bytes buffer KV starts at
273   private int length = 0;  // length of the KV starting from offset.
274 
275   /**
276    * @return True if a delete type, a {@link KeyValue.Type#Delete} or
277    * a {KeyValue.Type#DeleteFamily} or a {@link KeyValue.Type#DeleteColumn}
278    * KeyValue type.
279    */
280   public static boolean isDelete(byte t) {
281     return Type.Delete.getCode() <= t && t <= Type.DeleteFamily.getCode();
282   }
283 
284   /** Here be dragons **/
285 
286   // used to achieve atomic operations in the memstore.
287   @Override
288   public long getMvccVersion() {
289     return this.getSequenceId();
290   }
291 
292   /**
293    * used to achieve atomic operations in the memstore.
294    */
295   @Override
296   public long getSequenceId() {
297     return seqId;
298   }
299 
300   public void setSequenceId(long seqId) {
301     this.seqId = seqId;
302   }
303 
304   // multi-version concurrency control version.  default value is 0, aka do not care.
305   private long seqId = 0;
306 
307   /** Dragon time over, return to normal business */
308 
309 
310   /** Writable Constructor -- DO NOT USE */
311   public KeyValue() {}
312 
313   /**
314    * Creates a KeyValue from the start of the specified byte array.
315    * Presumes <code>bytes</code> content is formatted as a KeyValue blob.
316    * @param bytes byte array
317    */
318   public KeyValue(final byte [] bytes) {
319     this(bytes, 0);
320   }
321 
322   /**
323    * Creates a KeyValue from the specified byte array and offset.
324    * Presumes <code>bytes</code> content starting at <code>offset</code> is
325    * formatted as a KeyValue blob.
326    * @param bytes byte array
327    * @param offset offset to start of KeyValue
328    */
329   public KeyValue(final byte [] bytes, final int offset) {
330     this(bytes, offset, getLength(bytes, offset));
331   }
332 
333   /**
334    * Creates a KeyValue from the specified byte array, starting at offset, and
335    * for length <code>length</code>.
336    * @param bytes byte array
337    * @param offset offset to start of the KeyValue
338    * @param length length of the KeyValue
339    */
340   public KeyValue(final byte [] bytes, final int offset, final int length) {
341     this.bytes = bytes;
342     this.offset = offset;
343     this.length = length;
344   }
345 
346   /**
347    * Creates a KeyValue from the specified byte array, starting at offset, and
348    * for length <code>length</code>.
349    *
350    * @param bytes  byte array
351    * @param offset offset to start of the KeyValue
352    * @param length length of the KeyValue
353    * @param ts
354    */
355   public KeyValue(final byte[] bytes, final int offset, final int length, long ts) {
356     this(bytes, offset, length, null, 0, 0, null, 0, 0, ts, Type.Maximum, null, 0, 0, null);
357   }
358 
359   /** Constructors that build a new backing byte array from fields */
360 
361   /**
362    * Constructs KeyValue structure filled with null value.
363    * Sets type to {@link KeyValue.Type#Maximum}
364    * @param row - row key (arbitrary byte array)
365    * @param timestamp
366    */
367   public KeyValue(final byte [] row, final long timestamp) {
368     this(row, null, null, timestamp, Type.Maximum, null);
369   }
370 
371   /**
372    * Constructs KeyValue structure filled with null value.
373    * @param row - row key (arbitrary byte array)
374    * @param timestamp
375    */
376   public KeyValue(final byte [] row, final long timestamp, Type type) {
377     this(row, null, null, timestamp, type, null);
378   }
379 
380   /**
381    * Constructs KeyValue structure filled with null value.
382    * Sets type to {@link KeyValue.Type#Maximum}
383    * @param row - row key (arbitrary byte array)
384    * @param family family name
385    * @param qualifier column qualifier
386    */
387   public KeyValue(final byte [] row, final byte [] family,
388       final byte [] qualifier) {
389     this(row, family, qualifier, HConstants.LATEST_TIMESTAMP, Type.Maximum);
390   }
391 
392   /**
393    * Constructs KeyValue structure filled with null value.
394    * @param row - row key (arbitrary byte array)
395    * @param family family name
396    * @param qualifier column qualifier
397    */
398   public KeyValue(final byte [] row, final byte [] family,
399       final byte [] qualifier, final byte [] value) {
400     this(row, family, qualifier, HConstants.LATEST_TIMESTAMP, Type.Put, value);
401   }
402 
403   /**
404    * Constructs KeyValue structure filled with specified values.
405    * @param row row key
406    * @param family family name
407    * @param qualifier column qualifier
408    * @param timestamp version timestamp
409    * @param type key type
410    * @throws IllegalArgumentException
411    */
412   public KeyValue(final byte[] row, final byte[] family,
413       final byte[] qualifier, final long timestamp, Type type) {
414     this(row, family, qualifier, timestamp, type, null);
415   }
416 
417   /**
418    * Constructs KeyValue structure filled with specified values.
419    * @param row row key
420    * @param family family name
421    * @param qualifier column qualifier
422    * @param timestamp version timestamp
423    * @param value column value
424    * @throws IllegalArgumentException
425    */
426   public KeyValue(final byte[] row, final byte[] family,
427       final byte[] qualifier, final long timestamp, final byte[] value) {
428     this(row, family, qualifier, timestamp, Type.Put, value);
429   }
430 
431   /**
432    * Constructs KeyValue structure filled with specified values.
433    * @param row row key
434    * @param family family name
435    * @param qualifier column qualifier
436    * @param timestamp version timestamp
437    * @param value column value
438    * @param tags tags
439    * @throws IllegalArgumentException
440    */
441   public KeyValue(final byte[] row, final byte[] family,
442       final byte[] qualifier, final long timestamp, final byte[] value,
443       final Tag[] tags) {
444     this(row, family, qualifier, timestamp, value, tags != null ? Arrays.asList(tags) : null);
445   }
446 
447   /**
448    * Constructs KeyValue structure filled with specified values.
449    * @param row row key
450    * @param family family name
451    * @param qualifier column qualifier
452    * @param timestamp version timestamp
453    * @param value column value
454    * @param tags tags non-empty list of tags or null
455    * @throws IllegalArgumentException
456    */
457   public KeyValue(final byte[] row, final byte[] family,
458       final byte[] qualifier, final long timestamp, final byte[] value,
459       final List<Tag> tags) {
460     this(row, 0, row==null ? 0 : row.length,
461       family, 0, family==null ? 0 : family.length,
462       qualifier, 0, qualifier==null ? 0 : qualifier.length,
463       timestamp, Type.Put,
464       value, 0, value==null ? 0 : value.length, tags);
465   }
466 
467   /**
468    * Constructs KeyValue structure filled with specified values.
469    * @param row row key
470    * @param family family name
471    * @param qualifier column qualifier
472    * @param timestamp version timestamp
473    * @param type key type
474    * @param value column value
475    * @throws IllegalArgumentException
476    */
477   public KeyValue(final byte[] row, final byte[] family,
478       final byte[] qualifier, final long timestamp, Type type,
479       final byte[] value) {
480     this(row, 0, len(row),   family, 0, len(family),   qualifier, 0, len(qualifier),
481         timestamp, type,   value, 0, len(value));
482   }
483 
484   /**
485    * Constructs KeyValue structure filled with specified values.
486    * <p>
487    * Column is split into two fields, family and qualifier.
488    * @param row row key
489    * @param family family name
490    * @param qualifier column qualifier
491    * @param timestamp version timestamp
492    * @param type key type
493    * @param value column value
494    * @throws IllegalArgumentException
495    */
496   public KeyValue(final byte[] row, final byte[] family,
497       final byte[] qualifier, final long timestamp, Type type,
498       final byte[] value, final List<Tag> tags) {
499     this(row, family, qualifier, 0, qualifier==null ? 0 : qualifier.length,
500         timestamp, type, value, 0, value==null ? 0 : value.length, tags);
501   }
502 
503   /**
504    * Constructs KeyValue structure filled with specified values.
505    * @param row row key
506    * @param family family name
507    * @param qualifier column qualifier
508    * @param timestamp version timestamp
509    * @param type key type
510    * @param value column value
511    * @throws IllegalArgumentException
512    */
513   public KeyValue(final byte[] row, final byte[] family,
514       final byte[] qualifier, final long timestamp, Type type,
515       final byte[] value, final byte[] tags) {
516     this(row, family, qualifier, 0, qualifier==null ? 0 : qualifier.length,
517         timestamp, type, value, 0, value==null ? 0 : value.length, tags);
518   }
519 
520   /**
521    * Constructs KeyValue structure filled with specified values.
522    * @param row row key
523    * @param family family name
524    * @param qualifier column qualifier
525    * @param qoffset qualifier offset
526    * @param qlength qualifier length
527    * @param timestamp version timestamp
528    * @param type key type
529    * @param value column value
530    * @param voffset value offset
531    * @param vlength value length
532    * @throws IllegalArgumentException
533    */
534   public KeyValue(byte [] row, byte [] family,
535       byte [] qualifier, int qoffset, int qlength, long timestamp, Type type,
536       byte [] value, int voffset, int vlength, List<Tag> tags) {
537     this(row, 0, row==null ? 0 : row.length,
538         family, 0, family==null ? 0 : family.length,
539         qualifier, qoffset, qlength, timestamp, type,
540         value, voffset, vlength, tags);
541   }
542 
543   /**
544    * @param row
545    * @param family
546    * @param qualifier
547    * @param qoffset
548    * @param qlength
549    * @param timestamp
550    * @param type
551    * @param value
552    * @param voffset
553    * @param vlength
554    * @param tags
555    */
556   public KeyValue(byte [] row, byte [] family,
557       byte [] qualifier, int qoffset, int qlength, long timestamp, Type type,
558       byte [] value, int voffset, int vlength, byte[] tags) {
559     this(row, 0, row==null ? 0 : row.length,
560         family, 0, family==null ? 0 : family.length,
561         qualifier, qoffset, qlength, timestamp, type,
562         value, voffset, vlength, tags, 0, tags==null ? 0 : tags.length);
563   }
564 
565   /**
566    * Constructs KeyValue structure filled with specified values.
567    * <p>
568    * Column is split into two fields, family and qualifier.
569    * @param row row key
570    * @throws IllegalArgumentException
571    */
572   public KeyValue(final byte [] row, final int roffset, final int rlength,
573       final byte [] family, final int foffset, final int flength,
574       final byte [] qualifier, final int qoffset, final int qlength,
575       final long timestamp, final Type type,
576       final byte [] value, final int voffset, final int vlength) {
577     this(row, roffset, rlength, family, foffset, flength, qualifier, qoffset,
578       qlength, timestamp, type, value, voffset, vlength, null);
579   }
580   
581   /**
582    * Constructs KeyValue structure filled with specified values. Uses the provided buffer as the
583    * data buffer.
584    * <p>
585    * Column is split into two fields, family and qualifier.
586    *
587    * @param buffer the bytes buffer to use
588    * @param boffset buffer offset
589    * @param row row key
590    * @param roffset row offset
591    * @param rlength row length
592    * @param family family name
593    * @param foffset family offset
594    * @param flength family length
595    * @param qualifier column qualifier
596    * @param qoffset qualifier offset
597    * @param qlength qualifier length
598    * @param timestamp version timestamp
599    * @param type key type
600    * @param value column value
601    * @param voffset value offset
602    * @param vlength value length
603    * @param tags non-empty list of tags or null
604    * @throws IllegalArgumentException an illegal value was passed or there is insufficient space
605    * remaining in the buffer
606    */
607   public KeyValue(byte [] buffer, final int boffset,
608       final byte [] row, final int roffset, final int rlength,
609       final byte [] family, final int foffset, final int flength,
610       final byte [] qualifier, final int qoffset, final int qlength,
611       final long timestamp, final Type type,
612       final byte [] value, final int voffset, final int vlength,
613       final Tag[] tags) {
614      this.bytes  = buffer;
615      this.length = writeByteArray(buffer, boffset,
616          row, roffset, rlength,
617          family, foffset, flength, qualifier, qoffset, qlength,
618         timestamp, type, value, voffset, vlength, tags);
619      this.offset = boffset;
620    }
621 
622   /**
623    * Constructs KeyValue structure filled with specified values.
624    * <p>
625    * Column is split into two fields, family and qualifier.
626    * @param row row key
627    * @param roffset row offset
628    * @param rlength row length
629    * @param family family name
630    * @param foffset family offset
631    * @param flength family length
632    * @param qualifier column qualifier
633    * @param qoffset qualifier offset
634    * @param qlength qualifier length
635    * @param timestamp version timestamp
636    * @param type key type
637    * @param value column value
638    * @param voffset value offset
639    * @param vlength value length
640    * @param tags tags
641    * @throws IllegalArgumentException
642    */
643   public KeyValue(final byte [] row, final int roffset, final int rlength,
644       final byte [] family, final int foffset, final int flength,
645       final byte [] qualifier, final int qoffset, final int qlength,
646       final long timestamp, final Type type,
647       final byte [] value, final int voffset, final int vlength,
648       final List<Tag> tags) {
649     this.bytes = createByteArray(row, roffset, rlength,
650         family, foffset, flength, qualifier, qoffset, qlength,
651         timestamp, type, value, voffset, vlength, tags);
652     this.length = bytes.length;
653     this.offset = 0;
654   }
655 
656   /**
657    * @param row
658    * @param roffset
659    * @param rlength
660    * @param family
661    * @param foffset
662    * @param flength
663    * @param qualifier
664    * @param qoffset
665    * @param qlength
666    * @param timestamp
667    * @param type
668    * @param value
669    * @param voffset
670    * @param vlength
671    * @param tags
672    */
673   public KeyValue(final byte [] row, final int roffset, final int rlength,
674       final byte [] family, final int foffset, final int flength,
675       final byte [] qualifier, final int qoffset, final int qlength,
676       final long timestamp, final Type type,
677       final byte [] value, final int voffset, final int vlength,
678       final byte[] tags, final int tagsOffset, final int tagsLength) {
679     this.bytes = createByteArray(row, roffset, rlength,
680         family, foffset, flength, qualifier, qoffset, qlength,
681         timestamp, type, value, voffset, vlength, tags, tagsOffset, tagsLength);
682     this.length = bytes.length;
683     this.offset = 0;
684   }
685 
686   /**
687    * Constructs an empty KeyValue structure, with specified sizes.
688    * This can be used to partially fill up KeyValues.
689    * <p>
690    * Column is split into two fields, family and qualifier.
691    * @param rlength row length
692    * @param flength family length
693    * @param qlength qualifier length
694    * @param timestamp version timestamp
695    * @param type key type
696    * @param vlength value length
697    * @throws IllegalArgumentException
698    */
699   public KeyValue(final int rlength,
700       final int flength,
701       final int qlength,
702       final long timestamp, final Type type,
703       final int vlength) {
704     this(rlength, flength, qlength, timestamp, type, vlength, 0);
705   }
706 
707   /**
708    * Constructs an empty KeyValue structure, with specified sizes.
709    * This can be used to partially fill up KeyValues.
710    * <p>
711    * Column is split into two fields, family and qualifier.
712    * @param rlength row length
713    * @param flength family length
714    * @param qlength qualifier length
715    * @param timestamp version timestamp
716    * @param type key type
717    * @param vlength value length
718    * @param tagsLength
719    * @throws IllegalArgumentException
720    */
721   public KeyValue(final int rlength,
722       final int flength,
723       final int qlength,
724       final long timestamp, final Type type,
725       final int vlength, final int tagsLength) {
726     this.bytes = createEmptyByteArray(rlength, flength, qlength, timestamp, type, vlength,
727         tagsLength);
728     this.length = bytes.length;
729     this.offset = 0;
730   }
731 
732 
733   public KeyValue(byte[] row, int roffset, int rlength,
734                   byte[] family, int foffset, int flength,
735                   ByteBuffer qualifier, long ts, Type type, ByteBuffer value, List<Tag> tags) {
736     this.bytes = createByteArray(row, roffset, rlength, family, foffset, flength,
737         qualifier, 0, qualifier == null ? 0 : qualifier.remaining(), ts, type,
738         value, 0, value == null ? 0 : value.remaining(), tags);
739     this.length = bytes.length;
740     this.offset = 0;
741   }
742 
743   public KeyValue(Cell c) {
744     this(c.getRowArray(), c.getRowOffset(), (int)c.getRowLength(),
745         c.getFamilyArray(), c.getFamilyOffset(), (int)c.getFamilyLength(), 
746         c.getQualifierArray(), c.getQualifierOffset(), (int) c.getQualifierLength(), 
747         c.getTimestamp(), Type.codeToType(c.getTypeByte()), c.getValueArray(), c.getValueOffset(), 
748         c.getValueLength(), c.getTagsArray(), c.getTagsOffset(), c.getTagsLength());
749   }
750 
751   /**
752    * Create an empty byte[] representing a KeyValue
753    * All lengths are preset and can be filled in later.
754    * @param rlength
755    * @param flength
756    * @param qlength
757    * @param timestamp
758    * @param type
759    * @param vlength
760    * @return The newly created byte array.
761    */
762   private static byte[] createEmptyByteArray(final int rlength, int flength,
763       int qlength, final long timestamp, final Type type, int vlength, int tagsLength) {
764     if (rlength > Short.MAX_VALUE) {
765       throw new IllegalArgumentException("Row > " + Short.MAX_VALUE);
766     }
767     if (flength > Byte.MAX_VALUE) {
768       throw new IllegalArgumentException("Family > " + Byte.MAX_VALUE);
769     }
770     // Qualifier length
771     if (qlength > Integer.MAX_VALUE - rlength - flength) {
772       throw new IllegalArgumentException("Qualifier > " + Integer.MAX_VALUE);
773     }
774     checkForTagsLength(tagsLength);
775     // Key length
776     long longkeylength = getKeyDataStructureSize(rlength, flength, qlength);
777     if (longkeylength > Integer.MAX_VALUE) {
778       throw new IllegalArgumentException("keylength " + longkeylength + " > " +
779         Integer.MAX_VALUE);
780     }
781     int keylength = (int)longkeylength;
782     // Value length
783     if (vlength > HConstants.MAXIMUM_VALUE_LENGTH) { // FindBugs INT_VACUOUS_COMPARISON
784       throw new IllegalArgumentException("Valuer > " +
785           HConstants.MAXIMUM_VALUE_LENGTH);
786     }
787 
788     // Allocate right-sized byte array.
789     byte[] bytes= new byte[(int) getKeyValueDataStructureSize(rlength, flength, qlength, vlength,
790         tagsLength)];
791     // Write the correct size markers
792     int pos = 0;
793     pos = Bytes.putInt(bytes, pos, keylength);
794     pos = Bytes.putInt(bytes, pos, vlength);
795     pos = Bytes.putShort(bytes, pos, (short)(rlength & 0x0000ffff));
796     pos += rlength;
797     pos = Bytes.putByte(bytes, pos, (byte)(flength & 0x0000ff));
798     pos += flength + qlength;
799     pos = Bytes.putLong(bytes, pos, timestamp);
800     pos = Bytes.putByte(bytes, pos, type.getCode());
801     pos += vlength;
802     if (tagsLength > 0) {
803       pos = Bytes.putAsShort(bytes, pos, tagsLength);
804     }
805     return bytes;
806   }
807 
808   /**
809    * Checks the parameters passed to a constructor.
810    *
811    * @param row row key
812    * @param rlength row length
813    * @param family family name
814    * @param flength family length
815    * @param qlength qualifier length
816    * @param vlength value length
817    *
818    * @throws IllegalArgumentException an illegal value was passed
819    */
820   private static void checkParameters(final byte [] row, final int rlength,
821       final byte [] family, int flength, int qlength, int vlength)
822           throws IllegalArgumentException {
823     if (rlength > Short.MAX_VALUE) {
824       throw new IllegalArgumentException("Row > " + Short.MAX_VALUE);
825     }
826     if (row == null) {
827       throw new IllegalArgumentException("Row is null");
828     }
829     // Family length
830     flength = family == null ? 0 : flength;
831     if (flength > Byte.MAX_VALUE) {
832       throw new IllegalArgumentException("Family > " + Byte.MAX_VALUE);
833     }
834     // Qualifier length
835     if (qlength > Integer.MAX_VALUE - rlength - flength) {
836       throw new IllegalArgumentException("Qualifier > " + Integer.MAX_VALUE);
837     }
838     // Key length
839     long longKeyLength = getKeyDataStructureSize(rlength, flength, qlength);
840     if (longKeyLength > Integer.MAX_VALUE) {
841       throw new IllegalArgumentException("keylength " + longKeyLength + " > " +
842           Integer.MAX_VALUE);
843     }
844     // Value length
845     if (vlength > HConstants.MAXIMUM_VALUE_LENGTH) { // FindBugs INT_VACUOUS_COMPARISON
846       throw new IllegalArgumentException("Value length " + vlength + " > " +
847           HConstants.MAXIMUM_VALUE_LENGTH);
848     }
849   }
850 
851   /**
852    * Write KeyValue format into the provided byte array.
853    *
854    * @param buffer the bytes buffer to use
855    * @param boffset buffer offset
856    * @param row row key
857    * @param roffset row offset
858    * @param rlength row length
859    * @param family family name
860    * @param foffset family offset
861    * @param flength family length
862    * @param qualifier column qualifier
863    * @param qoffset qualifier offset
864    * @param qlength qualifier length
865    * @param timestamp version timestamp
866    * @param type key type
867    * @param value column value
868    * @param voffset value offset
869    * @param vlength value length
870    *
871    * @return The number of useful bytes in the buffer.
872    *
873    * @throws IllegalArgumentException an illegal value was passed or there is insufficient space
874    * remaining in the buffer
875    */
876   public static int writeByteArray(byte [] buffer, final int boffset,
877       final byte [] row, final int roffset, final int rlength,
878       final byte [] family, final int foffset, int flength,
879       final byte [] qualifier, final int qoffset, int qlength,
880       final long timestamp, final Type type,
881       final byte [] value, final int voffset, int vlength, Tag[] tags) {
882 
883     checkParameters(row, rlength, family, flength, qlength, vlength);
884 
885     // Calculate length of tags area
886     int tagsLength = 0;
887     if (tags != null && tags.length > 0) {
888       for (Tag t: tags) {
889         tagsLength += t.getLength();
890       }
891     }
892     checkForTagsLength(tagsLength);
893     int keyLength = (int) getKeyDataStructureSize(rlength, flength, qlength);
894     int keyValueLength = (int) getKeyValueDataStructureSize(rlength, flength, qlength, vlength,
895         tagsLength);
896     if (keyValueLength > buffer.length - boffset) {
897       throw new IllegalArgumentException("Buffer size " + (buffer.length - boffset) + " < " +
898           keyValueLength);
899     }
900 
901     // Write key, value and key row length.
902     int pos = boffset;
903     pos = Bytes.putInt(buffer, pos, keyLength);
904     pos = Bytes.putInt(buffer, pos, vlength);
905     pos = Bytes.putShort(buffer, pos, (short)(rlength & 0x0000ffff));
906     pos = Bytes.putBytes(buffer, pos, row, roffset, rlength);
907     pos = Bytes.putByte(buffer, pos, (byte) (flength & 0x0000ff));
908     if (flength != 0) {
909       pos = Bytes.putBytes(buffer, pos, family, foffset, flength);
910     }
911     if (qlength != 0) {
912       pos = Bytes.putBytes(buffer, pos, qualifier, qoffset, qlength);
913     }
914     pos = Bytes.putLong(buffer, pos, timestamp);
915     pos = Bytes.putByte(buffer, pos, type.getCode());
916     if (value != null && value.length > 0) {
917       pos = Bytes.putBytes(buffer, pos, value, voffset, vlength);
918     }
919     // Write the number of tags. If it is 0 then it means there are no tags.
920     if (tagsLength > 0) {
921       pos = Bytes.putAsShort(buffer, pos, tagsLength);
922       for (Tag t : tags) {
923         pos = Bytes.putBytes(buffer, pos, t.getBuffer(), t.getOffset(), t.getLength());
924       }
925     }
926     return keyValueLength;
927   }
928 
929   private static void checkForTagsLength(int tagsLength) {
930     if (tagsLength > MAX_TAGS_LENGTH) {
931       throw new IllegalArgumentException("tagslength "+ tagsLength + " > " + MAX_TAGS_LENGTH);
932     }
933   }
934 
935   /**
936    * Write KeyValue format into a byte array.
937    * @param row row key
938    * @param roffset row offset
939    * @param rlength row length
940    * @param family family name
941    * @param foffset family offset
942    * @param flength family length
943    * @param qualifier column qualifier
944    * @param qoffset qualifier offset
945    * @param qlength qualifier length
946    * @param timestamp version timestamp
947    * @param type key type
948    * @param value column value
949    * @param voffset value offset
950    * @param vlength value length
951    * @return The newly created byte array.
952    */
953   private static byte [] createByteArray(final byte [] row, final int roffset,
954       final int rlength, final byte [] family, final int foffset, int flength,
955       final byte [] qualifier, final int qoffset, int qlength,
956       final long timestamp, final Type type,
957       final byte [] value, final int voffset, 
958       int vlength, byte[] tags, int tagsOffset, int tagsLength) {
959 
960     checkParameters(row, rlength, family, flength, qlength, vlength);
961     checkForTagsLength(tagsLength);
962     // Allocate right-sized byte array.
963     int keyLength = (int) getKeyDataStructureSize(rlength, flength, qlength);
964     byte [] bytes =
965         new byte[(int) getKeyValueDataStructureSize(rlength, flength, qlength, vlength, tagsLength)];
966     // Write key, value and key row length.
967     int pos = 0;
968     pos = Bytes.putInt(bytes, pos, keyLength);
969     pos = Bytes.putInt(bytes, pos, vlength);
970     pos = Bytes.putShort(bytes, pos, (short)(rlength & 0x0000ffff));
971     pos = Bytes.putBytes(bytes, pos, row, roffset, rlength);
972     pos = Bytes.putByte(bytes, pos, (byte)(flength & 0x0000ff));
973     if(flength != 0) {
974       pos = Bytes.putBytes(bytes, pos, family, foffset, flength);
975     }
976     if(qlength != 0) {
977       pos = Bytes.putBytes(bytes, pos, qualifier, qoffset, qlength);
978     }
979     pos = Bytes.putLong(bytes, pos, timestamp);
980     pos = Bytes.putByte(bytes, pos, type.getCode());
981     if (value != null && value.length > 0) {
982       pos = Bytes.putBytes(bytes, pos, value, voffset, vlength);
983     }
984     // Add the tags after the value part
985     if (tagsLength > 0) {
986       pos = Bytes.putAsShort(bytes, pos, tagsLength);
987       pos = Bytes.putBytes(bytes, pos, tags, tagsOffset, tagsLength);
988     }
989     return bytes;
990   }
991 
992   /**
993    * @param qualifier can be a ByteBuffer or a byte[], or null.
994    * @param value can be a ByteBuffer or a byte[], or null.
995    */
996   private static byte [] createByteArray(final byte [] row, final int roffset,
997       final int rlength, final byte [] family, final int foffset, int flength,
998       final Object qualifier, final int qoffset, int qlength,
999       final long timestamp, final Type type,
1000       final Object value, final int voffset, int vlength, List<Tag> tags) {
1001 
1002     checkParameters(row, rlength, family, flength, qlength, vlength);
1003 
1004     // Calculate length of tags area
1005     int tagsLength = 0;
1006     if (tags != null && !tags.isEmpty()) {
1007       for (Tag t : tags) {
1008         tagsLength += t.getLength();
1009       }
1010     }
1011     checkForTagsLength(tagsLength);
1012     // Allocate right-sized byte array.
1013     int keyLength = (int) getKeyDataStructureSize(rlength, flength, qlength);
1014     byte[] bytes = new byte[(int) getKeyValueDataStructureSize(rlength, flength, qlength, vlength,
1015         tagsLength)];
1016 
1017     // Write key, value and key row length.
1018     int pos = 0;
1019     pos = Bytes.putInt(bytes, pos, keyLength);
1020 
1021     pos = Bytes.putInt(bytes, pos, vlength);
1022     pos = Bytes.putShort(bytes, pos, (short)(rlength & 0x0000ffff));
1023     pos = Bytes.putBytes(bytes, pos, row, roffset, rlength);
1024     pos = Bytes.putByte(bytes, pos, (byte)(flength & 0x0000ff));
1025     if(flength != 0) {
1026       pos = Bytes.putBytes(bytes, pos, family, foffset, flength);
1027     }
1028     if (qlength > 0) {
1029       if (qualifier instanceof ByteBuffer) {
1030         pos = Bytes.putByteBuffer(bytes, pos, (ByteBuffer) qualifier);
1031       } else {
1032         pos = Bytes.putBytes(bytes, pos, (byte[]) qualifier, qoffset, qlength);
1033       }
1034     }
1035     pos = Bytes.putLong(bytes, pos, timestamp);
1036     pos = Bytes.putByte(bytes, pos, type.getCode());
1037     if (vlength > 0) {
1038       if (value instanceof ByteBuffer) {
1039         pos = Bytes.putByteBuffer(bytes, pos, (ByteBuffer) value);
1040       } else {
1041         pos = Bytes.putBytes(bytes, pos, (byte[]) value, voffset, vlength);
1042       }
1043     }
1044     // Add the tags after the value part
1045     if (tagsLength > 0) {
1046       pos = Bytes.putAsShort(bytes, pos, tagsLength);
1047       for (Tag t : tags) {
1048         pos = Bytes.putBytes(bytes, pos, t.getBuffer(), t.getOffset(), t.getLength());
1049       }
1050     }
1051     return bytes;
1052   }
1053 
1054   /**
1055    * Needed doing 'contains' on List.  Only compares the key portion, not the value.
1056    */
1057   @Override
1058   public boolean equals(Object other) {
1059     if (!(other instanceof Cell)) {
1060       return false;
1061     }
1062     return CellComparator.equals(this, (Cell)other);
1063   }
1064 
1065   @Override
1066   public int hashCode() {
1067     byte[] b = getBuffer();
1068     int start = getOffset(), end = getOffset() + getLength();
1069     int h = b[start++];
1070     for (int i = start; i < end; i++) {
1071       h = (h * 13) ^ b[i];
1072     }
1073     return h;
1074   }
1075 
1076   //---------------------------------------------------------------------------
1077   //
1078   //  KeyValue cloning
1079   //
1080   //---------------------------------------------------------------------------
1081 
1082   /**
1083    * Clones a KeyValue.  This creates a copy, re-allocating the buffer.
1084    * @return Fully copied clone of this KeyValue
1085    * @throws CloneNotSupportedException
1086    */
1087   @Override
1088   public KeyValue clone() throws CloneNotSupportedException {
1089     super.clone();
1090     byte [] b = new byte[this.length];
1091     System.arraycopy(this.bytes, this.offset, b, 0, this.length);
1092     KeyValue ret = new KeyValue(b, 0, b.length);
1093     // Important to clone the memstoreTS as well - otherwise memstore's
1094     // update-in-place methods (eg increment) will end up creating
1095     // new entries
1096     ret.setSequenceId(seqId);
1097     return ret;
1098   }
1099 
1100   /**
1101    * Creates a shallow copy of this KeyValue, reusing the data byte buffer.
1102    * http://en.wikipedia.org/wiki/Object_copy
1103    * @return Shallow copy of this KeyValue
1104    */
1105   public KeyValue shallowCopy() {
1106     KeyValue shallowCopy = new KeyValue(this.bytes, this.offset, this.length);
1107     shallowCopy.setSequenceId(this.seqId);
1108     return shallowCopy;
1109   }
1110 
1111   //---------------------------------------------------------------------------
1112   //
1113   //  String representation
1114   //
1115   //---------------------------------------------------------------------------
1116 
1117   public String toString() {
1118     if (this.bytes == null || this.bytes.length == 0) {
1119       return "empty";
1120     }
1121     return keyToString(this.bytes, this.offset + ROW_OFFSET, getKeyLength()) + "/vlen="
1122       + getValueLength() + "/seqid=" + seqId;
1123   }
1124 
1125   /**
1126    * @param k Key portion of a KeyValue.
1127    * @return Key as a String, empty string if k is null. 
1128    */
1129   public static String keyToString(final byte [] k) {
1130     if (k == null) { 
1131       return "";
1132     }
1133     return keyToString(k, 0, k.length);
1134   }
1135 
1136   /**
1137    * Produces a string map for this key/value pair. Useful for programmatic use
1138    * and manipulation of the data stored in an HLogKey, for example, printing
1139    * as JSON. Values are left out due to their tendency to be large. If needed,
1140    * they can be added manually.
1141    *
1142    * @return the Map<String,?> containing data from this key
1143    */
1144   public Map<String, Object> toStringMap() {
1145     Map<String, Object> stringMap = new HashMap<String, Object>();
1146     stringMap.put("row", Bytes.toStringBinary(getRow()));
1147     stringMap.put("family", Bytes.toStringBinary(getFamily()));
1148     stringMap.put("qualifier", Bytes.toStringBinary(getQualifier()));
1149     stringMap.put("timestamp", getTimestamp());
1150     stringMap.put("vlen", getValueLength());
1151     List<Tag> tags = getTags();
1152     if (tags != null) {
1153       List<String> tagsString = new ArrayList<String>();
1154       for (Tag t : tags) {
1155         tagsString.add((t.getType()) + ":" +Bytes.toStringBinary(t.getValue()));
1156       }
1157       stringMap.put("tag", tagsString);
1158     }
1159     return stringMap;
1160   }
1161 
1162   /**
1163    * Use for logging.
1164    * @param b Key portion of a KeyValue.
1165    * @param o Offset to start of key
1166    * @param l Length of key.
1167    * @return Key as a String.
1168    */
1169   public static String keyToString(final byte [] b, final int o, final int l) {
1170     if (b == null) return "";
1171     int rowlength = Bytes.toShort(b, o);
1172     String row = Bytes.toStringBinary(b, o + Bytes.SIZEOF_SHORT, rowlength);
1173     int columnoffset = o + Bytes.SIZEOF_SHORT + 1 + rowlength;
1174     int familylength = b[columnoffset - 1];
1175     int columnlength = l - ((columnoffset - o) + TIMESTAMP_TYPE_SIZE);
1176     String family = familylength == 0? "":
1177       Bytes.toStringBinary(b, columnoffset, familylength);
1178     String qualifier = columnlength == 0? "":
1179       Bytes.toStringBinary(b, columnoffset + familylength,
1180       columnlength - familylength);
1181     long timestamp = Bytes.toLong(b, o + (l - TIMESTAMP_TYPE_SIZE));
1182     String timestampStr = humanReadableTimestamp(timestamp);
1183     byte type = b[o + l - 1];
1184     return row + "/" + family +
1185       (family != null && family.length() > 0? ":" :"") +
1186       qualifier + "/" + timestampStr + "/" + Type.codeToType(type);
1187   }
1188 
1189   public static String humanReadableTimestamp(final long timestamp) {
1190     if (timestamp == HConstants.LATEST_TIMESTAMP) {
1191       return "LATEST_TIMESTAMP";
1192     }
1193     if (timestamp == HConstants.OLDEST_TIMESTAMP) {
1194       return "OLDEST_TIMESTAMP";
1195     }
1196     return String.valueOf(timestamp);
1197   }
1198 
1199   //---------------------------------------------------------------------------
1200   //
1201   //  Public Member Accessors
1202   //
1203   //---------------------------------------------------------------------------
1204 
1205   /**
1206    * @return The byte array backing this KeyValue.
1207    * @deprecated Since 0.98.0.  Use Cell Interface instead.  Do not presume single backing buffer.
1208    */
1209   @Deprecated
1210   public byte [] getBuffer() {
1211     return this.bytes;
1212   }
1213 
1214   /**
1215    * @return Offset into {@link #getBuffer()} at which this KeyValue starts.
1216    */
1217   public int getOffset() {
1218     return this.offset;
1219   }
1220 
1221   /**
1222    * @return Length of bytes this KeyValue occupies in {@link #getBuffer()}.
1223    */
1224   public int getLength() {
1225     return length;
1226   }
1227 
1228   //---------------------------------------------------------------------------
1229   //
1230   //  Length and Offset Calculators
1231   //
1232   //---------------------------------------------------------------------------
1233 
1234   /**
1235    * Determines the total length of the KeyValue stored in the specified
1236    * byte array and offset.  Includes all headers.
1237    * @param bytes byte array
1238    * @param offset offset to start of the KeyValue
1239    * @return length of entire KeyValue, in bytes
1240    */
1241   private static int getLength(byte [] bytes, int offset) {
1242     int klength = ROW_OFFSET + Bytes.toInt(bytes, offset);
1243     int vlength = Bytes.toInt(bytes, offset + Bytes.SIZEOF_INT);
1244     return klength + vlength;
1245   }
1246 
1247   /**
1248    * @return Key offset in backing buffer..
1249    */
1250   public int getKeyOffset() {
1251     return this.offset + ROW_OFFSET;
1252   }
1253 
1254   public String getKeyString() {
1255     return Bytes.toStringBinary(getBuffer(), getKeyOffset(), getKeyLength());
1256   }
1257 
1258   /**
1259    * @return Length of key portion.
1260    */
1261   public int getKeyLength() {
1262     return Bytes.toInt(this.bytes, this.offset);
1263   }
1264 
1265   /**
1266    * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
1267    */
1268   @Override
1269   public byte[] getValueArray() {
1270     return bytes;
1271   }
1272 
1273   /**
1274    * @return the value offset
1275    */
1276   @Override
1277   public int getValueOffset() {
1278     int voffset = getKeyOffset() + getKeyLength();
1279     return voffset;
1280   }
1281 
1282   /**
1283    * @return Value length
1284    */
1285   @Override
1286   public int getValueLength() {
1287     int vlength = Bytes.toInt(this.bytes, this.offset + Bytes.SIZEOF_INT);
1288     return vlength;
1289   }
1290 
1291   /**
1292    * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
1293    */
1294   @Override
1295   public byte[] getRowArray() {
1296     return bytes;
1297   }
1298 
1299   /**
1300    * @return Row offset
1301    */
1302   @Override
1303   public int getRowOffset() {
1304     return getKeyOffset() + Bytes.SIZEOF_SHORT;
1305   }
1306 
1307   /**
1308    * @return Row length
1309    */
1310   @Override
1311   public short getRowLength() {
1312     return Bytes.toShort(this.bytes, getKeyOffset());
1313   }
1314 
1315   /**
1316    * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
1317    */
1318   @Override
1319   public byte[] getFamilyArray() {
1320     return bytes;
1321   }
1322 
1323   /**
1324    * @return Family offset
1325    */
1326   @Override
1327   public int getFamilyOffset() {
1328     return getFamilyOffset(getRowLength());
1329   }
1330 
1331   /**
1332    * @return Family offset
1333    */
1334   private int getFamilyOffset(int rlength) {
1335     return this.offset + ROW_OFFSET + Bytes.SIZEOF_SHORT + rlength + Bytes.SIZEOF_BYTE;
1336   }
1337 
1338   /**
1339    * @return Family length
1340    */
1341   @Override
1342   public byte getFamilyLength() {
1343     return getFamilyLength(getFamilyOffset());
1344   }
1345 
1346   /**
1347    * @return Family length
1348    */
1349   public byte getFamilyLength(int foffset) {
1350     return this.bytes[foffset-1];
1351   }
1352 
1353   /**
1354    * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
1355    */
1356   @Override
1357   public byte[] getQualifierArray() {
1358     return bytes;
1359   }
1360 
1361   /**
1362    * @return Qualifier offset
1363    */
1364   @Override
1365   public int getQualifierOffset() {
1366     return getQualifierOffset(getFamilyOffset());
1367   }
1368 
1369   /**
1370    * @return Qualifier offset
1371    */
1372   private int getQualifierOffset(int foffset) {
1373     return foffset + getFamilyLength(foffset);
1374   }
1375 
1376   /**
1377    * @return Qualifier length
1378    */
1379   @Override
1380   public int getQualifierLength() {
1381     return getQualifierLength(getRowLength(),getFamilyLength());
1382   }
1383 
1384   /**
1385    * @return Qualifier length
1386    */
1387   private int getQualifierLength(int rlength, int flength) {
1388     return getKeyLength() - (int) getKeyDataStructureSize(rlength, flength, 0);
1389   }
1390 
1391   /**
1392    * @return Timestamp offset
1393    */
1394   public int getTimestampOffset() {
1395     return getTimestampOffset(getKeyLength());
1396   }
1397 
1398   /**
1399    * @param keylength Pass if you have it to save on a int creation.
1400    * @return Timestamp offset
1401    */
1402   private int getTimestampOffset(final int keylength) {
1403     return getKeyOffset() + keylength - TIMESTAMP_TYPE_SIZE;
1404   }
1405 
1406   /**
1407    * @return True if this KeyValue has a LATEST_TIMESTAMP timestamp.
1408    */
1409   public boolean isLatestTimestamp() {
1410     return Bytes.equals(getBuffer(), getTimestampOffset(), Bytes.SIZEOF_LONG,
1411       HConstants.LATEST_TIMESTAMP_BYTES, 0, Bytes.SIZEOF_LONG);
1412   }
1413 
1414   /**
1415    * @param now Time to set into <code>this</code> IFF timestamp ==
1416    * {@link HConstants#LATEST_TIMESTAMP} (else, its a noop).
1417    * @return True is we modified this.
1418    */
1419   public boolean updateLatestStamp(final byte [] now) {
1420     if (this.isLatestTimestamp()) {
1421       int tsOffset = getTimestampOffset();
1422       System.arraycopy(now, 0, this.bytes, tsOffset, Bytes.SIZEOF_LONG);
1423       // clear cache or else getTimestamp() possibly returns an old value
1424       return true;
1425     }
1426     return false;
1427   }
1428 
1429   //---------------------------------------------------------------------------
1430   //
1431   //  Methods that return copies of fields
1432   //
1433   //---------------------------------------------------------------------------
1434 
1435   /**
1436    * Do not use unless you have to.  Used internally for compacting and testing.
1437    *
1438    * Use {@link #getRow()}, {@link #getFamily()}, {@link #getQualifier()}, and
1439    * {@link #getValue()} if accessing a KeyValue client-side.
1440    * @return Copy of the key portion only.
1441    */
1442   public byte [] getKey() {
1443     int keylength = getKeyLength();
1444     byte [] key = new byte[keylength];
1445     System.arraycopy(getBuffer(), getKeyOffset(), key, 0, keylength);
1446     return key;
1447   }
1448 
1449   /**
1450    * Returns value in a new byte array.
1451    * Primarily for use client-side. If server-side, use
1452    * {@link #getBuffer()} with appropriate offsets and lengths instead to
1453    * save on allocations.
1454    * @return Value in a new byte array.
1455    */
1456   @Deprecated // use CellUtil.getValueArray()
1457   public byte [] getValue() {
1458     return CellUtil.cloneValue(this);
1459   }
1460 
1461   /**
1462    * Primarily for use client-side.  Returns the row of this KeyValue in a new
1463    * byte array.<p>
1464    *
1465    * If server-side, use {@link #getBuffer()} with appropriate offsets and
1466    * lengths instead.
1467    * @return Row in a new byte array.
1468    */
1469   @Deprecated // use CellUtil.getRowArray()
1470   public byte [] getRow() {
1471     return CellUtil.cloneRow(this);
1472   }
1473 
1474   /**
1475    *
1476    * @return Timestamp
1477    */
1478   @Override
1479   public long getTimestamp() {
1480     return getTimestamp(getKeyLength());
1481   }
1482 
1483   /**
1484    * @param keylength Pass if you have it to save on a int creation.
1485    * @return Timestamp
1486    */
1487   long getTimestamp(final int keylength) {
1488     int tsOffset = getTimestampOffset(keylength);
1489     return Bytes.toLong(this.bytes, tsOffset);
1490   }
1491 
1492   /**
1493    * @return Type of this KeyValue.
1494    */
1495   @Deprecated
1496   public byte getType() {
1497     return getTypeByte();
1498   }
1499 
1500   /**
1501    * @return KeyValue.TYPE byte representation
1502    */
1503   @Override
1504   public byte getTypeByte() {
1505     return this.bytes[this.offset + getKeyLength() - 1 + ROW_OFFSET];
1506   }
1507 
1508   /**
1509    * @return True if a delete type, a {@link KeyValue.Type#Delete} or
1510    * a {KeyValue.Type#DeleteFamily} or a {@link KeyValue.Type#DeleteColumn}
1511    * KeyValue type.
1512    */
1513   @Deprecated // use CellUtil#isDelete
1514   public boolean isDelete() {
1515     return KeyValue.isDelete(getType());
1516   }
1517 
1518   /**
1519    * Primarily for use client-side.  Returns the family of this KeyValue in a
1520    * new byte array.<p>
1521    *
1522    * If server-side, use {@link #getBuffer()} with appropriate offsets and
1523    * lengths instead.
1524    * @return Returns family. Makes a copy.
1525    */
1526   @Deprecated // use CellUtil.getFamilyArray
1527   public byte [] getFamily() {
1528     return CellUtil.cloneFamily(this);
1529   }
1530 
1531   /**
1532    * Primarily for use client-side.  Returns the column qualifier of this
1533    * KeyValue in a new byte array.<p>
1534    *
1535    * If server-side, use {@link #getBuffer()} with appropriate offsets and
1536    * lengths instead.
1537    * Use {@link #getBuffer()} with appropriate offsets and lengths instead.
1538    * @return Returns qualifier. Makes a copy.
1539    */
1540   @Deprecated // use CellUtil.getQualifierArray
1541   public byte [] getQualifier() {
1542     return CellUtil.cloneQualifier(this);
1543   }
1544 
1545   /**
1546    * This returns the offset where the tag actually starts.
1547    */
1548   @Override
1549   public int getTagsOffset() {
1550     int tagsLen = getTagsLength();
1551     if (tagsLen == 0) {
1552       return this.offset + this.length;
1553     }
1554     return this.offset + this.length - tagsLen;
1555   }
1556 
1557   /**
1558    * This returns the total length of the tag bytes
1559    */
1560   @Override
1561   public int getTagsLength() {
1562     int tagsLen = this.length - (getKeyLength() + getValueLength() + KEYVALUE_INFRASTRUCTURE_SIZE);
1563     if (tagsLen > 0) {
1564       // There are some Tag bytes in the byte[]. So reduce 2 bytes which is added to denote the tags
1565       // length
1566       tagsLen -= TAGS_LENGTH_SIZE;
1567     }
1568     return tagsLen;
1569   }
1570 
1571   /**
1572    * Returns any tags embedded in the KeyValue.  Used in testcases.
1573    * @return The tags
1574    */
1575   public List<Tag> getTags() {
1576     int tagsLength = getTagsLength();
1577     if (tagsLength == 0) {
1578       return EMPTY_ARRAY_LIST;
1579     }
1580     return Tag.asList(getTagsArray(), getTagsOffset(), tagsLength);
1581   }
1582 
1583   /**
1584    * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
1585    */
1586   @Override
1587   public byte[] getTagsArray() {
1588     return bytes;
1589   }
1590 
1591   /**
1592    * Creates a new KeyValue that only contains the key portion (the value is
1593    * set to be null).
1594    *
1595    * TODO only used by KeyOnlyFilter -- move there.
1596    * @param lenAsVal replace value with the actual value length (false=empty)
1597    */
1598   public KeyValue createKeyOnly(boolean lenAsVal) {
1599     // KV format:  <keylen:4><valuelen:4><key:keylen><value:valuelen>
1600     // Rebuild as: <keylen:4><0:4><key:keylen>
1601     int dataLen = lenAsVal? Bytes.SIZEOF_INT : 0;
1602     byte [] newBuffer = new byte[getKeyLength() + ROW_OFFSET + dataLen];
1603     System.arraycopy(this.bytes, this.offset, newBuffer, 0,
1604         Math.min(newBuffer.length,this.length));
1605     Bytes.putInt(newBuffer, Bytes.SIZEOF_INT, dataLen);
1606     if (lenAsVal) {
1607       Bytes.putInt(newBuffer, newBuffer.length - dataLen, this.getValueLength());
1608     }
1609     return new KeyValue(newBuffer);
1610   }
1611 
1612   /**
1613    * Splits a column in {@code family:qualifier} form into separate byte arrays. An empty qualifier
1614    * (ie, {@code fam:}) is parsed as <code>{ fam, EMPTY_BYTE_ARRAY }</code> while no delimiter (ie,
1615    * {@code fam}) is parsed as an array of one element, <code>{ fam }</code>.
1616    * <p>
1617    * Don't forget, HBase DOES support empty qualifiers. (see HBASE-9549)
1618    * </p>
1619    * <p>
1620    * Not recommend to be used as this is old-style API.
1621    * </p>
1622    * @param c The column.
1623    * @return The parsed column.
1624    */
1625   public static byte [][] parseColumn(byte [] c) {
1626     final int index = getDelimiter(c, 0, c.length, COLUMN_FAMILY_DELIMITER);
1627     if (index == -1) {
1628       // If no delimiter, return array of size 1
1629       return new byte [][] { c };
1630     } else if(index == c.length - 1) {
1631       // family with empty qualifier, return array size 2
1632       byte [] family = new byte[c.length-1];
1633       System.arraycopy(c, 0, family, 0, family.length);
1634       return new byte [][] { family, HConstants.EMPTY_BYTE_ARRAY};
1635     }
1636     // Family and column, return array size 2
1637     final byte [][] result = new byte [2][];
1638     result[0] = new byte [index];
1639     System.arraycopy(c, 0, result[0], 0, index);
1640     final int len = c.length - (index + 1);
1641     result[1] = new byte[len];
1642     System.arraycopy(c, index + 1 /* Skip delimiter */, result[1], 0, len);
1643     return result;
1644   }
1645 
1646   /**
1647    * Makes a column in family:qualifier form from separate byte arrays.
1648    * <p>
1649    * Not recommended for usage as this is old-style API.
1650    * @param family
1651    * @param qualifier
1652    * @return family:qualifier
1653    */
1654   public static byte [] makeColumn(byte [] family, byte [] qualifier) {
1655     return Bytes.add(family, COLUMN_FAMILY_DELIM_ARRAY, qualifier);
1656   }
1657 
1658   /**
1659    * @param b
1660    * @param delimiter
1661    * @return Index of delimiter having started from start of <code>b</code>
1662    * moving rightward.
1663    */
1664   public static int getDelimiter(final byte [] b, int offset, final int length,
1665       final int delimiter) {
1666     if (b == null) {
1667       throw new IllegalArgumentException("Passed buffer is null");
1668     }
1669     int result = -1;
1670     for (int i = offset; i < length + offset; i++) {
1671       if (b[i] == delimiter) {
1672         result = i;
1673         break;
1674       }
1675     }
1676     return result;
1677   }
1678 
1679   /**
1680    * Find index of passed delimiter walking from end of buffer backwards.
1681    * @param b
1682    * @param delimiter
1683    * @return Index of delimiter
1684    */
1685   public static int getDelimiterInReverse(final byte [] b, final int offset,
1686       final int length, final int delimiter) {
1687     if (b == null) {
1688       throw new IllegalArgumentException("Passed buffer is null");
1689     }
1690     int result = -1;
1691     for (int i = (offset + length) - 1; i >= offset; i--) {
1692       if (b[i] == delimiter) {
1693         result = i;
1694         break;
1695       }
1696     }
1697     return result;
1698   }
1699 
1700   /**
1701    * A {@link KVComparator} for <code>hbase:meta</code> catalog table
1702    * {@link KeyValue}s.
1703    */
1704   public static class MetaComparator extends KVComparator {
1705     /**
1706      * Compare key portion of a {@link KeyValue} for keys in <code>hbase:meta</code>
1707      * table.
1708      */
1709     @Override
1710     public int compare(final Cell left, final Cell right) {
1711       int c = compareRowKey(left, right);
1712       if (c != 0) {
1713         return c;
1714       }
1715       return CellComparator.compareWithoutRow(left, right);
1716     }
1717 
1718     @Override
1719     public int compareOnlyKeyPortion(Cell left, Cell right) {
1720       return compare(left, right);
1721     }
1722 
1723     @Override
1724     public int compareRows(byte [] left, int loffset, int llength,
1725         byte [] right, int roffset, int rlength) {
1726       int leftDelimiter = getDelimiter(left, loffset, llength,
1727           HConstants.DELIMITER);
1728       int rightDelimiter = getDelimiter(right, roffset, rlength,
1729           HConstants.DELIMITER);
1730       // Compare up to the delimiter
1731       int lpart = (leftDelimiter < 0 ? llength :leftDelimiter - loffset);
1732       int rpart = (rightDelimiter < 0 ? rlength :rightDelimiter - roffset);
1733       int result = Bytes.compareTo(left, loffset, lpart, right, roffset, rpart);
1734       if (result != 0) {
1735         return result;
1736       } else {
1737         if (leftDelimiter < 0 && rightDelimiter >= 0) {
1738           return -1;
1739         } else if (rightDelimiter < 0 && leftDelimiter >= 0) {
1740           return 1;
1741         } else if (leftDelimiter < 0 && rightDelimiter < 0) {
1742           return 0;
1743         }
1744       }
1745       // Compare middle bit of the row.
1746       // Move past delimiter
1747       leftDelimiter++;
1748       rightDelimiter++;
1749       int leftFarDelimiter = getDelimiterInReverse(left, leftDelimiter,
1750           llength - (leftDelimiter - loffset), HConstants.DELIMITER);
1751       int rightFarDelimiter = getDelimiterInReverse(right,
1752           rightDelimiter, rlength - (rightDelimiter - roffset),
1753           HConstants.DELIMITER);
1754       // Now compare middlesection of row.
1755       lpart = (leftFarDelimiter < 0 ? llength + loffset: leftFarDelimiter) - leftDelimiter;
1756       rpart = (rightFarDelimiter < 0 ? rlength + roffset: rightFarDelimiter)- rightDelimiter;
1757       result = super.compareRows(left, leftDelimiter, lpart, right, rightDelimiter, rpart);
1758       if (result != 0) {
1759         return result;
1760       }  else {
1761         if (leftDelimiter < 0 && rightDelimiter >= 0) {
1762           return -1;
1763         } else if (rightDelimiter < 0 && leftDelimiter >= 0) {
1764           return 1;
1765         } else if (leftDelimiter < 0 && rightDelimiter < 0) {
1766           return 0;
1767         }
1768       }
1769       // Compare last part of row, the rowid.
1770       leftFarDelimiter++;
1771       rightFarDelimiter++;
1772       result = Bytes.compareTo(left, leftFarDelimiter, llength - (leftFarDelimiter - loffset),
1773           right, rightFarDelimiter, rlength - (rightFarDelimiter - roffset));
1774       return result;
1775     }
1776 
1777     /**
1778      * Don't do any fancy Block Index splitting tricks.
1779      */
1780     @Override
1781     public byte[] getShortMidpointKey(final byte[] leftKey, final byte[] rightKey) {
1782       return Arrays.copyOf(rightKey, rightKey.length);
1783     }
1784 
1785     /**
1786      * The HFileV2 file format's trailer contains this class name.  We reinterpret this and
1787      * instantiate the appropriate comparator.
1788      * TODO: With V3 consider removing this.
1789      * @return legacy class name for FileFileTrailer#comparatorClassName
1790      */
1791     @Override
1792     public String getLegacyKeyComparatorName() {
1793       return "org.apache.hadoop.hbase.KeyValue$MetaKeyComparator";
1794     }
1795 
1796     @Override
1797     protected Object clone() throws CloneNotSupportedException {
1798       return new MetaComparator();
1799     }
1800 
1801     /**
1802      * Override the row key comparison to parse and compare the meta row key parts.
1803      */
1804     @Override
1805     protected int compareRowKey(final Cell l, final Cell r) {
1806       byte[] left = l.getRowArray();
1807       int loffset = l.getRowOffset();
1808       int llength = l.getRowLength();
1809       byte[] right = r.getRowArray();
1810       int roffset = r.getRowOffset();
1811       int rlength = r.getRowLength();
1812       return compareRows(left, loffset, llength, right, roffset, rlength);
1813     }
1814   }
1815 
1816   /**
1817    * Compare KeyValues.  When we compare KeyValues, we only compare the Key
1818    * portion.  This means two KeyValues with same Key but different Values are
1819    * considered the same as far as this Comparator is concerned.
1820    */
1821   public static class KVComparator implements RawComparator<Cell>, SamePrefixComparator<byte[]> {
1822 
1823     /**
1824      * The HFileV2 file format's trailer contains this class name.  We reinterpret this and
1825      * instantiate the appropriate comparator.
1826      * TODO: With V3 consider removing this.
1827      * @return legacy class name for FileFileTrailer#comparatorClassName
1828      */
1829     public String getLegacyKeyComparatorName() {
1830       return "org.apache.hadoop.hbase.KeyValue$KeyComparator";
1831     }
1832 
1833     @Override // RawComparator
1834     public int compare(byte[] l, int loff, int llen, byte[] r, int roff, int rlen) {
1835       return compareFlatKey(l,loff,llen, r,roff,rlen);
1836     }
1837 
1838     
1839     /**
1840      * Compares the only the user specified portion of a Key.  This is overridden by MetaComparator.
1841      * @param left
1842      * @param right
1843      * @return 0 if equal, <0 if left smaller, >0 if right smaller
1844      */
1845     protected int compareRowKey(final Cell left, final Cell right) {
1846       return CellComparator.compareRows(left, right);
1847     }
1848 
1849     /**
1850      * Compares left to right assuming that left,loffset,llength and right,roffset,rlength are
1851      * full KVs laid out in a flat byte[]s.
1852      * @param left
1853      * @param loffset
1854      * @param llength
1855      * @param right
1856      * @param roffset
1857      * @param rlength
1858      * @return  0 if equal, <0 if left smaller, >0 if right smaller
1859      */
1860     public int compareFlatKey(byte[] left, int loffset, int llength,
1861         byte[] right, int roffset, int rlength) {
1862       // Compare row
1863       short lrowlength = Bytes.toShort(left, loffset);
1864       short rrowlength = Bytes.toShort(right, roffset);
1865       int compare = compareRows(left, loffset + Bytes.SIZEOF_SHORT,
1866           lrowlength, right, roffset + Bytes.SIZEOF_SHORT, rrowlength);
1867       if (compare != 0) {
1868         return compare;
1869       }
1870 
1871       // Compare the rest of the two KVs without making any assumptions about
1872       // the common prefix. This function will not compare rows anyway, so we
1873       // don't need to tell it that the common prefix includes the row.
1874       return compareWithoutRow(0, left, loffset, llength, right, roffset,
1875           rlength, rrowlength);
1876     }
1877 
1878     public int compareFlatKey(byte[] left, byte[] right) {
1879       return compareFlatKey(left, 0, left.length, right, 0, right.length);
1880     }
1881 
1882     public int compareOnlyKeyPortion(Cell left, Cell right) {
1883       return CellComparator.compareStatic(left, right, true);
1884     }
1885 
1886     /**
1887      * Compares the Key of a cell -- with fields being more significant in this order:
1888      * rowkey, colfam/qual, timestamp, type, mvcc
1889      */
1890     @Override
1891     public int compare(final Cell left, final Cell right) {
1892       int compare = CellComparator.compareStatic(left, right, false);
1893       return compare;
1894     }
1895 
1896     public int compareTimestamps(final Cell left, final Cell right) {
1897       return CellComparator.compareTimestamps(left, right);
1898     }
1899 
1900     /**
1901      * @param left
1902      * @param right
1903      * @return Result comparing rows.
1904      */
1905     public int compareRows(final Cell left, final Cell right) {
1906       return compareRows(left.getRowArray(),left.getRowOffset(), left.getRowLength(),
1907       right.getRowArray(), right.getRowOffset(), right.getRowLength());
1908     }
1909 
1910     /**
1911      * Get the b[],o,l for left and right rowkey portions and compare.
1912      * @param left
1913      * @param loffset
1914      * @param llength
1915      * @param right
1916      * @param roffset
1917      * @param rlength
1918      * @return 0 if equal, <0 if left smaller, >0 if right smaller
1919      */
1920     public int compareRows(byte [] left, int loffset, int llength,
1921         byte [] right, int roffset, int rlength) {
1922       return Bytes.compareTo(left, loffset, llength, right, roffset, rlength);
1923     }
1924 
1925     int compareColumns(final Cell left, final short lrowlength, final Cell right,
1926         final short rrowlength) {
1927       return CellComparator.compareColumns(left, right);
1928     }
1929 
1930     protected int compareColumns(
1931         byte [] left, int loffset, int llength, final int lfamilylength,
1932         byte [] right, int roffset, int rlength, final int rfamilylength) {
1933       // Compare family portion first.
1934       int diff = Bytes.compareTo(left, loffset, lfamilylength,
1935         right, roffset, rfamilylength);
1936       if (diff != 0) {
1937         return diff;
1938       }
1939       // Compare qualifier portion
1940       return Bytes.compareTo(left, loffset + lfamilylength,
1941         llength - lfamilylength,
1942         right, roffset + rfamilylength, rlength - rfamilylength);
1943       }
1944 
1945     static int compareTimestamps(final long ltimestamp, final long rtimestamp) {
1946       // The below older timestamps sorting ahead of newer timestamps looks
1947       // wrong but it is intentional. This way, newer timestamps are first
1948       // found when we iterate over a memstore and newer versions are the
1949       // first we trip over when reading from a store file.
1950       if (ltimestamp < rtimestamp) {
1951         return 1;
1952       } else if (ltimestamp > rtimestamp) {
1953         return -1;
1954       }
1955       return 0;
1956     }
1957 
1958     /**
1959      * Overridden
1960      * @param commonPrefix
1961      * @param left
1962      * @param loffset
1963      * @param llength
1964      * @param right
1965      * @param roffset
1966      * @param rlength
1967      * @return 0 if equal, <0 if left smaller, >0 if right smaller
1968      */
1969     @Override // SamePrefixComparator
1970     public int compareIgnoringPrefix(int commonPrefix, byte[] left,
1971         int loffset, int llength, byte[] right, int roffset, int rlength) {
1972       // Compare row
1973       short lrowlength = Bytes.toShort(left, loffset);
1974       short rrowlength;
1975 
1976       int comparisonResult = 0;
1977       if (commonPrefix < ROW_LENGTH_SIZE) {
1978         // almost nothing in common
1979         rrowlength = Bytes.toShort(right, roffset);
1980         comparisonResult = compareRows(left, loffset + ROW_LENGTH_SIZE,
1981             lrowlength, right, roffset + ROW_LENGTH_SIZE, rrowlength);
1982       } else { // the row length is the same
1983         rrowlength = lrowlength;
1984         if (commonPrefix < ROW_LENGTH_SIZE + rrowlength) {
1985           // The rows are not the same. Exclude the common prefix and compare
1986           // the rest of the two rows.
1987           int common = commonPrefix - ROW_LENGTH_SIZE;
1988           comparisonResult = compareRows(
1989               left, loffset + common + ROW_LENGTH_SIZE, lrowlength - common,
1990               right, roffset + common + ROW_LENGTH_SIZE, rrowlength - common);
1991         }
1992       }
1993       if (comparisonResult != 0) {
1994         return comparisonResult;
1995       }
1996 
1997       assert lrowlength == rrowlength;
1998       return compareWithoutRow(commonPrefix, left, loffset, llength, right,
1999           roffset, rlength, lrowlength);
2000     }
2001 
2002     /**
2003      * Compare columnFamily, qualifier, timestamp, and key type (everything
2004      * except the row). This method is used both in the normal comparator and
2005      * the "same-prefix" comparator. Note that we are assuming that row portions
2006      * of both KVs have already been parsed and found identical, and we don't
2007      * validate that assumption here.
2008      * @param commonPrefix
2009      *          the length of the common prefix of the two key-values being
2010      *          compared, including row length and row
2011      */
2012     private int compareWithoutRow(int commonPrefix, byte[] left, int loffset,
2013         int llength, byte[] right, int roffset, int rlength, short rowlength) {
2014       /***
2015        * KeyValue Format and commonLength:
2016        * |_keyLen_|_valLen_|_rowLen_|_rowKey_|_famiLen_|_fami_|_Quali_|....
2017        * ------------------|-------commonLength--------|--------------
2018        */
2019       int commonLength = ROW_LENGTH_SIZE + FAMILY_LENGTH_SIZE + rowlength;
2020 
2021       // commonLength + TIMESTAMP_TYPE_SIZE
2022       int commonLengthWithTSAndType = TIMESTAMP_TYPE_SIZE + commonLength;
2023       // ColumnFamily + Qualifier length.
2024       int lcolumnlength = llength - commonLengthWithTSAndType;
2025       int rcolumnlength = rlength - commonLengthWithTSAndType;
2026 
2027       byte ltype = left[loffset + (llength - 1)];
2028       byte rtype = right[roffset + (rlength - 1)];
2029 
2030       // If the column is not specified, the "minimum" key type appears the
2031       // latest in the sorted order, regardless of the timestamp. This is used
2032       // for specifying the last key/value in a given row, because there is no
2033       // "lexicographically last column" (it would be infinitely long). The
2034       // "maximum" key type does not need this behavior.
2035       if (lcolumnlength == 0 && ltype == Type.Minimum.getCode()) {
2036         // left is "bigger", i.e. it appears later in the sorted order
2037         return 1;
2038       }
2039       if (rcolumnlength == 0 && rtype == Type.Minimum.getCode()) {
2040         return -1;
2041       }
2042 
2043       int lfamilyoffset = commonLength + loffset;
2044       int rfamilyoffset = commonLength + roffset;
2045 
2046       // Column family length.
2047       int lfamilylength = left[lfamilyoffset - 1];
2048       int rfamilylength = right[rfamilyoffset - 1];
2049       // If left family size is not equal to right family size, we need not
2050       // compare the qualifiers.
2051       boolean sameFamilySize = (lfamilylength == rfamilylength);
2052       int common = 0;
2053       if (commonPrefix > 0) {
2054         common = Math.max(0, commonPrefix - commonLength);
2055         if (!sameFamilySize) {
2056           // Common should not be larger than Math.min(lfamilylength,
2057           // rfamilylength).
2058           common = Math.min(common, Math.min(lfamilylength, rfamilylength));
2059         } else {
2060           common = Math.min(common, Math.min(lcolumnlength, rcolumnlength));
2061         }
2062       }
2063       if (!sameFamilySize) {
2064         // comparing column family is enough.
2065         return Bytes.compareTo(left, lfamilyoffset + common, lfamilylength
2066             - common, right, rfamilyoffset + common, rfamilylength - common);
2067       }
2068       // Compare family & qualifier together.
2069       final int comparison = Bytes.compareTo(left, lfamilyoffset + common,
2070           lcolumnlength - common, right, rfamilyoffset + common,
2071           rcolumnlength - common);
2072       if (comparison != 0) {
2073         return comparison;
2074       }
2075 
2076       ////
2077       // Next compare timestamps.
2078       long ltimestamp = Bytes.toLong(left,
2079           loffset + (llength - TIMESTAMP_TYPE_SIZE));
2080       long rtimestamp = Bytes.toLong(right,
2081           roffset + (rlength - TIMESTAMP_TYPE_SIZE));
2082       int compare = compareTimestamps(ltimestamp, rtimestamp);
2083       if (compare != 0) {
2084         return compare;
2085       }
2086 
2087       // Compare types. Let the delete types sort ahead of puts; i.e. types
2088       // of higher numbers sort before those of lesser numbers. Maximum (255)
2089       // appears ahead of everything, and minimum (0) appears after
2090       // everything.
2091       return (0xff & rtype) - (0xff & ltype);
2092     }
2093 
2094     protected int compareFamilies(final byte[] left, final int loffset, final int lfamilylength,
2095         final byte[] right, final int roffset, final int rfamilylength) {
2096       int diff = Bytes.compareTo(left, loffset, lfamilylength, right, roffset, rfamilylength);
2097       return diff;
2098     }
2099 
2100     protected int compareColumns(final byte[] left, final int loffset, final int lquallength,
2101         final byte[] right, final int roffset, final int rquallength) {
2102       int diff = Bytes.compareTo(left, loffset, lquallength, right, roffset, rquallength);
2103       return diff;
2104     }
2105     /**
2106      * Compares the row and column of two keyvalues for equality
2107      * @param left
2108      * @param right
2109      * @return True if same row and column.
2110      */
2111     public boolean matchingRowColumn(final Cell left,
2112         final Cell right) {
2113       short lrowlength = left.getRowLength();
2114       short rrowlength = right.getRowLength();
2115 
2116       // TsOffset = end of column data. just comparing Row+CF length of each
2117       if ((left.getRowLength() + left.getFamilyLength() + left.getQualifierLength()) != (right
2118           .getRowLength() + right.getFamilyLength() + right.getQualifierLength())) {
2119         return false;
2120       }
2121 
2122       if (!matchingRows(left, lrowlength, right, rrowlength)) {
2123         return false;
2124       }
2125 
2126       int lfoffset = left.getFamilyOffset();
2127       int rfoffset = right.getFamilyOffset();
2128       int lclength = left.getQualifierLength();
2129       int rclength = right.getQualifierLength();
2130       int lfamilylength = left.getFamilyLength();
2131       int rfamilylength = right.getFamilyLength();
2132       int diff = compareFamilies(left.getFamilyArray(), lfoffset, lfamilylength,
2133           right.getFamilyArray(), rfoffset, rfamilylength);
2134       if (diff != 0) {
2135         return false;
2136       } else {
2137         diff = compareColumns(left.getQualifierArray(), left.getQualifierOffset(), lclength,
2138             right.getQualifierArray(), right.getQualifierOffset(), rclength);
2139         return diff == 0;
2140       }
2141     }
2142 
2143     /**
2144      * Compares the row of two keyvalues for equality
2145      * @param left
2146      * @param right
2147      * @return True if rows match.
2148      */
2149     public boolean matchingRows(final Cell left, final Cell right) {
2150       short lrowlength = left.getRowLength();
2151       short rrowlength = right.getRowLength();
2152       return matchingRows(left, lrowlength, right, rrowlength);
2153     }
2154 
2155     /**
2156      * @param left
2157      * @param lrowlength
2158      * @param right
2159      * @param rrowlength
2160      * @return True if rows match.
2161      */
2162     private boolean matchingRows(final Cell left, final short lrowlength,
2163         final Cell right, final short rrowlength) {
2164       return lrowlength == rrowlength &&
2165           matchingRows(left.getRowArray(), left.getRowOffset(), lrowlength,
2166               right.getRowArray(), right.getRowOffset(), rrowlength);
2167     }
2168 
2169     /**
2170      * Compare rows. Just calls Bytes.equals, but it's good to have this encapsulated.
2171      * @param left Left row array.
2172      * @param loffset Left row offset.
2173      * @param llength Left row length.
2174      * @param right Right row array.
2175      * @param roffset Right row offset.
2176      * @param rlength Right row length.
2177      * @return Whether rows are the same row.
2178      */
2179     public boolean matchingRows(final byte [] left, final int loffset, final int llength,
2180         final byte [] right, final int roffset, final int rlength) {
2181       return Bytes.equals(left, loffset, llength, right, roffset, rlength);
2182     }
2183 
2184     public byte[] calcIndexKey(byte[] lastKeyOfPreviousBlock, byte[] firstKeyInBlock) {
2185       byte[] fakeKey = getShortMidpointKey(lastKeyOfPreviousBlock, firstKeyInBlock);
2186       if (compareFlatKey(fakeKey, firstKeyInBlock) > 0) {
2187         LOG.error("Unexpected getShortMidpointKey result, fakeKey:"
2188             + Bytes.toStringBinary(fakeKey) + ", firstKeyInBlock:"
2189             + Bytes.toStringBinary(firstKeyInBlock));
2190         return firstKeyInBlock;
2191       }
2192       if (lastKeyOfPreviousBlock != null && compareFlatKey(lastKeyOfPreviousBlock, fakeKey) >= 0) {
2193         LOG.error("Unexpected getShortMidpointKey result, lastKeyOfPreviousBlock:" +
2194             Bytes.toStringBinary(lastKeyOfPreviousBlock) + ", fakeKey:" +
2195             Bytes.toStringBinary(fakeKey));
2196         return firstKeyInBlock;
2197       }
2198       return fakeKey;
2199     }
2200 
2201     /**
2202      * This is a HFile block index key optimization.
2203      * @param leftKey
2204      * @param rightKey
2205      * @return 0 if equal, <0 if left smaller, >0 if right smaller
2206      */
2207     public byte[] getShortMidpointKey(final byte[] leftKey, final byte[] rightKey) {
2208       if (rightKey == null) {
2209         throw new IllegalArgumentException("rightKey can not be null");
2210       }
2211       if (leftKey == null) {
2212         return Arrays.copyOf(rightKey, rightKey.length);
2213       }
2214       if (compareFlatKey(leftKey, rightKey) >= 0) {
2215         throw new IllegalArgumentException("Unexpected input, leftKey:" + Bytes.toString(leftKey)
2216           + ", rightKey:" + Bytes.toString(rightKey));
2217       }
2218 
2219       short leftRowLength = Bytes.toShort(leftKey, 0);
2220       short rightRowLength = Bytes.toShort(rightKey, 0);
2221       int leftCommonLength = ROW_LENGTH_SIZE + FAMILY_LENGTH_SIZE + leftRowLength;
2222       int rightCommonLength = ROW_LENGTH_SIZE + FAMILY_LENGTH_SIZE + rightRowLength;
2223       int leftCommonLengthWithTSAndType = TIMESTAMP_TYPE_SIZE + leftCommonLength;
2224       int rightCommonLengthWithTSAndType = TIMESTAMP_TYPE_SIZE + rightCommonLength;
2225       int leftColumnLength = leftKey.length - leftCommonLengthWithTSAndType;
2226       int rightColumnLength = rightKey.length - rightCommonLengthWithTSAndType;
2227       // rows are equal
2228       if (leftRowLength == rightRowLength && compareRows(leftKey, ROW_LENGTH_SIZE, leftRowLength,
2229         rightKey, ROW_LENGTH_SIZE, rightRowLength) == 0) {
2230         // Compare family & qualifier together.
2231         int comparison = Bytes.compareTo(leftKey, leftCommonLength, leftColumnLength, rightKey,
2232           rightCommonLength, rightColumnLength);
2233         // same with "row + family + qualifier", return rightKey directly
2234         if (comparison == 0) {
2235           return Arrays.copyOf(rightKey, rightKey.length);
2236         }
2237         // "family + qualifier" are different, generate a faked key per rightKey
2238         byte[] newKey = Arrays.copyOf(rightKey, rightKey.length);
2239         Bytes.putLong(newKey, rightKey.length - TIMESTAMP_TYPE_SIZE, HConstants.LATEST_TIMESTAMP);
2240         Bytes.putByte(newKey, rightKey.length - TYPE_SIZE, Type.Maximum.getCode());
2241         return newKey;
2242       }
2243       // rows are different
2244       short minLength = leftRowLength < rightRowLength ? leftRowLength : rightRowLength;
2245       short diffIdx = 0;
2246       while (diffIdx < minLength
2247           && leftKey[ROW_LENGTH_SIZE + diffIdx] == rightKey[ROW_LENGTH_SIZE + diffIdx]) {
2248         diffIdx++;
2249       }
2250       byte[] newRowKey = null;
2251       if (diffIdx >= minLength) {
2252         // leftKey's row is prefix of rightKey's.
2253         newRowKey = new byte[diffIdx + 1];
2254         System.arraycopy(rightKey, ROW_LENGTH_SIZE, newRowKey, 0, diffIdx + 1);
2255       } else {
2256         int diffByte = leftKey[ROW_LENGTH_SIZE + diffIdx];
2257         if ((0xff & diffByte) < 0xff && (diffByte + 1) <
2258             (rightKey[ROW_LENGTH_SIZE + diffIdx] & 0xff)) {
2259           newRowKey = new byte[diffIdx + 1];
2260           System.arraycopy(leftKey, ROW_LENGTH_SIZE, newRowKey, 0, diffIdx);
2261           newRowKey[diffIdx] = (byte) (diffByte + 1);
2262         } else {
2263           newRowKey = new byte[diffIdx + 1];
2264           System.arraycopy(rightKey, ROW_LENGTH_SIZE, newRowKey, 0, diffIdx + 1);
2265         }
2266       }
2267       return new KeyValue(newRowKey, null, null, HConstants.LATEST_TIMESTAMP,
2268         Type.Maximum).getKey();
2269     }
2270 
2271     @Override
2272     protected Object clone() throws CloneNotSupportedException {
2273       return new KVComparator();
2274     }
2275 
2276   }
2277 
2278   /**
2279    * @param b
2280    * @return A KeyValue made of a byte array that holds the key-only part.
2281    * Needed to convert hfile index members to KeyValues.
2282    */
2283   public static KeyValue createKeyValueFromKey(final byte [] b) {
2284     return createKeyValueFromKey(b, 0, b.length);
2285   }
2286 
2287   /**
2288    * @param bb
2289    * @return A KeyValue made of a byte buffer that holds the key-only part.
2290    * Needed to convert hfile index members to KeyValues.
2291    */
2292   public static KeyValue createKeyValueFromKey(final ByteBuffer bb) {
2293     return createKeyValueFromKey(bb.array(), bb.arrayOffset(), bb.limit());
2294   }
2295 
2296   /**
2297    * @param b
2298    * @param o
2299    * @param l
2300    * @return A KeyValue made of a byte array that holds the key-only part.
2301    * Needed to convert hfile index members to KeyValues.
2302    */
2303   public static KeyValue createKeyValueFromKey(final byte [] b, final int o,
2304       final int l) {
2305     byte [] newb = new byte[l + ROW_OFFSET];
2306     System.arraycopy(b, o, newb, ROW_OFFSET, l);
2307     Bytes.putInt(newb, 0, l);
2308     Bytes.putInt(newb, Bytes.SIZEOF_INT, 0);
2309     return new KeyValue(newb);
2310   }
2311 
2312   /**
2313    * @param in Where to read bytes from.  Creates a byte array to hold the KeyValue
2314    * backing bytes copied from the steam.
2315    * @return KeyValue created by deserializing from <code>in</code> OR if we find a length
2316    * of zero, we will return null which can be useful marking a stream as done.
2317    * @throws IOException
2318    */
2319   public static KeyValue create(final DataInput in) throws IOException {
2320     return create(in.readInt(), in);
2321   }
2322 
2323   /**
2324    * Create a KeyValue reading <code>length</code> from <code>in</code>
2325    * @param length
2326    * @param in
2327    * @return Created KeyValue OR if we find a length of zero, we will return null which
2328    * can be useful marking a stream as done.
2329    * @throws IOException
2330    */
2331   public static KeyValue create(int length, final DataInput in) throws IOException {
2332 
2333     if (length <= 0) {
2334       if (length == 0) return null;
2335       throw new IOException("Failed read " + length + " bytes, stream corrupt?");
2336     }
2337 
2338     // This is how the old Writables.readFrom used to deserialize.  Didn't even vint.
2339     byte [] bytes = new byte[length];
2340     in.readFully(bytes);
2341     return new KeyValue(bytes, 0, length);
2342   }
2343   
2344   /**
2345    * Create a new KeyValue by copying existing cell and adding new tags
2346    * @param c
2347    * @param newTags
2348    * @return a new KeyValue instance with new tags
2349    */
2350   public static KeyValue cloneAndAddTags(Cell c, List<Tag> newTags) {
2351     List<Tag> existingTags = null;
2352     if(c.getTagsLength() > 0) {
2353       existingTags = Tag.asList(c.getTagsArray(), c.getTagsOffset(), c.getTagsLength());
2354       existingTags.addAll(newTags);
2355     } else {
2356       existingTags = newTags;
2357     }
2358     return new KeyValue(c.getRowArray(), c.getRowOffset(), (int)c.getRowLength(),
2359       c.getFamilyArray(), c.getFamilyOffset(), (int)c.getFamilyLength(), 
2360       c.getQualifierArray(), c.getQualifierOffset(), (int) c.getQualifierLength(), 
2361       c.getTimestamp(), Type.codeToType(c.getTypeByte()), c.getValueArray(), c.getValueOffset(), 
2362       c.getValueLength(), existingTags);
2363   }
2364 
2365   /**
2366    * Create a KeyValue reading from the raw InputStream.
2367    * Named <code>iscreate</code> so doesn't clash with {@link #create(DataInput)}
2368    * @param in
2369    * @return Created KeyValue OR if we find a length of zero, we will return null which
2370    * can be useful marking a stream as done.
2371    * @throws IOException
2372    */
2373   public static KeyValue iscreate(final InputStream in) throws IOException {
2374     byte [] intBytes = new byte[Bytes.SIZEOF_INT];
2375     int bytesRead = 0;
2376     while (bytesRead < intBytes.length) {
2377       int n = in.read(intBytes, bytesRead, intBytes.length - bytesRead);
2378       if (n < 0) {
2379         if (bytesRead == 0) return null; // EOF at start is ok
2380         throw new IOException("Failed read of int, read " + bytesRead + " bytes");
2381       }
2382       bytesRead += n;
2383     }
2384     // TODO: perhaps some sanity check is needed here.
2385     byte [] bytes = new byte[Bytes.toInt(intBytes)];
2386     IOUtils.readFully(in, bytes, 0, bytes.length);
2387     return new KeyValue(bytes, 0, bytes.length);
2388   }
2389 
2390   /**
2391    * Write out a KeyValue in the manner in which we used to when KeyValue was a Writable.
2392    * @param kv
2393    * @param out
2394    * @return Length written on stream
2395    * @throws IOException
2396    * @see #create(DataInput) for the inverse function
2397    */
2398   public static long write(final KeyValue kv, final DataOutput out) throws IOException {
2399     // This is how the old Writables write used to serialize KVs.  Need to figure way to make it
2400     // work for all implementations.
2401     int length = kv.getLength();
2402     out.writeInt(length);
2403     out.write(kv.getBuffer(), kv.getOffset(), length);
2404     return length + Bytes.SIZEOF_INT;
2405   }
2406 
2407   /**
2408    * Write out a KeyValue in the manner in which we used to when KeyValue was a Writable but do
2409    * not require a {@link DataOutput}, just take plain {@link OutputStream}
2410    * Named <code>oswrite</code> so does not clash with {@link #write(KeyValue, DataOutput)}
2411    * @param kv
2412    * @param out
2413    * @return Length written on stream
2414    * @throws IOException
2415    * @see #create(DataInput) for the inverse function
2416    * @see #write(KeyValue, DataOutput)
2417    * @deprecated use {@link #oswrite(KeyValue, OutputStream, boolean)} instead
2418    */
2419   @Deprecated
2420   public static long oswrite(final KeyValue kv, final OutputStream out)
2421       throws IOException {
2422     int length = kv.getLength();
2423     // This does same as DataOuput#writeInt (big-endian, etc.)
2424     out.write(Bytes.toBytes(length));
2425     out.write(kv.getBuffer(), kv.getOffset(), length);
2426     return length + Bytes.SIZEOF_INT;
2427   }
2428 
2429   /**
2430    * Write out a KeyValue in the manner in which we used to when KeyValue was a Writable but do
2431    * not require a {@link DataOutput}, just take plain {@link OutputStream}
2432    * Named <code>oswrite</code> so does not clash with {@link #write(KeyValue, DataOutput)}
2433    * @param kv
2434    * @param out
2435    * @param withTags
2436    * @return Length written on stream
2437    * @throws IOException
2438    * @see #create(DataInput) for the inverse function
2439    * @see #write(KeyValue, DataOutput)
2440    * @see KeyValueUtil#oswrite(Cell, OutputStream, boolean)
2441    */
2442   public static long oswrite(final KeyValue kv, final OutputStream out, final boolean withTags)
2443       throws IOException {
2444     // In KeyValueUtil#oswrite we do a Cell serialization as KeyValue. Any changes doing here, pls
2445     // check KeyValueUtil#oswrite also and do necessary changes.
2446     int length = kv.getLength();
2447     if (!withTags) {
2448       length = kv.getKeyLength() + kv.getValueLength() + KEYVALUE_INFRASTRUCTURE_SIZE;
2449     }
2450     // This does same as DataOuput#writeInt (big-endian, etc.)
2451     StreamUtils.writeInt(out, length);
2452     out.write(kv.getBuffer(), kv.getOffset(), length);
2453     return length + Bytes.SIZEOF_INT;
2454   }
2455 
2456   /**
2457    * Comparator that compares row component only of a KeyValue.
2458    */
2459   public static class RowOnlyComparator implements Comparator<KeyValue> {
2460     final KVComparator comparator;
2461 
2462     public RowOnlyComparator(final KVComparator c) {
2463       this.comparator = c;
2464     }
2465 
2466     public int compare(KeyValue left, KeyValue right) {
2467       return comparator.compareRows(left, right);
2468     }
2469   }
2470 
2471 
2472   /**
2473    * Avoids redundant comparisons for better performance.
2474    * 
2475    * TODO get rid of this wart
2476    */
2477   public interface SamePrefixComparator<T> {
2478     /**
2479      * Compare two keys assuming that the first n bytes are the same.
2480      * @param commonPrefix How many bytes are the same.
2481      */
2482     int compareIgnoringPrefix(
2483       int commonPrefix, byte[] left, int loffset, int llength, byte[] right, int roffset, int rlength
2484     );
2485   }
2486 
2487   /**
2488    * This is a TEST only Comparator used in TestSeekTo and TestReseekTo.
2489    */
2490   public static class RawBytesComparator extends KVComparator {
2491     /**
2492      * The HFileV2 file format's trailer contains this class name.  We reinterpret this and
2493      * instantiate the appropriate comparator.
2494      * TODO: With V3 consider removing this.
2495      * @return legacy class name for FileFileTrailer#comparatorClassName
2496      */
2497     public String getLegacyKeyComparatorName() {
2498       return "org.apache.hadoop.hbase.util.Bytes$ByteArrayComparator";
2499     }
2500 
2501     public int compareFlatKey(byte[] left, int loffset, int llength, byte[] right,
2502         int roffset, int rlength) {
2503       return Bytes.BYTES_RAWCOMPARATOR.compare(left,  loffset, llength, right, roffset, rlength);
2504     }
2505 
2506     @Override
2507     public int compare(Cell left, Cell right) {
2508       return compareOnlyKeyPortion(left, right);
2509     }
2510 
2511     @VisibleForTesting
2512     public int compareOnlyKeyPortion(Cell left, Cell right) {
2513       int c = Bytes.BYTES_RAWCOMPARATOR.compare(left.getRowArray(), left.getRowOffset(),
2514           left.getRowLength(), right.getRowArray(), right.getRowOffset(), right.getRowLength());
2515       if (c != 0) {
2516         return c;
2517       }
2518       c = Bytes.BYTES_RAWCOMPARATOR.compare(left.getFamilyArray(), left.getFamilyOffset(),
2519           left.getFamilyLength(), right.getFamilyArray(), right.getFamilyOffset(),
2520           right.getFamilyLength());
2521       if (c != 0) {
2522         return c;
2523       }
2524       c = Bytes.BYTES_RAWCOMPARATOR.compare(left.getQualifierArray(), left.getQualifierOffset(),
2525           left.getQualifierLength(), right.getQualifierArray(), right.getQualifierOffset(),
2526           right.getQualifierLength());
2527       if (c != 0) {
2528         return c;
2529       }
2530       c = compareTimestamps(left.getTimestamp(), right.getTimestamp());
2531       if (c != 0) {
2532         return c;
2533       }
2534       return (0xff & left.getTypeByte()) - (0xff & right.getTypeByte());
2535     }
2536 
2537     public byte[] calcIndexKey(byte[] lastKeyOfPreviousBlock, byte[] firstKeyInBlock) {
2538       return firstKeyInBlock;
2539     }
2540 
2541   }
2542 
2543   /**
2544    * HeapSize implementation
2545    *
2546    * We do not count the bytes in the rowCache because it should be empty for a KeyValue in the
2547    * MemStore.
2548    */
2549   @Override
2550   public long heapSize() {
2551     int sum = 0;
2552     sum += ClassSize.OBJECT;// the KeyValue object itself
2553     sum += ClassSize.REFERENCE;// pointer to "bytes"
2554     sum += ClassSize.align(ClassSize.ARRAY);// "bytes"
2555     sum += ClassSize.align(length);// number of bytes of data in the "bytes" array
2556     sum += 2 * Bytes.SIZEOF_INT;// offset, length
2557     sum += Bytes.SIZEOF_LONG;// memstoreTS
2558     return ClassSize.align(sum);
2559   }
2560 
2561   /**
2562    * A simple form of KeyValue that creates a keyvalue with only the key part of the byte[]
2563    * Mainly used in places where we need to compare two cells.  Avoids copying of bytes
2564    * In places like block index keys, we need to compare the key byte[] with a cell.
2565    * Hence create a Keyvalue(aka Cell) that would help in comparing as two cells
2566    */
2567   public static class KeyOnlyKeyValue extends KeyValue {
2568     private int length = 0;
2569     private int offset = 0;
2570     private byte[] b;
2571 
2572     public KeyOnlyKeyValue() {
2573 
2574     }
2575 
2576     public KeyOnlyKeyValue(byte[] b, int offset, int length) {
2577       this.b = b;
2578       this.length = length;
2579       this.offset = offset;
2580     }
2581 
2582     @Override
2583     public int getKeyOffset() {
2584       return this.offset;
2585     }
2586 
2587     /**
2588      * A setter that helps to avoid object creation every time and whenever
2589      * there is a need to create new KeyOnlyKeyValue.
2590      * @param key
2591      * @param offset
2592      * @param length
2593      */
2594     public void setKey(byte[] key, int offset, int length) {
2595       this.b = key;
2596       this.offset = offset;
2597       this.length = length;
2598     }
2599 
2600     @Override
2601     public byte[] getKey() {
2602       int keylength = getKeyLength();
2603       byte[] key = new byte[keylength];
2604       System.arraycopy(this.b, getKeyOffset(), key, 0, keylength);
2605       return key;
2606     }
2607 
2608     @Override
2609     public byte[] getRowArray() {
2610       return b;
2611     }
2612 
2613     @Override
2614     public int getRowOffset() {
2615       return getKeyOffset() + Bytes.SIZEOF_SHORT;
2616     }
2617 
2618     @Override
2619     public byte[] getFamilyArray() {
2620       return b;
2621     }
2622 
2623     @Override
2624     public byte getFamilyLength() {
2625       return this.b[getFamilyOffset() - 1];
2626     }
2627 
2628     @Override
2629     public int getFamilyOffset() {
2630       return this.offset + Bytes.SIZEOF_SHORT + getRowLength() + Bytes.SIZEOF_BYTE;
2631     }
2632 
2633     @Override
2634     public byte[] getQualifierArray() {
2635       return b;
2636     }
2637 
2638     @Override
2639     public int getQualifierLength() {
2640       return getQualifierLength(getRowLength(), getFamilyLength());
2641     }
2642 
2643     @Override
2644     public int getQualifierOffset() {
2645       return getFamilyOffset() + getFamilyLength();
2646     }
2647 
2648     @Override
2649     public int getKeyLength() {
2650       return length;
2651     }
2652 
2653     @Override
2654     public short getRowLength() {
2655       return Bytes.toShort(this.b, getKeyOffset());
2656     }
2657 
2658     @Override
2659     public byte getTypeByte() {
2660       return this.b[this.offset + getKeyLength() - 1];
2661     }
2662 
2663     private int getQualifierLength(int rlength, int flength) {
2664       return getKeyLength() - (int) getKeyDataStructureSize(rlength, flength, 0);
2665     }
2666 
2667     @Override
2668     public long getTimestamp() {
2669       int tsOffset = getTimestampOffset();
2670       return Bytes.toLong(this.b, tsOffset);
2671     }
2672 
2673     @Override
2674     public int getTimestampOffset() {
2675       return getKeyOffset() + getKeyLength() - TIMESTAMP_TYPE_SIZE;
2676     }
2677 
2678     @Override
2679     public byte[] getTagsArray() {
2680       return HConstants.EMPTY_BYTE_ARRAY;
2681     }
2682 
2683     @Override
2684     public int getTagsOffset() {
2685       return 0;
2686     }
2687 
2688     @Override
2689     public byte[] getValueArray() {
2690       throw new IllegalArgumentException("KeyOnlyKeyValue does not work with values.");
2691     }
2692 
2693     @Override
2694     public int getValueOffset() {
2695       throw new IllegalArgumentException("KeyOnlyKeyValue does not work with values.");
2696     }
2697 
2698     @Override
2699     public int getValueLength() {
2700       throw new IllegalArgumentException("KeyOnlyKeyValue does not work with values.");
2701     }
2702 
2703     @Override
2704     public int getTagsLength() {
2705       return 0;
2706     }
2707 
2708     @Override
2709     public String toString() {
2710       if (this.b == null || this.b.length == 0) {
2711         return "empty";
2712       }
2713       return keyToString(this.b, this.offset, getKeyLength()) + "/vlen=0/mvcc=0";
2714     }
2715 
2716     @Override
2717     public int hashCode() {
2718       return super.hashCode();
2719     }
2720 
2721     @Override
2722     public boolean equals(Object other) {
2723       return super.equals(other);
2724     }
2725   }
2726 }