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