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