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   @Override
1068   public int hashCode() {
1069     byte[] b = getBuffer();
1070     int start = getOffset(), end = getOffset() + getLength();
1071     int h = b[start++];
1072     for (int i = start; i < end; i++) {
1073       h = (h * 13) ^ b[i];
1074     }
1075     return h;
1076   }
1077 
1078   //---------------------------------------------------------------------------
1079   //
1080   //  KeyValue cloning
1081   //
1082   //---------------------------------------------------------------------------
1083 
1084   /**
1085    * Clones a KeyValue.  This creates a copy, re-allocating the buffer.
1086    * @return Fully copied clone of this KeyValue
1087    * @throws CloneNotSupportedException
1088    */
1089   @Override
1090   public KeyValue clone() throws CloneNotSupportedException {
1091     super.clone();
1092     byte [] b = new byte[this.length];
1093     System.arraycopy(this.bytes, this.offset, b, 0, this.length);
1094     KeyValue ret = new KeyValue(b, 0, b.length);
1095     // Important to clone the memstoreTS as well - otherwise memstore's
1096     // update-in-place methods (eg increment) will end up creating
1097     // new entries
1098     ret.setSequenceId(seqId);
1099     return ret;
1100   }
1101 
1102   /**
1103    * Creates a shallow copy of this KeyValue, reusing the data byte buffer.
1104    * http://en.wikipedia.org/wiki/Object_copy
1105    * @return Shallow copy of this KeyValue
1106    */
1107   public KeyValue shallowCopy() {
1108     KeyValue shallowCopy = new KeyValue(this.bytes, this.offset, this.length);
1109     shallowCopy.setSequenceId(this.seqId);
1110     return shallowCopy;
1111   }
1112 
1113   //---------------------------------------------------------------------------
1114   //
1115   //  String representation
1116   //
1117   //---------------------------------------------------------------------------
1118 
1119   @Override
1120   public String toString() {
1121     if (this.bytes == null || this.bytes.length == 0) {
1122       return "empty";
1123     }
1124     return keyToString(this.bytes, this.offset + ROW_OFFSET, getKeyLength()) + "/vlen="
1125       + getValueLength() + "/seqid=" + seqId;
1126   }
1127 
1128   /**
1129    * @param k Key portion of a KeyValue.
1130    * @return Key as a String, empty string if k is null.
1131    */
1132   public static String keyToString(final byte [] k) {
1133     if (k == null) {
1134       return "";
1135     }
1136     return keyToString(k, 0, k.length);
1137   }
1138 
1139   /**
1140    * Produces a string map for this key/value pair. Useful for programmatic use
1141    * and manipulation of the data stored in an WALKey, for example, printing
1142    * as JSON. Values are left out due to their tendency to be large. If needed,
1143    * they can be added manually.
1144    *
1145    * @return the Map<String,?> containing data from this key
1146    */
1147   public Map<String, Object> toStringMap() {
1148     Map<String, Object> stringMap = new HashMap<String, Object>();
1149     stringMap.put("row", Bytes.toStringBinary(getRow()));
1150     stringMap.put("family", Bytes.toStringBinary(getFamily()));
1151     stringMap.put("qualifier", Bytes.toStringBinary(getQualifier()));
1152     stringMap.put("timestamp", getTimestamp());
1153     stringMap.put("vlen", getValueLength());
1154     List<Tag> tags = getTags();
1155     if (tags != null) {
1156       List<String> tagsString = new ArrayList<String>();
1157       for (Tag t : tags) {
1158         tagsString.add((t.getType()) + ":" +Bytes.toStringBinary(t.getValue()));
1159       }
1160       stringMap.put("tag", tagsString);
1161     }
1162     return stringMap;
1163   }
1164 
1165   /**
1166    * Use for logging.
1167    * @param b Key portion of a KeyValue.
1168    * @param o Offset to start of key
1169    * @param l Length of key.
1170    * @return Key as a String.
1171    */
1172   public static String keyToString(final byte [] b, final int o, final int l) {
1173     if (b == null) return "";
1174     int rowlength = Bytes.toShort(b, o);
1175     String row = Bytes.toStringBinary(b, o + Bytes.SIZEOF_SHORT, rowlength);
1176     int columnoffset = o + Bytes.SIZEOF_SHORT + 1 + rowlength;
1177     int familylength = b[columnoffset - 1];
1178     int columnlength = l - ((columnoffset - o) + TIMESTAMP_TYPE_SIZE);
1179     String family = familylength == 0? "":
1180       Bytes.toStringBinary(b, columnoffset, familylength);
1181     String qualifier = columnlength == 0? "":
1182       Bytes.toStringBinary(b, columnoffset + familylength,
1183       columnlength - familylength);
1184     long timestamp = Bytes.toLong(b, o + (l - TIMESTAMP_TYPE_SIZE));
1185     String timestampStr = humanReadableTimestamp(timestamp);
1186     byte type = b[o + l - 1];
1187     return row + "/" + family +
1188       (family != null && family.length() > 0? ":" :"") +
1189       qualifier + "/" + timestampStr + "/" + Type.codeToType(type);
1190   }
1191 
1192   public static String humanReadableTimestamp(final long timestamp) {
1193     if (timestamp == HConstants.LATEST_TIMESTAMP) {
1194       return "LATEST_TIMESTAMP";
1195     }
1196     if (timestamp == HConstants.OLDEST_TIMESTAMP) {
1197       return "OLDEST_TIMESTAMP";
1198     }
1199     return String.valueOf(timestamp);
1200   }
1201 
1202   //---------------------------------------------------------------------------
1203   //
1204   //  Public Member Accessors
1205   //
1206   //---------------------------------------------------------------------------
1207 
1208   /**
1209    * @return The byte array backing this KeyValue.
1210    * @deprecated Since 0.98.0.  Use Cell Interface instead.  Do not presume single backing buffer.
1211    */
1212   @Deprecated
1213   public byte [] getBuffer() {
1214     return this.bytes;
1215   }
1216 
1217   /**
1218    * @return Offset into {@link #getBuffer()} at which this KeyValue starts.
1219    */
1220   public int getOffset() {
1221     return this.offset;
1222   }
1223 
1224   /**
1225    * @return Length of bytes this KeyValue occupies in {@link #getBuffer()}.
1226    */
1227   public int getLength() {
1228     return length;
1229   }
1230 
1231   //---------------------------------------------------------------------------
1232   //
1233   //  Length and Offset Calculators
1234   //
1235   //---------------------------------------------------------------------------
1236 
1237   /**
1238    * Determines the total length of the KeyValue stored in the specified
1239    * byte array and offset.  Includes all headers.
1240    * @param bytes byte array
1241    * @param offset offset to start of the KeyValue
1242    * @return length of entire KeyValue, in bytes
1243    */
1244   private static int getLength(byte [] bytes, int offset) {
1245     int klength = ROW_OFFSET + Bytes.toInt(bytes, offset);
1246     int vlength = Bytes.toInt(bytes, offset + Bytes.SIZEOF_INT);
1247     return klength + vlength;
1248   }
1249 
1250   /**
1251    * @return Key offset in backing buffer..
1252    */
1253   public int getKeyOffset() {
1254     return this.offset + ROW_OFFSET;
1255   }
1256 
1257   public String getKeyString() {
1258     return Bytes.toStringBinary(getBuffer(), getKeyOffset(), getKeyLength());
1259   }
1260 
1261   /**
1262    * @return Length of key portion.
1263    */
1264   public int getKeyLength() {
1265     return Bytes.toInt(this.bytes, this.offset);
1266   }
1267 
1268   /**
1269    * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
1270    */
1271   @Override
1272   public byte[] getValueArray() {
1273     return bytes;
1274   }
1275 
1276   /**
1277    * @return the value offset
1278    */
1279   @Override
1280   public int getValueOffset() {
1281     int voffset = getKeyOffset() + getKeyLength();
1282     return voffset;
1283   }
1284 
1285   /**
1286    * @return Value length
1287    */
1288   @Override
1289   public int getValueLength() {
1290     int vlength = Bytes.toInt(this.bytes, this.offset + Bytes.SIZEOF_INT);
1291     return vlength;
1292   }
1293 
1294   /**
1295    * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
1296    */
1297   @Override
1298   public byte[] getRowArray() {
1299     return bytes;
1300   }
1301 
1302   /**
1303    * @return Row offset
1304    */
1305   @Override
1306   public int getRowOffset() {
1307     return getKeyOffset() + Bytes.SIZEOF_SHORT;
1308   }
1309 
1310   /**
1311    * @return Row length
1312    */
1313   @Override
1314   public short getRowLength() {
1315     return Bytes.toShort(this.bytes, getKeyOffset());
1316   }
1317 
1318   /**
1319    * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
1320    */
1321   @Override
1322   public byte[] getFamilyArray() {
1323     return bytes;
1324   }
1325 
1326   /**
1327    * @return Family offset
1328    */
1329   @Override
1330   public int getFamilyOffset() {
1331     return getFamilyOffset(getRowLength());
1332   }
1333 
1334   /**
1335    * @return Family offset
1336    */
1337   private int getFamilyOffset(int rlength) {
1338     return this.offset + ROW_OFFSET + Bytes.SIZEOF_SHORT + rlength + Bytes.SIZEOF_BYTE;
1339   }
1340 
1341   /**
1342    * @return Family length
1343    */
1344   @Override
1345   public byte getFamilyLength() {
1346     return getFamilyLength(getFamilyOffset());
1347   }
1348 
1349   /**
1350    * @return Family length
1351    */
1352   public byte getFamilyLength(int foffset) {
1353     return this.bytes[foffset-1];
1354   }
1355 
1356   /**
1357    * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
1358    */
1359   @Override
1360   public byte[] getQualifierArray() {
1361     return bytes;
1362   }
1363 
1364   /**
1365    * @return Qualifier offset
1366    */
1367   @Override
1368   public int getQualifierOffset() {
1369     return getQualifierOffset(getFamilyOffset());
1370   }
1371 
1372   /**
1373    * @return Qualifier offset
1374    */
1375   private int getQualifierOffset(int foffset) {
1376     return foffset + getFamilyLength(foffset);
1377   }
1378 
1379   /**
1380    * @return Qualifier length
1381    */
1382   @Override
1383   public int getQualifierLength() {
1384     return getQualifierLength(getRowLength(),getFamilyLength());
1385   }
1386 
1387   /**
1388    * @return Qualifier length
1389    */
1390   private int getQualifierLength(int rlength, int flength) {
1391     return getKeyLength() - (int) getKeyDataStructureSize(rlength, flength, 0);
1392   }
1393 
1394   /**
1395    * @return Timestamp offset
1396    */
1397   public int getTimestampOffset() {
1398     return getTimestampOffset(getKeyLength());
1399   }
1400 
1401   /**
1402    * @param keylength Pass if you have it to save on a int creation.
1403    * @return Timestamp offset
1404    */
1405   private int getTimestampOffset(final int keylength) {
1406     return getKeyOffset() + keylength - TIMESTAMP_TYPE_SIZE;
1407   }
1408 
1409   /**
1410    * @return True if this KeyValue has a LATEST_TIMESTAMP timestamp.
1411    */
1412   public boolean isLatestTimestamp() {
1413     return Bytes.equals(getBuffer(), getTimestampOffset(), Bytes.SIZEOF_LONG,
1414       HConstants.LATEST_TIMESTAMP_BYTES, 0, Bytes.SIZEOF_LONG);
1415   }
1416 
1417   /**
1418    * @param now Time to set into <code>this</code> IFF timestamp ==
1419    * {@link HConstants#LATEST_TIMESTAMP} (else, its a noop).
1420    * @return True is we modified this.
1421    */
1422   public boolean updateLatestStamp(final byte [] now) {
1423     if (this.isLatestTimestamp()) {
1424       int tsOffset = getTimestampOffset();
1425       System.arraycopy(now, 0, this.bytes, tsOffset, Bytes.SIZEOF_LONG);
1426       // clear cache or else getTimestamp() possibly returns an old value
1427       return true;
1428     }
1429     return false;
1430   }
1431 
1432   @Override
1433   public void setTimestamp(long ts) {
1434     Bytes.putBytes(this.bytes, this.getTimestampOffset(), Bytes.toBytes(ts), 0, Bytes.SIZEOF_LONG);
1435   }
1436 
1437   @Override
1438   public void setTimestamp(byte[] ts, int tsOffset) {
1439     Bytes.putBytes(this.bytes, this.getTimestampOffset(), ts, tsOffset, Bytes.SIZEOF_LONG);
1440   }
1441 
1442   //---------------------------------------------------------------------------
1443   //
1444   //  Methods that return copies of fields
1445   //
1446   //---------------------------------------------------------------------------
1447 
1448   /**
1449    * Do not use unless you have to.  Used internally for compacting and testing.
1450    *
1451    * Use {@link #getRow()}, {@link #getFamily()}, {@link #getQualifier()}, and
1452    * {@link #getValue()} if accessing a KeyValue client-side.
1453    * @return Copy of the key portion only.
1454    */
1455   public byte [] getKey() {
1456     int keylength = getKeyLength();
1457     byte [] key = new byte[keylength];
1458     System.arraycopy(getBuffer(), getKeyOffset(), key, 0, keylength);
1459     return key;
1460   }
1461 
1462   /**
1463    * Returns value in a new byte array.
1464    * Primarily for use client-side. If server-side, use
1465    * {@link #getBuffer()} with appropriate offsets and lengths instead to
1466    * save on allocations.
1467    * @return Value in a new byte array.
1468    */
1469   @Override
1470   @Deprecated // use CellUtil.getValueArray()
1471   public byte [] getValue() {
1472     return CellUtil.cloneValue(this);
1473   }
1474 
1475   /**
1476    * Primarily for use client-side.  Returns the row of this KeyValue in a new
1477    * byte array.<p>
1478    *
1479    * If server-side, use {@link #getBuffer()} with appropriate offsets and
1480    * lengths instead.
1481    * @return Row in a new byte array.
1482    */
1483   @Override
1484   @Deprecated // use CellUtil.getRowArray()
1485   public byte [] getRow() {
1486     return CellUtil.cloneRow(this);
1487   }
1488 
1489   /**
1490    *
1491    * @return Timestamp
1492    */
1493   @Override
1494   public long getTimestamp() {
1495     return getTimestamp(getKeyLength());
1496   }
1497 
1498   /**
1499    * @param keylength Pass if you have it to save on a int creation.
1500    * @return Timestamp
1501    */
1502   long getTimestamp(final int keylength) {
1503     int tsOffset = getTimestampOffset(keylength);
1504     return Bytes.toLong(this.bytes, tsOffset);
1505   }
1506 
1507   /**
1508    * @return Type of this KeyValue.
1509    */
1510   @Deprecated
1511   public byte getType() {
1512     return getTypeByte();
1513   }
1514 
1515   /**
1516    * @return KeyValue.TYPE byte representation
1517    */
1518   @Override
1519   public byte getTypeByte() {
1520     return this.bytes[this.offset + getKeyLength() - 1 + ROW_OFFSET];
1521   }
1522 
1523   /**
1524    * @return True if a delete type, a {@link KeyValue.Type#Delete} or
1525    * a {KeyValue.Type#DeleteFamily} or a {@link KeyValue.Type#DeleteColumn}
1526    * KeyValue type.
1527    */
1528   @Deprecated // use CellUtil#isDelete
1529   public boolean isDelete() {
1530     return KeyValue.isDelete(getType());
1531   }
1532 
1533   /**
1534    * Primarily for use client-side.  Returns the family of this KeyValue in a
1535    * new byte array.<p>
1536    *
1537    * If server-side, use {@link #getBuffer()} with appropriate offsets and
1538    * lengths instead.
1539    * @return Returns family. Makes a copy.
1540    */
1541   @Override
1542   @Deprecated // use CellUtil.getFamilyArray
1543   public byte [] getFamily() {
1544     return CellUtil.cloneFamily(this);
1545   }
1546 
1547   /**
1548    * Primarily for use client-side.  Returns the column qualifier of this
1549    * KeyValue in a new byte array.<p>
1550    *
1551    * If server-side, use {@link #getBuffer()} with appropriate offsets and
1552    * lengths instead.
1553    * Use {@link #getBuffer()} with appropriate offsets and lengths instead.
1554    * @return Returns qualifier. Makes a copy.
1555    */
1556   @Override
1557   @Deprecated // use CellUtil.getQualifierArray
1558   public byte [] getQualifier() {
1559     return CellUtil.cloneQualifier(this);
1560   }
1561 
1562   /**
1563    * This returns the offset where the tag actually starts.
1564    */
1565   @Override
1566   public int getTagsOffset() {
1567     int tagsLen = getTagsLength();
1568     if (tagsLen == 0) {
1569       return this.offset + this.length;
1570     }
1571     return this.offset + this.length - tagsLen;
1572   }
1573 
1574   /**
1575    * This returns the total length of the tag bytes
1576    */
1577   @Override
1578   public int getTagsLength() {
1579     int tagsLen = this.length - (getKeyLength() + getValueLength() + KEYVALUE_INFRASTRUCTURE_SIZE);
1580     if (tagsLen > 0) {
1581       // There are some Tag bytes in the byte[]. So reduce 2 bytes which is added to denote the tags
1582       // length
1583       tagsLen -= TAGS_LENGTH_SIZE;
1584     }
1585     return tagsLen;
1586   }
1587 
1588   /**
1589    * Returns any tags embedded in the KeyValue.  Used in testcases.
1590    * @return The tags
1591    */
1592   public List<Tag> getTags() {
1593     int tagsLength = getTagsLength();
1594     if (tagsLength == 0) {
1595       return EMPTY_ARRAY_LIST;
1596     }
1597     return Tag.asList(getTagsArray(), getTagsOffset(), tagsLength);
1598   }
1599 
1600   /**
1601    * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
1602    */
1603   @Override
1604   public byte[] getTagsArray() {
1605     return bytes;
1606   }
1607 
1608   /**
1609    * Creates a new KeyValue that only contains the key portion (the value is
1610    * set to be null).
1611    *
1612    * TODO only used by KeyOnlyFilter -- move there.
1613    * @param lenAsVal replace value with the actual value length (false=empty)
1614    */
1615   public KeyValue createKeyOnly(boolean lenAsVal) {
1616     // KV format:  <keylen:4><valuelen:4><key:keylen><value:valuelen>
1617     // Rebuild as: <keylen:4><0:4><key:keylen>
1618     int dataLen = lenAsVal? Bytes.SIZEOF_INT : 0;
1619     byte [] newBuffer = new byte[getKeyLength() + ROW_OFFSET + dataLen];
1620     System.arraycopy(this.bytes, this.offset, newBuffer, 0,
1621         Math.min(newBuffer.length,this.length));
1622     Bytes.putInt(newBuffer, Bytes.SIZEOF_INT, dataLen);
1623     if (lenAsVal) {
1624       Bytes.putInt(newBuffer, newBuffer.length - dataLen, this.getValueLength());
1625     }
1626     return new KeyValue(newBuffer);
1627   }
1628 
1629   /**
1630    * Splits a column in {@code family:qualifier} form into separate byte arrays. An empty qualifier
1631    * (ie, {@code fam:}) is parsed as <code>{ fam, EMPTY_BYTE_ARRAY }</code> while no delimiter (ie,
1632    * {@code fam}) is parsed as an array of one element, <code>{ fam }</code>.
1633    * <p>
1634    * Don't forget, HBase DOES support empty qualifiers. (see HBASE-9549)
1635    * </p>
1636    * <p>
1637    * Not recommend to be used as this is old-style API.
1638    * </p>
1639    * @param c The column.
1640    * @return The parsed column.
1641    */
1642   public static byte [][] parseColumn(byte [] c) {
1643     final int index = getDelimiter(c, 0, c.length, COLUMN_FAMILY_DELIMITER);
1644     if (index == -1) {
1645       // If no delimiter, return array of size 1
1646       return new byte [][] { c };
1647     } else if(index == c.length - 1) {
1648       // family with empty qualifier, return array size 2
1649       byte [] family = new byte[c.length-1];
1650       System.arraycopy(c, 0, family, 0, family.length);
1651       return new byte [][] { family, HConstants.EMPTY_BYTE_ARRAY};
1652     }
1653     // Family and column, return array size 2
1654     final byte [][] result = new byte [2][];
1655     result[0] = new byte [index];
1656     System.arraycopy(c, 0, result[0], 0, index);
1657     final int len = c.length - (index + 1);
1658     result[1] = new byte[len];
1659     System.arraycopy(c, index + 1 /* Skip delimiter */, result[1], 0, len);
1660     return result;
1661   }
1662 
1663   /**
1664    * Makes a column in family:qualifier form from separate byte arrays.
1665    * <p>
1666    * Not recommended for usage as this is old-style API.
1667    * @param family
1668    * @param qualifier
1669    * @return family:qualifier
1670    */
1671   public static byte [] makeColumn(byte [] family, byte [] qualifier) {
1672     return Bytes.add(family, COLUMN_FAMILY_DELIM_ARRAY, qualifier);
1673   }
1674 
1675   /**
1676    * @param b
1677    * @param delimiter
1678    * @return Index of delimiter having started from start of <code>b</code>
1679    * moving rightward.
1680    */
1681   public static int getDelimiter(final byte [] b, int offset, final int length,
1682       final int delimiter) {
1683     if (b == null) {
1684       throw new IllegalArgumentException("Passed buffer is null");
1685     }
1686     int result = -1;
1687     for (int i = offset; i < length + offset; i++) {
1688       if (b[i] == delimiter) {
1689         result = i;
1690         break;
1691       }
1692     }
1693     return result;
1694   }
1695 
1696   /**
1697    * Find index of passed delimiter walking from end of buffer backwards.
1698    * @param b
1699    * @param delimiter
1700    * @return Index of delimiter
1701    */
1702   public static int getDelimiterInReverse(final byte [] b, final int offset,
1703       final int length, final int delimiter) {
1704     if (b == null) {
1705       throw new IllegalArgumentException("Passed buffer is null");
1706     }
1707     int result = -1;
1708     for (int i = (offset + length) - 1; i >= offset; i--) {
1709       if (b[i] == delimiter) {
1710         result = i;
1711         break;
1712       }
1713     }
1714     return result;
1715   }
1716 
1717   /**
1718    * A {@link KVComparator} for <code>hbase:meta</code> catalog table
1719    * {@link KeyValue}s.
1720    */
1721   public static class MetaComparator extends KVComparator {
1722     /**
1723      * Compare key portion of a {@link KeyValue} for keys in <code>hbase:meta</code>
1724      * table.
1725      */
1726     @Override
1727     public int compare(final Cell left, final Cell right) {
1728       int c = compareRowKey(left, right);
1729       if (c != 0) {
1730         return c;
1731       }
1732       return CellComparator.compareWithoutRow(left, right);
1733     }
1734 
1735     @Override
1736     public int compareOnlyKeyPortion(Cell left, Cell right) {
1737       return compare(left, right);
1738     }
1739 
1740     @Override
1741     public int compareRows(byte [] left, int loffset, int llength,
1742         byte [] right, int roffset, int rlength) {
1743       int leftDelimiter = getDelimiter(left, loffset, llength,
1744           HConstants.DELIMITER);
1745       int rightDelimiter = getDelimiter(right, roffset, rlength,
1746           HConstants.DELIMITER);
1747       // Compare up to the delimiter
1748       int lpart = (leftDelimiter < 0 ? llength :leftDelimiter - loffset);
1749       int rpart = (rightDelimiter < 0 ? rlength :rightDelimiter - roffset);
1750       int result = Bytes.compareTo(left, loffset, lpart, right, roffset, rpart);
1751       if (result != 0) {
1752         return result;
1753       } else {
1754         if (leftDelimiter < 0 && rightDelimiter >= 0) {
1755           return -1;
1756         } else if (rightDelimiter < 0 && leftDelimiter >= 0) {
1757           return 1;
1758         } else if (leftDelimiter < 0 && rightDelimiter < 0) {
1759           return 0;
1760         }
1761       }
1762       // Compare middle bit of the row.
1763       // Move past delimiter
1764       leftDelimiter++;
1765       rightDelimiter++;
1766       int leftFarDelimiter = getDelimiterInReverse(left, leftDelimiter,
1767           llength - (leftDelimiter - loffset), HConstants.DELIMITER);
1768       int rightFarDelimiter = getDelimiterInReverse(right,
1769           rightDelimiter, rlength - (rightDelimiter - roffset),
1770           HConstants.DELIMITER);
1771       // Now compare middlesection of row.
1772       lpart = (leftFarDelimiter < 0 ? llength + loffset: leftFarDelimiter) - leftDelimiter;
1773       rpart = (rightFarDelimiter < 0 ? rlength + roffset: rightFarDelimiter)- rightDelimiter;
1774       result = super.compareRows(left, leftDelimiter, lpart, right, rightDelimiter, rpart);
1775       if (result != 0) {
1776         return result;
1777       }  else {
1778         if (leftDelimiter < 0 && rightDelimiter >= 0) {
1779           return -1;
1780         } else if (rightDelimiter < 0 && leftDelimiter >= 0) {
1781           return 1;
1782         } else if (leftDelimiter < 0 && rightDelimiter < 0) {
1783           return 0;
1784         }
1785       }
1786       // Compare last part of row, the rowid.
1787       leftFarDelimiter++;
1788       rightFarDelimiter++;
1789       result = Bytes.compareTo(left, leftFarDelimiter, llength - (leftFarDelimiter - loffset),
1790           right, rightFarDelimiter, rlength - (rightFarDelimiter - roffset));
1791       return result;
1792     }
1793 
1794     /**
1795      * Don't do any fancy Block Index splitting tricks.
1796      */
1797     @Override
1798     public byte[] getShortMidpointKey(final byte[] leftKey, final byte[] rightKey) {
1799       return Arrays.copyOf(rightKey, rightKey.length);
1800     }
1801 
1802     /**
1803      * The HFileV2 file format's trailer contains this class name.  We reinterpret this and
1804      * instantiate the appropriate comparator.
1805      * TODO: With V3 consider removing this.
1806      * @return legacy class name for FileFileTrailer#comparatorClassName
1807      */
1808     @Override
1809     public String getLegacyKeyComparatorName() {
1810       return "org.apache.hadoop.hbase.KeyValue$MetaKeyComparator";
1811     }
1812 
1813     @Override
1814     protected Object clone() throws CloneNotSupportedException {
1815       return new MetaComparator();
1816     }
1817 
1818     /**
1819      * Override the row key comparison to parse and compare the meta row key parts.
1820      */
1821     @Override
1822     protected int compareRowKey(final Cell l, final Cell r) {
1823       byte[] left = l.getRowArray();
1824       int loffset = l.getRowOffset();
1825       int llength = l.getRowLength();
1826       byte[] right = r.getRowArray();
1827       int roffset = r.getRowOffset();
1828       int rlength = r.getRowLength();
1829       return compareRows(left, loffset, llength, right, roffset, rlength);
1830     }
1831   }
1832 
1833   /**
1834    * Compare KeyValues.  When we compare KeyValues, we only compare the Key
1835    * portion.  This means two KeyValues with same Key but different Values are
1836    * considered the same as far as this Comparator is concerned.
1837    */
1838   public static class KVComparator implements RawComparator<Cell>, SamePrefixComparator<byte[]> {
1839 
1840     /**
1841      * The HFileV2 file format's trailer contains this class name.  We reinterpret this and
1842      * instantiate the appropriate comparator.
1843      * TODO: With V3 consider removing this.
1844      * @return legacy class name for FileFileTrailer#comparatorClassName
1845      */
1846     public String getLegacyKeyComparatorName() {
1847       return "org.apache.hadoop.hbase.KeyValue$KeyComparator";
1848     }
1849 
1850     @Override // RawComparator
1851     public int compare(byte[] l, int loff, int llen, byte[] r, int roff, int rlen) {
1852       return compareFlatKey(l,loff,llen, r,roff,rlen);
1853     }
1854 
1855 
1856     /**
1857      * Compares the only the user specified portion of a Key.  This is overridden by MetaComparator.
1858      * @param left
1859      * @param right
1860      * @return 0 if equal, <0 if left smaller, >0 if right smaller
1861      */
1862     protected int compareRowKey(final Cell left, final Cell right) {
1863       return CellComparator.compareRows(left, right);
1864     }
1865 
1866     /**
1867      * Compares left to right assuming that left,loffset,llength and right,roffset,rlength are
1868      * full KVs laid out in a flat byte[]s.
1869      * @param left
1870      * @param loffset
1871      * @param llength
1872      * @param right
1873      * @param roffset
1874      * @param rlength
1875      * @return  0 if equal, <0 if left smaller, >0 if right smaller
1876      */
1877     public int compareFlatKey(byte[] left, int loffset, int llength,
1878         byte[] right, int roffset, int rlength) {
1879       // Compare row
1880       short lrowlength = Bytes.toShort(left, loffset);
1881       short rrowlength = Bytes.toShort(right, roffset);
1882       int compare = compareRows(left, loffset + Bytes.SIZEOF_SHORT,
1883           lrowlength, right, roffset + Bytes.SIZEOF_SHORT, rrowlength);
1884       if (compare != 0) {
1885         return compare;
1886       }
1887 
1888       // Compare the rest of the two KVs without making any assumptions about
1889       // the common prefix. This function will not compare rows anyway, so we
1890       // don't need to tell it that the common prefix includes the row.
1891       return compareWithoutRow(0, left, loffset, llength, right, roffset,
1892           rlength, rrowlength);
1893     }
1894 
1895     public int compareFlatKey(byte[] left, byte[] right) {
1896       return compareFlatKey(left, 0, left.length, right, 0, right.length);
1897     }
1898 
1899     // compare a key against row/fam/qual/ts/type
1900     public int compareKey(Cell cell,
1901         byte[] row, int roff, int rlen,
1902         byte[] fam, int foff, int flen,
1903         byte[] col, int coff, int clen,
1904         long ts, byte type) {
1905 
1906       int compare = compareRows(
1907         cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(),
1908         row, roff, rlen);
1909       if (compare != 0) {
1910         return compare;
1911       }
1912       // If the column is not specified, the "minimum" key type appears the
1913       // latest in the sorted order, regardless of the timestamp. This is used
1914       // for specifying the last key/value in a given row, because there is no
1915       // "lexicographically last column" (it would be infinitely long). The
1916       // "maximum" key type does not need this behavior.
1917       if (cell.getFamilyLength() + cell.getQualifierLength() == 0
1918           && cell.getTypeByte() == Type.Minimum.getCode()) {
1919         // left is "bigger", i.e. it appears later in the sorted order
1920         return 1;
1921       }
1922       if (flen+clen == 0 && type == Type.Minimum.getCode()) {
1923         return -1;
1924       }
1925 
1926       compare = compareFamilies(
1927         cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(),
1928         fam, foff, flen);
1929       if (compare != 0) {
1930         return compare;
1931       }
1932       compare = compareColumns(
1933         cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength(),
1934         col, coff, clen);
1935       if (compare != 0) {
1936         return compare;
1937       }
1938       // Next compare timestamps.
1939       compare = compareTimestamps(cell.getTimestamp(), ts);
1940       if (compare != 0) {
1941         return compare;
1942       }
1943 
1944       // Compare types. Let the delete types sort ahead of puts; i.e. types
1945       // of higher numbers sort before those of lesser numbers. Maximum (255)
1946       // appears ahead of everything, and minimum (0) appears after
1947       // everything.
1948       return (0xff & type) - (0xff & cell.getTypeByte());
1949     }
1950 
1951     public int compareOnlyKeyPortion(Cell left, Cell right) {
1952       return CellComparator.compare(left, right, true);
1953     }
1954 
1955     /**
1956      * Compares the Key of a cell -- with fields being more significant in this order:
1957      * rowkey, colfam/qual, timestamp, type, mvcc
1958      */
1959     @Override
1960     public int compare(final Cell left, final Cell right) {
1961       int compare = CellComparator.compare(left, right, false);
1962       return compare;
1963     }
1964 
1965     public int compareTimestamps(final Cell left, final Cell right) {
1966       return CellComparator.compareTimestamps(left, right);
1967     }
1968 
1969     /**
1970      * @param left
1971      * @param right
1972      * @return Result comparing rows.
1973      */
1974     public int compareRows(final Cell left, final Cell right) {
1975       return compareRows(left.getRowArray(),left.getRowOffset(), left.getRowLength(),
1976       right.getRowArray(), right.getRowOffset(), right.getRowLength());
1977     }
1978 
1979     /**
1980      * Get the b[],o,l for left and right rowkey portions and compare.
1981      * @param left
1982      * @param loffset
1983      * @param llength
1984      * @param right
1985      * @param roffset
1986      * @param rlength
1987      * @return 0 if equal, <0 if left smaller, >0 if right smaller
1988      */
1989     public int compareRows(byte [] left, int loffset, int llength,
1990         byte [] right, int roffset, int rlength) {
1991       return Bytes.compareTo(left, loffset, llength, right, roffset, rlength);
1992     }
1993 
1994     int compareColumns(final Cell left, final short lrowlength, final Cell right,
1995         final short rrowlength) {
1996       return CellComparator.compareColumns(left, right);
1997     }
1998 
1999     protected int compareColumns(
2000         byte [] left, int loffset, int llength, final int lfamilylength,
2001         byte [] right, int roffset, int rlength, final int rfamilylength) {
2002       // Compare family portion first.
2003       int diff = Bytes.compareTo(left, loffset, lfamilylength,
2004         right, roffset, rfamilylength);
2005       if (diff != 0) {
2006         return diff;
2007       }
2008       // Compare qualifier portion
2009       return Bytes.compareTo(left, loffset + lfamilylength,
2010         llength - lfamilylength,
2011         right, roffset + rfamilylength, rlength - rfamilylength);
2012       }
2013 
2014     static int compareTimestamps(final long ltimestamp, final long rtimestamp) {
2015       // The below older timestamps sorting ahead of newer timestamps looks
2016       // wrong but it is intentional. This way, newer timestamps are first
2017       // found when we iterate over a memstore and newer versions are the
2018       // first we trip over when reading from a store file.
2019       if (ltimestamp < rtimestamp) {
2020         return 1;
2021       } else if (ltimestamp > rtimestamp) {
2022         return -1;
2023       }
2024       return 0;
2025     }
2026 
2027     /**
2028      * Overridden
2029      * @param commonPrefix
2030      * @param left
2031      * @param loffset
2032      * @param llength
2033      * @param right
2034      * @param roffset
2035      * @param rlength
2036      * @return 0 if equal, <0 if left smaller, >0 if right smaller
2037      */
2038     @Override // SamePrefixComparator
2039     public int compareIgnoringPrefix(int commonPrefix, byte[] left,
2040         int loffset, int llength, byte[] right, int roffset, int rlength) {
2041       // Compare row
2042       short lrowlength = Bytes.toShort(left, loffset);
2043       short rrowlength;
2044 
2045       int comparisonResult = 0;
2046       if (commonPrefix < ROW_LENGTH_SIZE) {
2047         // almost nothing in common
2048         rrowlength = Bytes.toShort(right, roffset);
2049         comparisonResult = compareRows(left, loffset + ROW_LENGTH_SIZE,
2050             lrowlength, right, roffset + ROW_LENGTH_SIZE, rrowlength);
2051       } else { // the row length is the same
2052         rrowlength = lrowlength;
2053         if (commonPrefix < ROW_LENGTH_SIZE + rrowlength) {
2054           // The rows are not the same. Exclude the common prefix and compare
2055           // the rest of the two rows.
2056           int common = commonPrefix - ROW_LENGTH_SIZE;
2057           comparisonResult = compareRows(
2058               left, loffset + common + ROW_LENGTH_SIZE, lrowlength - common,
2059               right, roffset + common + ROW_LENGTH_SIZE, rrowlength - common);
2060         }
2061       }
2062       if (comparisonResult != 0) {
2063         return comparisonResult;
2064       }
2065 
2066       assert lrowlength == rrowlength;
2067       return compareWithoutRow(commonPrefix, left, loffset, llength, right,
2068           roffset, rlength, lrowlength);
2069     }
2070 
2071     /**
2072      * Compare columnFamily, qualifier, timestamp, and key type (everything
2073      * except the row). This method is used both in the normal comparator and
2074      * the "same-prefix" comparator. Note that we are assuming that row portions
2075      * of both KVs have already been parsed and found identical, and we don't
2076      * validate that assumption here.
2077      * @param commonPrefix
2078      *          the length of the common prefix of the two key-values being
2079      *          compared, including row length and row
2080      */
2081     private int compareWithoutRow(int commonPrefix, byte[] left, int loffset,
2082         int llength, byte[] right, int roffset, int rlength, short rowlength) {
2083       /***
2084        * KeyValue Format and commonLength:
2085        * |_keyLen_|_valLen_|_rowLen_|_rowKey_|_famiLen_|_fami_|_Quali_|....
2086        * ------------------|-------commonLength--------|--------------
2087        */
2088       int commonLength = ROW_LENGTH_SIZE + FAMILY_LENGTH_SIZE + rowlength;
2089 
2090       // commonLength + TIMESTAMP_TYPE_SIZE
2091       int commonLengthWithTSAndType = TIMESTAMP_TYPE_SIZE + commonLength;
2092       // ColumnFamily + Qualifier length.
2093       int lcolumnlength = llength - commonLengthWithTSAndType;
2094       int rcolumnlength = rlength - commonLengthWithTSAndType;
2095 
2096       byte ltype = left[loffset + (llength - 1)];
2097       byte rtype = right[roffset + (rlength - 1)];
2098 
2099       // If the column is not specified, the "minimum" key type appears the
2100       // latest in the sorted order, regardless of the timestamp. This is used
2101       // for specifying the last key/value in a given row, because there is no
2102       // "lexicographically last column" (it would be infinitely long). The
2103       // "maximum" key type does not need this behavior.
2104       if (lcolumnlength == 0 && ltype == Type.Minimum.getCode()) {
2105         // left is "bigger", i.e. it appears later in the sorted order
2106         return 1;
2107       }
2108       if (rcolumnlength == 0 && rtype == Type.Minimum.getCode()) {
2109         return -1;
2110       }
2111 
2112       int lfamilyoffset = commonLength + loffset;
2113       int rfamilyoffset = commonLength + roffset;
2114 
2115       // Column family length.
2116       int lfamilylength = left[lfamilyoffset - 1];
2117       int rfamilylength = right[rfamilyoffset - 1];
2118       // If left family size is not equal to right family size, we need not
2119       // compare the qualifiers.
2120       boolean sameFamilySize = (lfamilylength == rfamilylength);
2121       int common = 0;
2122       if (commonPrefix > 0) {
2123         common = Math.max(0, commonPrefix - commonLength);
2124         if (!sameFamilySize) {
2125           // Common should not be larger than Math.min(lfamilylength,
2126           // rfamilylength).
2127           common = Math.min(common, Math.min(lfamilylength, rfamilylength));
2128         } else {
2129           common = Math.min(common, Math.min(lcolumnlength, rcolumnlength));
2130         }
2131       }
2132       if (!sameFamilySize) {
2133         // comparing column family is enough.
2134         return Bytes.compareTo(left, lfamilyoffset + common, lfamilylength
2135             - common, right, rfamilyoffset + common, rfamilylength - common);
2136       }
2137       // Compare family & qualifier together.
2138       final int comparison = Bytes.compareTo(left, lfamilyoffset + common,
2139           lcolumnlength - common, right, rfamilyoffset + common,
2140           rcolumnlength - common);
2141       if (comparison != 0) {
2142         return comparison;
2143       }
2144 
2145       ////
2146       // Next compare timestamps.
2147       long ltimestamp = Bytes.toLong(left,
2148           loffset + (llength - TIMESTAMP_TYPE_SIZE));
2149       long rtimestamp = Bytes.toLong(right,
2150           roffset + (rlength - TIMESTAMP_TYPE_SIZE));
2151       int compare = compareTimestamps(ltimestamp, rtimestamp);
2152       if (compare != 0) {
2153         return compare;
2154       }
2155 
2156       // Compare types. Let the delete types sort ahead of puts; i.e. types
2157       // of higher numbers sort before those of lesser numbers. Maximum (255)
2158       // appears ahead of everything, and minimum (0) appears after
2159       // everything.
2160       return (0xff & rtype) - (0xff & ltype);
2161     }
2162 
2163     protected int compareFamilies(final byte[] left, final int loffset, final int lfamilylength,
2164         final byte[] right, final int roffset, final int rfamilylength) {
2165       int diff = Bytes.compareTo(left, loffset, lfamilylength, right, roffset, rfamilylength);
2166       return diff;
2167     }
2168 
2169     protected int compareColumns(final byte[] left, final int loffset, final int lquallength,
2170         final byte[] right, final int roffset, final int rquallength) {
2171       int diff = Bytes.compareTo(left, loffset, lquallength, right, roffset, rquallength);
2172       return diff;
2173     }
2174     /**
2175      * Compares the row and column of two keyvalues for equality
2176      * @param left
2177      * @param right
2178      * @return True if same row and column.
2179      */
2180     public boolean matchingRowColumn(final Cell left,
2181         final Cell right) {
2182       short lrowlength = left.getRowLength();
2183       short rrowlength = right.getRowLength();
2184 
2185       // TsOffset = end of column data. just comparing Row+CF length of each
2186       if ((left.getRowLength() + left.getFamilyLength() + left.getQualifierLength()) != (right
2187           .getRowLength() + right.getFamilyLength() + right.getQualifierLength())) {
2188         return false;
2189       }
2190 
2191       if (!matchingRows(left, lrowlength, right, rrowlength)) {
2192         return false;
2193       }
2194 
2195       int lfoffset = left.getFamilyOffset();
2196       int rfoffset = right.getFamilyOffset();
2197       int lclength = left.getQualifierLength();
2198       int rclength = right.getQualifierLength();
2199       int lfamilylength = left.getFamilyLength();
2200       int rfamilylength = right.getFamilyLength();
2201       int diff = compareFamilies(left.getFamilyArray(), lfoffset, lfamilylength,
2202           right.getFamilyArray(), rfoffset, rfamilylength);
2203       if (diff != 0) {
2204         return false;
2205       } else {
2206         diff = compareColumns(left.getQualifierArray(), left.getQualifierOffset(), lclength,
2207             right.getQualifierArray(), right.getQualifierOffset(), rclength);
2208         return diff == 0;
2209       }
2210     }
2211 
2212     /**
2213      * Compares the row of two keyvalues for equality
2214      * @param left
2215      * @param right
2216      * @return True if rows match.
2217      */
2218     public boolean matchingRows(final Cell left, final Cell right) {
2219       short lrowlength = left.getRowLength();
2220       short rrowlength = right.getRowLength();
2221       return matchingRows(left, lrowlength, right, rrowlength);
2222     }
2223 
2224     /**
2225      * @param left
2226      * @param lrowlength
2227      * @param right
2228      * @param rrowlength
2229      * @return True if rows match.
2230      */
2231     private boolean matchingRows(final Cell left, final short lrowlength,
2232         final Cell right, final short rrowlength) {
2233       return lrowlength == rrowlength &&
2234           matchingRows(left.getRowArray(), left.getRowOffset(), lrowlength,
2235               right.getRowArray(), right.getRowOffset(), rrowlength);
2236     }
2237 
2238     /**
2239      * Compare rows. Just calls Bytes.equals, but it's good to have this encapsulated.
2240      * @param left Left row array.
2241      * @param loffset Left row offset.
2242      * @param llength Left row length.
2243      * @param right Right row array.
2244      * @param roffset Right row offset.
2245      * @param rlength Right row length.
2246      * @return Whether rows are the same row.
2247      */
2248     public boolean matchingRows(final byte [] left, final int loffset, final int llength,
2249         final byte [] right, final int roffset, final int rlength) {
2250       return Bytes.equals(left, loffset, llength, right, roffset, rlength);
2251     }
2252 
2253     public byte[] calcIndexKey(byte[] lastKeyOfPreviousBlock, byte[] firstKeyInBlock) {
2254       byte[] fakeKey = getShortMidpointKey(lastKeyOfPreviousBlock, firstKeyInBlock);
2255       if (compareFlatKey(fakeKey, firstKeyInBlock) > 0) {
2256         LOG.error("Unexpected getShortMidpointKey result, fakeKey:"
2257             + Bytes.toStringBinary(fakeKey) + ", firstKeyInBlock:"
2258             + Bytes.toStringBinary(firstKeyInBlock));
2259         return firstKeyInBlock;
2260       }
2261       if (lastKeyOfPreviousBlock != null && compareFlatKey(lastKeyOfPreviousBlock, fakeKey) >= 0) {
2262         LOG.error("Unexpected getShortMidpointKey result, lastKeyOfPreviousBlock:" +
2263             Bytes.toStringBinary(lastKeyOfPreviousBlock) + ", fakeKey:" +
2264             Bytes.toStringBinary(fakeKey));
2265         return firstKeyInBlock;
2266       }
2267       return fakeKey;
2268     }
2269 
2270     /**
2271      * This is a HFile block index key optimization.
2272      * @param leftKey
2273      * @param rightKey
2274      * @return 0 if equal, <0 if left smaller, >0 if right smaller
2275      * @deprecated Since 0.99.2; Use
2276      *             {@link CellComparator#getMidpoint(KeyValue.KVComparator, Cell, Cell) instead}
2277      */
2278     @Deprecated
2279     public byte[] getShortMidpointKey(final byte[] leftKey, final byte[] rightKey) {
2280       if (rightKey == null) {
2281         throw new IllegalArgumentException("rightKey can not be null");
2282       }
2283       if (leftKey == null) {
2284         return Arrays.copyOf(rightKey, rightKey.length);
2285       }
2286       if (compareFlatKey(leftKey, rightKey) >= 0) {
2287         throw new IllegalArgumentException("Unexpected input, leftKey:" + Bytes.toString(leftKey)
2288           + ", rightKey:" + Bytes.toString(rightKey));
2289       }
2290 
2291       short leftRowLength = Bytes.toShort(leftKey, 0);
2292       short rightRowLength = Bytes.toShort(rightKey, 0);
2293       int leftCommonLength = ROW_LENGTH_SIZE + FAMILY_LENGTH_SIZE + leftRowLength;
2294       int rightCommonLength = ROW_LENGTH_SIZE + FAMILY_LENGTH_SIZE + rightRowLength;
2295       int leftCommonLengthWithTSAndType = TIMESTAMP_TYPE_SIZE + leftCommonLength;
2296       int rightCommonLengthWithTSAndType = TIMESTAMP_TYPE_SIZE + rightCommonLength;
2297       int leftColumnLength = leftKey.length - leftCommonLengthWithTSAndType;
2298       int rightColumnLength = rightKey.length - rightCommonLengthWithTSAndType;
2299       // rows are equal
2300       if (leftRowLength == rightRowLength && compareRows(leftKey, ROW_LENGTH_SIZE, leftRowLength,
2301         rightKey, ROW_LENGTH_SIZE, rightRowLength) == 0) {
2302         // Compare family & qualifier together.
2303         int comparison = Bytes.compareTo(leftKey, leftCommonLength, leftColumnLength, rightKey,
2304           rightCommonLength, rightColumnLength);
2305         // same with "row + family + qualifier", return rightKey directly
2306         if (comparison == 0) {
2307           return Arrays.copyOf(rightKey, rightKey.length);
2308         }
2309         // "family + qualifier" are different, generate a faked key per rightKey
2310         byte[] newKey = Arrays.copyOf(rightKey, rightKey.length);
2311         Bytes.putLong(newKey, rightKey.length - TIMESTAMP_TYPE_SIZE, HConstants.LATEST_TIMESTAMP);
2312         Bytes.putByte(newKey, rightKey.length - TYPE_SIZE, Type.Maximum.getCode());
2313         return newKey;
2314       }
2315       // rows are different
2316       short minLength = leftRowLength < rightRowLength ? leftRowLength : rightRowLength;
2317       short diffIdx = 0;
2318       while (diffIdx < minLength
2319           && leftKey[ROW_LENGTH_SIZE + diffIdx] == rightKey[ROW_LENGTH_SIZE + diffIdx]) {
2320         diffIdx++;
2321       }
2322       byte[] newRowKey = null;
2323       if (diffIdx >= minLength) {
2324         // leftKey's row is prefix of rightKey's.
2325         newRowKey = new byte[diffIdx + 1];
2326         System.arraycopy(rightKey, ROW_LENGTH_SIZE, newRowKey, 0, diffIdx + 1);
2327       } else {
2328         int diffByte = leftKey[ROW_LENGTH_SIZE + diffIdx];
2329         if ((0xff & diffByte) < 0xff && (diffByte + 1) <
2330             (rightKey[ROW_LENGTH_SIZE + diffIdx] & 0xff)) {
2331           newRowKey = new byte[diffIdx + 1];
2332           System.arraycopy(leftKey, ROW_LENGTH_SIZE, newRowKey, 0, diffIdx);
2333           newRowKey[diffIdx] = (byte) (diffByte + 1);
2334         } else {
2335           newRowKey = new byte[diffIdx + 1];
2336           System.arraycopy(rightKey, ROW_LENGTH_SIZE, newRowKey, 0, diffIdx + 1);
2337         }
2338       }
2339       return new KeyValue(newRowKey, null, null, HConstants.LATEST_TIMESTAMP,
2340         Type.Maximum).getKey();
2341     }
2342 
2343     @Override
2344     protected Object clone() throws CloneNotSupportedException {
2345       super.clone();
2346       return new KVComparator();
2347     }
2348 
2349   }
2350 
2351   /**
2352    * @param b
2353    * @return A KeyValue made of a byte array that holds the key-only part.
2354    * Needed to convert hfile index members to KeyValues.
2355    */
2356   public static KeyValue createKeyValueFromKey(final byte [] b) {
2357     return createKeyValueFromKey(b, 0, b.length);
2358   }
2359 
2360   /**
2361    * @param bb
2362    * @return A KeyValue made of a byte buffer that holds the key-only part.
2363    * Needed to convert hfile index members to KeyValues.
2364    */
2365   public static KeyValue createKeyValueFromKey(final ByteBuffer bb) {
2366     return createKeyValueFromKey(bb.array(), bb.arrayOffset(), bb.limit());
2367   }
2368 
2369   /**
2370    * @param b
2371    * @param o
2372    * @param l
2373    * @return A KeyValue made of a byte array that holds the key-only part.
2374    * Needed to convert hfile index members to KeyValues.
2375    */
2376   public static KeyValue createKeyValueFromKey(final byte [] b, final int o,
2377       final int l) {
2378     byte [] newb = new byte[l + ROW_OFFSET];
2379     System.arraycopy(b, o, newb, ROW_OFFSET, l);
2380     Bytes.putInt(newb, 0, l);
2381     Bytes.putInt(newb, Bytes.SIZEOF_INT, 0);
2382     return new KeyValue(newb);
2383   }
2384 
2385   /**
2386    * @param in Where to read bytes from.  Creates a byte array to hold the KeyValue
2387    * backing bytes copied from the steam.
2388    * @return KeyValue created by deserializing from <code>in</code> OR if we find a length
2389    * of zero, we will return null which can be useful marking a stream as done.
2390    * @throws IOException
2391    */
2392   public static KeyValue create(final DataInput in) throws IOException {
2393     return create(in.readInt(), in);
2394   }
2395 
2396   /**
2397    * Create a KeyValue reading <code>length</code> from <code>in</code>
2398    * @param length
2399    * @param in
2400    * @return Created KeyValue OR if we find a length of zero, we will return null which
2401    * can be useful marking a stream as done.
2402    * @throws IOException
2403    */
2404   public static KeyValue create(int length, final DataInput in) throws IOException {
2405 
2406     if (length <= 0) {
2407       if (length == 0) return null;
2408       throw new IOException("Failed read " + length + " bytes, stream corrupt?");
2409     }
2410 
2411     // This is how the old Writables.readFrom used to deserialize.  Didn't even vint.
2412     byte [] bytes = new byte[length];
2413     in.readFully(bytes);
2414     return new KeyValue(bytes, 0, length);
2415   }
2416 
2417   /**
2418    * Create a new KeyValue by copying existing cell and adding new tags
2419    * @param c
2420    * @param newTags
2421    * @return a new KeyValue instance with new tags
2422    */
2423   public static KeyValue cloneAndAddTags(Cell c, List<Tag> newTags) {
2424     List<Tag> existingTags = null;
2425     if(c.getTagsLength() > 0) {
2426       existingTags = Tag.asList(c.getTagsArray(), c.getTagsOffset(), c.getTagsLength());
2427       existingTags.addAll(newTags);
2428     } else {
2429       existingTags = newTags;
2430     }
2431     return new KeyValue(c.getRowArray(), c.getRowOffset(), (int)c.getRowLength(),
2432       c.getFamilyArray(), c.getFamilyOffset(), (int)c.getFamilyLength(),
2433       c.getQualifierArray(), c.getQualifierOffset(), (int) c.getQualifierLength(),
2434       c.getTimestamp(), Type.codeToType(c.getTypeByte()), c.getValueArray(), c.getValueOffset(),
2435       c.getValueLength(), existingTags);
2436   }
2437 
2438   /**
2439    * Create a KeyValue reading from the raw InputStream.
2440    * Named <code>iscreate</code> so doesn't clash with {@link #create(DataInput)}
2441    * @param in
2442    * @return Created KeyValue OR if we find a length of zero, we will return null which
2443    * can be useful marking a stream as done.
2444    * @throws IOException
2445    */
2446   public static KeyValue iscreate(final InputStream in) throws IOException {
2447     byte [] intBytes = new byte[Bytes.SIZEOF_INT];
2448     int bytesRead = 0;
2449     while (bytesRead < intBytes.length) {
2450       int n = in.read(intBytes, bytesRead, intBytes.length - bytesRead);
2451       if (n < 0) {
2452         if (bytesRead == 0) return null; // EOF at start is ok
2453         throw new IOException("Failed read of int, read " + bytesRead + " bytes");
2454       }
2455       bytesRead += n;
2456     }
2457     // TODO: perhaps some sanity check is needed here.
2458     byte [] bytes = new byte[Bytes.toInt(intBytes)];
2459     IOUtils.readFully(in, bytes, 0, bytes.length);
2460     return new KeyValue(bytes, 0, bytes.length);
2461   }
2462 
2463   /**
2464    * Write out a KeyValue in the manner in which we used to when KeyValue was a Writable.
2465    * @param kv
2466    * @param out
2467    * @return Length written on stream
2468    * @throws IOException
2469    * @see #create(DataInput) for the inverse function
2470    */
2471   public static long write(final KeyValue kv, final DataOutput out) throws IOException {
2472     // This is how the old Writables write used to serialize KVs.  Need to figure way to make it
2473     // work for all implementations.
2474     int length = kv.getLength();
2475     out.writeInt(length);
2476     out.write(kv.getBuffer(), kv.getOffset(), length);
2477     return length + Bytes.SIZEOF_INT;
2478   }
2479 
2480   /**
2481    * Write out a KeyValue in the manner in which we used to when KeyValue was a Writable but do
2482    * not require a {@link DataOutput}, just take plain {@link OutputStream}
2483    * Named <code>oswrite</code> so does not clash with {@link #write(KeyValue, DataOutput)}
2484    * @param kv
2485    * @param out
2486    * @return Length written on stream
2487    * @throws IOException
2488    * @see #create(DataInput) for the inverse function
2489    * @see #write(KeyValue, DataOutput)
2490    * @deprecated use {@link #oswrite(KeyValue, OutputStream, boolean)} instead
2491    */
2492   @Deprecated
2493   public static long oswrite(final KeyValue kv, final OutputStream out)
2494       throws IOException {
2495     int length = kv.getLength();
2496     // This does same as DataOuput#writeInt (big-endian, etc.)
2497     out.write(Bytes.toBytes(length));
2498     out.write(kv.getBuffer(), kv.getOffset(), length);
2499     return length + Bytes.SIZEOF_INT;
2500   }
2501 
2502   /**
2503    * Write out a KeyValue in the manner in which we used to when KeyValue was a Writable but do
2504    * not require a {@link DataOutput}, just take plain {@link OutputStream}
2505    * Named <code>oswrite</code> so does not clash with {@link #write(KeyValue, DataOutput)}
2506    * @param kv
2507    * @param out
2508    * @param withTags
2509    * @return Length written on stream
2510    * @throws IOException
2511    * @see #create(DataInput) for the inverse function
2512    * @see #write(KeyValue, DataOutput)
2513    * @see KeyValueUtil#oswrite(Cell, OutputStream, boolean)
2514    */
2515   public static long oswrite(final KeyValue kv, final OutputStream out, final boolean withTags)
2516       throws IOException {
2517     // In KeyValueUtil#oswrite we do a Cell serialization as KeyValue. Any changes doing here, pls
2518     // check KeyValueUtil#oswrite also and do necessary changes.
2519     int length = kv.getLength();
2520     if (!withTags) {
2521       length = kv.getKeyLength() + kv.getValueLength() + KEYVALUE_INFRASTRUCTURE_SIZE;
2522     }
2523     // This does same as DataOuput#writeInt (big-endian, etc.)
2524     StreamUtils.writeInt(out, length);
2525     out.write(kv.getBuffer(), kv.getOffset(), length);
2526     return length + Bytes.SIZEOF_INT;
2527   }
2528 
2529   /**
2530    * Comparator that compares row component only of a KeyValue.
2531    */
2532   public static class RowOnlyComparator implements Comparator<KeyValue> {
2533     final KVComparator comparator;
2534 
2535     public RowOnlyComparator(final KVComparator c) {
2536       this.comparator = c;
2537     }
2538 
2539     @Override
2540     public int compare(KeyValue left, KeyValue right) {
2541       return comparator.compareRows(left, right);
2542     }
2543   }
2544 
2545 
2546   /**
2547    * Avoids redundant comparisons for better performance.
2548    *
2549    * TODO get rid of this wart
2550    */
2551   public interface SamePrefixComparator<T> {
2552     /**
2553      * Compare two keys assuming that the first n bytes are the same.
2554      * @param commonPrefix How many bytes are the same.
2555      */
2556     int compareIgnoringPrefix(int commonPrefix, byte[] left, int loffset, int llength,
2557         byte[] right, int roffset, int rlength
2558     );
2559   }
2560 
2561   /**
2562    * This is a TEST only Comparator used in TestSeekTo and TestReseekTo.
2563    */
2564   public static class RawBytesComparator extends KVComparator {
2565     /**
2566      * The HFileV2 file format's trailer contains this class name.  We reinterpret this and
2567      * instantiate the appropriate comparator.
2568      * TODO: With V3 consider removing this.
2569      * @return legacy class name for FileFileTrailer#comparatorClassName
2570      */
2571     @Override
2572     public String getLegacyKeyComparatorName() {
2573       return "org.apache.hadoop.hbase.util.Bytes$ByteArrayComparator";
2574     }
2575 
2576     /**
2577      * @deprecated Since 0.99.2.
2578      */
2579     @Override
2580     @Deprecated
2581     public int compareFlatKey(byte[] left, int loffset, int llength, byte[] right,
2582         int roffset, int rlength) {
2583       return Bytes.BYTES_RAWCOMPARATOR.compare(left,  loffset, llength, right, roffset, rlength);
2584     }
2585 
2586     @Override
2587     public int compare(Cell left, Cell right) {
2588       return compareOnlyKeyPortion(left, right);
2589     }
2590 
2591     @Override
2592     @VisibleForTesting
2593     public int compareOnlyKeyPortion(Cell left, Cell right) {
2594       int c = Bytes.BYTES_RAWCOMPARATOR.compare(left.getRowArray(), left.getRowOffset(),
2595           left.getRowLength(), right.getRowArray(), right.getRowOffset(), right.getRowLength());
2596       if (c != 0) {
2597         return c;
2598       }
2599       c = Bytes.BYTES_RAWCOMPARATOR.compare(left.getFamilyArray(), left.getFamilyOffset(),
2600           left.getFamilyLength(), right.getFamilyArray(), right.getFamilyOffset(),
2601           right.getFamilyLength());
2602       if (c != 0) {
2603         return c;
2604       }
2605       c = Bytes.BYTES_RAWCOMPARATOR.compare(left.getQualifierArray(), left.getQualifierOffset(),
2606           left.getQualifierLength(), right.getQualifierArray(), right.getQualifierOffset(),
2607           right.getQualifierLength());
2608       if (c != 0) {
2609         return c;
2610       }
2611       c = compareTimestamps(left.getTimestamp(), right.getTimestamp());
2612       if (c != 0) {
2613         return c;
2614       }
2615       return (0xff & left.getTypeByte()) - (0xff & right.getTypeByte());
2616     }
2617 
2618     @Override
2619     public byte[] calcIndexKey(byte[] lastKeyOfPreviousBlock, byte[] firstKeyInBlock) {
2620       return firstKeyInBlock;
2621     }
2622 
2623   }
2624 
2625   /**
2626    * HeapSize implementation
2627    *
2628    * We do not count the bytes in the rowCache because it should be empty for a KeyValue in the
2629    * MemStore.
2630    */
2631   @Override
2632   public long heapSize() {
2633     int sum = 0;
2634     sum += ClassSize.OBJECT;// the KeyValue object itself
2635     sum += ClassSize.REFERENCE;// pointer to "bytes"
2636     sum += ClassSize.align(ClassSize.ARRAY);// "bytes"
2637     sum += ClassSize.align(length);// number of bytes of data in the "bytes" array
2638     sum += 2 * Bytes.SIZEOF_INT;// offset, length
2639     sum += Bytes.SIZEOF_LONG;// memstoreTS
2640     return ClassSize.align(sum);
2641   }
2642 
2643   /**
2644    * A simple form of KeyValue that creates a keyvalue with only the key part of the byte[]
2645    * Mainly used in places where we need to compare two cells.  Avoids copying of bytes
2646    * In places like block index keys, we need to compare the key byte[] with a cell.
2647    * Hence create a Keyvalue(aka Cell) that would help in comparing as two cells
2648    */
2649   public static class KeyOnlyKeyValue extends KeyValue {
2650     public KeyOnlyKeyValue() {
2651 
2652     }
2653     public KeyOnlyKeyValue(byte[] b) {
2654       this(b, 0, b.length);
2655     }
2656 
2657     public KeyOnlyKeyValue(byte[] b, int offset, int length) {
2658       this.bytes = b;
2659       this.length = length;
2660       this.offset = offset;
2661     }
2662 
2663     @Override
2664     public int getKeyOffset() {
2665       return this.offset;
2666     }
2667 
2668     /**
2669      * A setter that helps to avoid object creation every time and whenever
2670      * there is a need to create new KeyOnlyKeyValue.
2671      * @param key
2672      * @param offset
2673      * @param length
2674      */
2675     public void setKey(byte[] key, int offset, int length) {
2676       this.bytes = key;
2677       this.offset = offset;
2678       this.length = length;
2679     }
2680 
2681     @Override
2682     public byte[] getKey() {
2683       int keylength = getKeyLength();
2684       byte[] key = new byte[keylength];
2685       System.arraycopy(this.bytes, getKeyOffset(), key, 0, keylength);
2686       return key;
2687     }
2688 
2689     @Override
2690     public byte[] getRowArray() {
2691       return bytes;
2692     }
2693 
2694     @Override
2695     public int getRowOffset() {
2696       return getKeyOffset() + Bytes.SIZEOF_SHORT;
2697     }
2698 
2699     @Override
2700     public byte[] getFamilyArray() {
2701       return bytes;
2702     }
2703 
2704     @Override
2705     public byte getFamilyLength() {
2706       return this.bytes[getFamilyOffset() - 1];
2707     }
2708 
2709     @Override
2710     public int getFamilyOffset() {
2711       return this.offset + Bytes.SIZEOF_SHORT + getRowLength() + Bytes.SIZEOF_BYTE;
2712     }
2713 
2714     @Override
2715     public byte[] getQualifierArray() {
2716       return bytes;
2717     }
2718 
2719     @Override
2720     public int getQualifierLength() {
2721       return getQualifierLength(getRowLength(), getFamilyLength());
2722     }
2723 
2724     @Override
2725     public int getQualifierOffset() {
2726       return getFamilyOffset() + getFamilyLength();
2727     }
2728 
2729     @Override
2730     public int getKeyLength() {
2731       return length;
2732     }
2733 
2734     @Override
2735     public short getRowLength() {
2736       return Bytes.toShort(this.bytes, getKeyOffset());
2737     }
2738 
2739     @Override
2740     public byte getTypeByte() {
2741       return this.bytes[this.offset + getKeyLength() - 1];
2742     }
2743 
2744     private int getQualifierLength(int rlength, int flength) {
2745       return getKeyLength() - (int) getKeyDataStructureSize(rlength, flength, 0);
2746     }
2747 
2748     @Override
2749     public long getTimestamp() {
2750       int tsOffset = getTimestampOffset();
2751       return Bytes.toLong(this.bytes, tsOffset);
2752     }
2753 
2754     @Override
2755     public int getTimestampOffset() {
2756       return getKeyOffset() + getKeyLength() - TIMESTAMP_TYPE_SIZE;
2757     }
2758 
2759     @Override
2760     public byte[] getTagsArray() {
2761       return HConstants.EMPTY_BYTE_ARRAY;
2762     }
2763 
2764     @Override
2765     public int getTagsOffset() {
2766       return 0;
2767     }
2768 
2769     @Override
2770     public byte[] getValueArray() {
2771       throw new IllegalArgumentException("KeyOnlyKeyValue does not work with values.");
2772     }
2773 
2774     @Override
2775     public int getValueOffset() {
2776       throw new IllegalArgumentException("KeyOnlyKeyValue does not work with values.");
2777     }
2778 
2779     @Override
2780     public int getValueLength() {
2781       throw new IllegalArgumentException("KeyOnlyKeyValue does not work with values.");
2782     }
2783 
2784     @Override
2785     public int getTagsLength() {
2786       return 0;
2787     }
2788 
2789     @Override
2790     public String toString() {
2791       if (this.bytes == null || this.bytes.length == 0) {
2792         return "empty";
2793       }
2794       return keyToString(this.bytes, this.offset, getKeyLength()) + "/vlen=0/mvcc=0";
2795     }
2796 
2797     @Override
2798     public int hashCode() {
2799       return super.hashCode();
2800     }
2801 
2802     @Override
2803     public boolean equals(Object other) {
2804       return super.equals(other);
2805     }
2806   }
2807 }