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