001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018
019package org.apache.hadoop.hbase.client;
020
021import java.io.IOException;
022import java.nio.ByteBuffer;
023import java.util.ArrayList;
024import java.util.Arrays;
025import java.util.HashMap;
026import java.util.Iterator;
027import java.util.List;
028import java.util.Map;
029import java.util.NavigableMap;
030import java.util.Optional;
031import java.util.TreeMap;
032import java.util.UUID;
033import java.util.stream.Collectors;
034import org.apache.hadoop.hbase.Cell;
035import org.apache.hadoop.hbase.CellBuilder;
036import org.apache.hadoop.hbase.CellBuilderFactory;
037import org.apache.hadoop.hbase.CellBuilderType;
038import org.apache.hadoop.hbase.CellScannable;
039import org.apache.hadoop.hbase.CellScanner;
040import org.apache.hadoop.hbase.CellUtil;
041import org.apache.hadoop.hbase.ExtendedCell;
042import org.apache.hadoop.hbase.HConstants;
043import org.apache.hadoop.hbase.IndividualBytesFieldCell;
044import org.apache.hadoop.hbase.KeyValue;
045import org.apache.hadoop.hbase.PrivateCellUtil;
046import org.apache.hadoop.hbase.Tag;
047import org.apache.hadoop.hbase.exceptions.DeserializationException;
048import org.apache.hadoop.hbase.io.HeapSize;
049import org.apache.hadoop.hbase.security.access.AccessControlConstants;
050import org.apache.hadoop.hbase.security.access.AccessControlUtil;
051import org.apache.hadoop.hbase.security.access.Permission;
052import org.apache.hadoop.hbase.security.visibility.CellVisibility;
053import org.apache.hadoop.hbase.security.visibility.VisibilityConstants;
054import org.apache.hadoop.hbase.util.Bytes;
055import org.apache.hadoop.hbase.util.ClassSize;
056import org.apache.yetus.audience.InterfaceAudience;
057
058import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
059import org.apache.hbase.thirdparty.com.google.common.collect.ArrayListMultimap;
060import org.apache.hbase.thirdparty.com.google.common.collect.ListMultimap;
061import org.apache.hbase.thirdparty.com.google.common.io.ByteArrayDataInput;
062import org.apache.hbase.thirdparty.com.google.common.io.ByteArrayDataOutput;
063import org.apache.hbase.thirdparty.com.google.common.io.ByteStreams;
064
065import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
066import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
067
068@InterfaceAudience.Public
069public abstract class Mutation extends OperationWithAttributes implements Row, CellScannable,
070    HeapSize {
071  public static final long MUTATION_OVERHEAD = ClassSize.align(
072      // This
073      ClassSize.OBJECT +
074      // row + OperationWithAttributes.attributes
075      2 * ClassSize.REFERENCE +
076      // Timestamp
077      1 * Bytes.SIZEOF_LONG +
078      // durability
079      ClassSize.REFERENCE +
080      // familyMap
081      ClassSize.REFERENCE +
082      // familyMap
083      ClassSize.TREEMAP +
084      // priority
085      ClassSize.INTEGER
086  );
087
088  /**
089   * The attribute for storing the list of clusters that have consumed the change.
090   */
091  private static final String CONSUMED_CLUSTER_IDS = "_cs.id";
092
093  /**
094   * The attribute for storing TTL for the result of the mutation.
095   */
096  private static final String OP_ATTRIBUTE_TTL = "_ttl";
097
098  private static final String RETURN_RESULTS = "_rr_";
099
100  // TODO: row should be final
101  protected byte [] row = null;
102  protected long ts = HConstants.LATEST_TIMESTAMP;
103  protected Durability durability = Durability.USE_DEFAULT;
104
105  // TODO: familyMap should be final
106  // A Map sorted by column family.
107  protected NavigableMap<byte [], List<Cell>> familyMap;
108
109  /**
110   * empty construction.
111   * We need this empty construction to keep binary compatibility.
112   */
113  protected Mutation() {
114    this.familyMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
115  }
116
117  protected Mutation(Mutation clone) {
118    super(clone);
119    this.row = clone.getRow();
120    this.ts = clone.getTimestamp();
121    this.familyMap = clone.getFamilyCellMap().entrySet().stream().
122      collect(Collectors.toMap(e -> e.getKey(), e -> new ArrayList<>(e.getValue()), (k, v) -> {
123        throw new RuntimeException("collisions!!!");
124      }, () -> new TreeMap<>(Bytes.BYTES_COMPARATOR)));
125  }
126
127  /**
128   * Construct the mutation with user defined data.
129   * @param row row. CAN'T be null
130   * @param ts timestamp
131   * @param familyMap the map to collect all cells internally. CAN'T be null
132   */
133  protected Mutation(byte[] row, long ts, NavigableMap<byte [], List<Cell>> familyMap) {
134    this.row = Preconditions.checkNotNull(row);
135    if (row.length == 0) {
136      throw new IllegalArgumentException("Row can't be empty");
137    }
138    this.ts = ts;
139    this.familyMap = Preconditions.checkNotNull(familyMap);
140  }
141
142  @Override
143  public CellScanner cellScanner() {
144    return CellUtil.createCellScanner(getFamilyCellMap());
145  }
146
147  /**
148   * Creates an empty list if one doesn't exist for the given column family
149   * or else it returns the associated list of Cell objects.
150   *
151   * @param family column family
152   * @return a list of Cell objects, returns an empty list if one doesn't exist.
153   */
154  List<Cell> getCellList(byte[] family) {
155    List<Cell> list = getFamilyCellMap().get(family);
156    if (list == null) {
157      list = new ArrayList<>();
158      getFamilyCellMap().put(family, list);
159    }
160    return list;
161  }
162
163  /*
164   * Create a KeyValue with this objects row key and the Put identifier.
165   *
166   * @return a KeyValue with this objects row key and the Put identifier.
167   */
168  KeyValue createPutKeyValue(byte[] family, byte[] qualifier, long ts, byte[] value) {
169    return new KeyValue(this.row, family, qualifier, ts, KeyValue.Type.Put, value);
170  }
171
172  /**
173   * Create a KeyValue with this objects row key and the Put identifier.
174   * @param family
175   * @param qualifier
176   * @param ts
177   * @param value
178   * @param tags - Specify the Tags as an Array
179   * @return a KeyValue with this objects row key and the Put identifier.
180   */
181  KeyValue createPutKeyValue(byte[] family, byte[] qualifier, long ts, byte[] value, Tag[] tags) {
182    KeyValue kvWithTag = new KeyValue(this.row, family, qualifier, ts, value, tags);
183    return kvWithTag;
184  }
185
186  /*
187   * Create a KeyValue with this objects row key and the Put identifier.
188   *
189   * @return a KeyValue with this objects row key and the Put identifier.
190   */
191  KeyValue createPutKeyValue(byte[] family, ByteBuffer qualifier, long ts, ByteBuffer value,
192      Tag[] tags) {
193    return new KeyValue(this.row, 0, this.row == null ? 0 : this.row.length,
194        family, 0, family == null ? 0 : family.length,
195        qualifier, ts, KeyValue.Type.Put, value, tags != null ? Arrays.asList(tags) : null);
196  }
197
198  /**
199   * Compile the column family (i.e. schema) information
200   * into a Map. Useful for parsing and aggregation by debugging,
201   * logging, and administration tools.
202   * @return Map
203   */
204  @Override
205  public Map<String, Object> getFingerprint() {
206    Map<String, Object> map = new HashMap<>();
207    List<String> families = new ArrayList<>(getFamilyCellMap().entrySet().size());
208    // ideally, we would also include table information, but that information
209    // is not stored in each Operation instance.
210    map.put("families", families);
211    for (Map.Entry<byte [], List<Cell>> entry : getFamilyCellMap().entrySet()) {
212      families.add(Bytes.toStringBinary(entry.getKey()));
213    }
214    return map;
215  }
216
217  /**
218   * Compile the details beyond the scope of getFingerprint (row, columns,
219   * timestamps, etc.) into a Map along with the fingerprinted information.
220   * Useful for debugging, logging, and administration tools.
221   * @param maxCols a limit on the number of columns output prior to truncation
222   * @return Map
223   */
224  @Override
225  public Map<String, Object> toMap(int maxCols) {
226    // we start with the fingerprint map and build on top of it.
227    Map<String, Object> map = getFingerprint();
228    // replace the fingerprint's simple list of families with a
229    // map from column families to lists of qualifiers and kv details
230    Map<String, List<Map<String, Object>>> columns = new HashMap<>();
231    map.put("families", columns);
232    map.put("row", Bytes.toStringBinary(this.row));
233    int colCount = 0;
234    // iterate through all column families affected
235    for (Map.Entry<byte [], List<Cell>> entry : getFamilyCellMap().entrySet()) {
236      // map from this family to details for each cell affected within the family
237      List<Map<String, Object>> qualifierDetails = new ArrayList<>();
238      columns.put(Bytes.toStringBinary(entry.getKey()), qualifierDetails);
239      colCount += entry.getValue().size();
240      if (maxCols <= 0) {
241        continue;
242      }
243      // add details for each cell
244      for (Cell cell: entry.getValue()) {
245        if (--maxCols <= 0) {
246          continue;
247        }
248        Map<String, Object> cellMap = cellToStringMap(cell);
249        qualifierDetails.add(cellMap);
250      }
251    }
252    map.put("totalColumns", colCount);
253    // add the id if set
254    if (getId() != null) {
255      map.put("id", getId());
256    }
257    // Add the TTL if set
258    // Long.MAX_VALUE is the default, and is interpreted to mean this attribute
259    // has not been set.
260    if (getTTL() != Long.MAX_VALUE) {
261      map.put("ttl", getTTL());
262    }
263    map.put("ts", this.ts);
264    return map;
265  }
266
267  private static Map<String, Object> cellToStringMap(Cell c) {
268    Map<String, Object> stringMap = new HashMap<>();
269    stringMap.put("qualifier", Bytes.toStringBinary(c.getQualifierArray(), c.getQualifierOffset(),
270                c.getQualifierLength()));
271    stringMap.put("timestamp", c.getTimestamp());
272    stringMap.put("vlen", c.getValueLength());
273    List<Tag> tags = PrivateCellUtil.getTags(c);
274    if (tags != null) {
275      List<String> tagsString = new ArrayList<>(tags.size());
276      for (Tag t : tags) {
277        tagsString
278            .add((t.getType()) + ":" + Bytes.toStringBinary(Tag.cloneValue(t)));
279      }
280      stringMap.put("tag", tagsString);
281    }
282    return stringMap;
283  }
284
285  /**
286   * Set the durability for this mutation
287   * @param d
288   */
289  public Mutation setDurability(Durability d) {
290    this.durability = d;
291    return this;
292  }
293
294  /** Get the current durability */
295  public Durability getDurability() {
296    return this.durability;
297  }
298
299  /**
300   * Method for retrieving the put's familyMap
301   * @return familyMap
302   */
303  public NavigableMap<byte [], List<Cell>> getFamilyCellMap() {
304    return this.familyMap;
305  }
306
307  /**
308   * Method to check if the familyMap is empty
309   * @return true if empty, false otherwise
310   */
311  public boolean isEmpty() {
312    return getFamilyCellMap().isEmpty();
313  }
314
315  /**
316   * Method for retrieving the delete's row
317   * @return row
318   */
319  @Override
320  public byte [] getRow() {
321    return this.row;
322  }
323
324  /**
325   * Method for retrieving the timestamp.
326   *
327   * @return timestamp
328   */
329  public long getTimestamp() {
330    return this.ts;
331  }
332
333  /**
334   * Marks that the clusters with the given clusterIds have consumed the mutation
335   * @param clusterIds of the clusters that have consumed the mutation
336   */
337  public Mutation setClusterIds(List<UUID> clusterIds) {
338    ByteArrayDataOutput out = ByteStreams.newDataOutput();
339    out.writeInt(clusterIds.size());
340    for (UUID clusterId : clusterIds) {
341      out.writeLong(clusterId.getMostSignificantBits());
342      out.writeLong(clusterId.getLeastSignificantBits());
343    }
344    setAttribute(CONSUMED_CLUSTER_IDS, out.toByteArray());
345    return this;
346  }
347
348  /**
349   * @return the set of clusterIds that have consumed the mutation
350   */
351  public List<UUID> getClusterIds() {
352    List<UUID> clusterIds = new ArrayList<>();
353    byte[] bytes = getAttribute(CONSUMED_CLUSTER_IDS);
354    if(bytes != null) {
355      ByteArrayDataInput in = ByteStreams.newDataInput(bytes);
356      int numClusters = in.readInt();
357      for(int i=0; i<numClusters; i++){
358        clusterIds.add(new UUID(in.readLong(), in.readLong()));
359      }
360    }
361    return clusterIds;
362  }
363
364  /**
365   * Sets the visibility expression associated with cells in this Mutation.
366   * @param expression
367   */
368  public Mutation setCellVisibility(CellVisibility expression) {
369    this.setAttribute(VisibilityConstants.VISIBILITY_LABELS_ATTR_KEY,
370        toCellVisibility(expression).toByteArray());
371    return this;
372  }
373
374  /**
375   * @return CellVisibility associated with cells in this Mutation.
376   * @throws DeserializationException
377   */
378  public CellVisibility getCellVisibility() throws DeserializationException {
379    byte[] cellVisibilityBytes = this.getAttribute(VisibilityConstants.VISIBILITY_LABELS_ATTR_KEY);
380    if (cellVisibilityBytes == null) return null;
381    return toCellVisibility(cellVisibilityBytes);
382  }
383
384  /**
385   * Create a protocol buffer CellVisibility based on a client CellVisibility.
386   *
387   * @param cellVisibility
388   * @return a protocol buffer CellVisibility
389   */
390  static ClientProtos.CellVisibility toCellVisibility(CellVisibility cellVisibility) {
391    ClientProtos.CellVisibility.Builder builder = ClientProtos.CellVisibility.newBuilder();
392    builder.setExpression(cellVisibility.getExpression());
393    return builder.build();
394  }
395
396  /**
397   * Convert a protocol buffer CellVisibility to a client CellVisibility
398   *
399   * @param proto
400   * @return the converted client CellVisibility
401   */
402  private static CellVisibility toCellVisibility(ClientProtos.CellVisibility proto) {
403    if (proto == null) return null;
404    return new CellVisibility(proto.getExpression());
405  }
406
407  /**
408   * Convert a protocol buffer CellVisibility bytes to a client CellVisibility
409   *
410   * @param protoBytes
411   * @return the converted client CellVisibility
412   * @throws DeserializationException
413   */
414  private static CellVisibility toCellVisibility(byte[] protoBytes) throws DeserializationException {
415    if (protoBytes == null) return null;
416    ClientProtos.CellVisibility.Builder builder = ClientProtos.CellVisibility.newBuilder();
417    ClientProtos.CellVisibility proto = null;
418    try {
419      ProtobufUtil.mergeFrom(builder, protoBytes);
420      proto = builder.build();
421    } catch (IOException e) {
422      throw new DeserializationException(e);
423    }
424    return toCellVisibility(proto);
425  }
426
427  /**
428   * Number of KeyValues carried by this Mutation.
429   * @return the total number of KeyValues
430   */
431  public int size() {
432    int size = 0;
433    for (List<Cell> cells : getFamilyCellMap().values()) {
434      size += cells.size();
435    }
436    return size;
437  }
438
439  /**
440   * @return the number of different families
441   */
442  public int numFamilies() {
443    return getFamilyCellMap().size();
444  }
445
446  /**
447   * @return Calculate what Mutation adds to class heap size.
448   */
449  @Override
450  public long heapSize() {
451    long heapsize = MUTATION_OVERHEAD;
452    // Adding row
453    heapsize += ClassSize.align(ClassSize.ARRAY + this.row.length);
454
455    // Adding map overhead
456    heapsize +=
457      ClassSize.align(getFamilyCellMap().size() * ClassSize.MAP_ENTRY);
458    for(Map.Entry<byte [], List<Cell>> entry : getFamilyCellMap().entrySet()) {
459      //Adding key overhead
460      heapsize +=
461        ClassSize.align(ClassSize.ARRAY + entry.getKey().length);
462
463      //This part is kinds tricky since the JVM can reuse references if you
464      //store the same value, but have a good match with SizeOf at the moment
465      //Adding value overhead
466      heapsize += ClassSize.align(ClassSize.ARRAYLIST);
467      int size = entry.getValue().size();
468      heapsize += ClassSize.align(ClassSize.ARRAY +
469          size * ClassSize.REFERENCE);
470
471      for (Cell cell : entry.getValue()) {
472        heapsize += cell.heapSize();
473      }
474    }
475    heapsize += getAttributeSize();
476    heapsize += extraHeapSize();
477    return ClassSize.align(heapsize);
478  }
479
480  /**
481   * @return The serialized ACL for this operation, or null if none
482   */
483  public byte[] getACL() {
484    return getAttribute(AccessControlConstants.OP_ATTRIBUTE_ACL);
485  }
486
487  /**
488   * @param user User short name
489   * @param perms Permissions for the user
490   */
491  public Mutation setACL(String user, Permission perms) {
492    setAttribute(AccessControlConstants.OP_ATTRIBUTE_ACL,
493      AccessControlUtil.toUsersAndPermissions(user, perms).toByteArray());
494    return this;
495  }
496
497  /**
498   * @param perms A map of permissions for a user or users
499   */
500  public Mutation setACL(Map<String, Permission> perms) {
501    ListMultimap<String, Permission> permMap = ArrayListMultimap.create();
502    for (Map.Entry<String, Permission> entry : perms.entrySet()) {
503      permMap.put(entry.getKey(), entry.getValue());
504    }
505    setAttribute(AccessControlConstants.OP_ATTRIBUTE_ACL,
506      AccessControlUtil.toUsersAndPermissions(permMap).toByteArray());
507    return this;
508  }
509
510  /**
511   * Return the TTL requested for the result of the mutation, in milliseconds.
512   * @return the TTL requested for the result of the mutation, in milliseconds,
513   * or Long.MAX_VALUE if unset
514   */
515  public long getTTL() {
516    byte[] ttlBytes = getAttribute(OP_ATTRIBUTE_TTL);
517    if (ttlBytes != null) {
518      return Bytes.toLong(ttlBytes);
519    }
520    return Long.MAX_VALUE;
521  }
522
523  /**
524   * Set the TTL desired for the result of the mutation, in milliseconds.
525   * @param ttl the TTL desired for the result of the mutation, in milliseconds
526   * @return this
527   */
528  public Mutation setTTL(long ttl) {
529    setAttribute(OP_ATTRIBUTE_TTL, Bytes.toBytes(ttl));
530    return this;
531  }
532
533  /**
534   * @return current value for returnResults
535   */
536  // Used by Increment and Append only.
537  @InterfaceAudience.Private
538  protected boolean isReturnResults() {
539    byte[] v = getAttribute(RETURN_RESULTS);
540    return v == null ? true : Bytes.toBoolean(v);
541  }
542
543  @InterfaceAudience.Private
544  // Used by Increment and Append only.
545  protected Mutation setReturnResults(boolean returnResults) {
546    setAttribute(RETURN_RESULTS, Bytes.toBytes(returnResults));
547    return this;
548  }
549
550  /**
551   * Subclasses should override this method to add the heap size of their own fields.
552   * @return the heap size to add (will be aligned).
553   */
554  protected long extraHeapSize(){
555    return 0L;
556  }
557
558  /**
559   * Set the timestamp of the delete.
560   */
561  public Mutation setTimestamp(long timestamp) {
562    if (timestamp < 0) {
563      throw new IllegalArgumentException("Timestamp cannot be negative. ts=" + timestamp);
564    }
565    this.ts = timestamp;
566    return this;
567  }
568
569  /**
570   * A convenience method to determine if this object's familyMap contains
571   * a value assigned to the given family &amp; qualifier.
572   * Both given arguments must match the KeyValue object to return true.
573   *
574   * @param family column family
575   * @param qualifier column qualifier
576   * @return returns true if the given family and qualifier already has an
577   * existing KeyValue object in the family map.
578   */
579  public boolean has(byte [] family, byte [] qualifier) {
580    return has(family, qualifier, this.ts, HConstants.EMPTY_BYTE_ARRAY, true, true);
581  }
582
583  /**
584   * A convenience method to determine if this object's familyMap contains
585   * a value assigned to the given family, qualifier and timestamp.
586   * All 3 given arguments must match the KeyValue object to return true.
587   *
588   * @param family column family
589   * @param qualifier column qualifier
590   * @param ts timestamp
591   * @return returns true if the given family, qualifier and timestamp already has an
592   * existing KeyValue object in the family map.
593   */
594  public boolean has(byte [] family, byte [] qualifier, long ts) {
595    return has(family, qualifier, ts, HConstants.EMPTY_BYTE_ARRAY, false, true);
596  }
597
598  /**
599   * A convenience method to determine if this object's familyMap contains
600   * a value assigned to the given family, qualifier and timestamp.
601   * All 3 given arguments must match the KeyValue object to return true.
602   *
603   * @param family column family
604   * @param qualifier column qualifier
605   * @param value value to check
606   * @return returns true if the given family, qualifier and value already has an
607   * existing KeyValue object in the family map.
608   */
609  public boolean has(byte [] family, byte [] qualifier, byte [] value) {
610    return has(family, qualifier, this.ts, value, true, false);
611  }
612
613  /**
614   * A convenience method to determine if this object's familyMap contains
615   * the given value assigned to the given family, qualifier and timestamp.
616   * All 4 given arguments must match the KeyValue object to return true.
617   *
618   * @param family column family
619   * @param qualifier column qualifier
620   * @param ts timestamp
621   * @param value value to check
622   * @return returns true if the given family, qualifier timestamp and value
623   *   already has an existing KeyValue object in the family map.
624   */
625  public boolean has(byte [] family, byte [] qualifier, long ts, byte [] value) {
626    return has(family, qualifier, ts, value, false, false);
627  }
628
629  /**
630   * Returns a list of all KeyValue objects with matching column family and qualifier.
631   *
632   * @param family column family
633   * @param qualifier column qualifier
634   * @return a list of KeyValue objects with the matching family and qualifier,
635   *   returns an empty list if one doesn't exist for the given family.
636   */
637  public List<Cell> get(byte[] family, byte[] qualifier) {
638    List<Cell> filteredList = new ArrayList<>();
639    for (Cell cell: getCellList(family)) {
640      if (CellUtil.matchingQualifier(cell, qualifier)) {
641        filteredList.add(cell);
642      }
643    }
644    return filteredList;
645  }
646
647  /*
648   * Private method to determine if this object's familyMap contains
649   * the given value assigned to the given family, qualifier and timestamp
650   * respecting the 2 boolean arguments
651   *
652   * @param family
653   * @param qualifier
654   * @param ts
655   * @param value
656   * @param ignoreTS
657   * @param ignoreValue
658   * @return returns true if the given family, qualifier timestamp and value
659   * already has an existing KeyValue object in the family map.
660   */
661  protected boolean has(byte[] family, byte[] qualifier, long ts, byte[] value,
662      boolean ignoreTS, boolean ignoreValue) {
663    List<Cell> list = getCellList(family);
664    if (list.isEmpty()) {
665      return false;
666    }
667    // Boolean analysis of ignoreTS/ignoreValue.
668    // T T => 2
669    // T F => 3 (first is always true)
670    // F T => 2
671    // F F => 1
672    if (!ignoreTS && !ignoreValue) {
673      for (Cell cell : list) {
674        if (CellUtil.matchingFamily(cell, family) &&
675            CellUtil.matchingQualifier(cell, qualifier)  &&
676            CellUtil.matchingValue(cell, value) &&
677            cell.getTimestamp() == ts) {
678          return true;
679        }
680      }
681    } else if (ignoreValue && !ignoreTS) {
682      for (Cell cell : list) {
683        if (CellUtil.matchingFamily(cell, family) && CellUtil.matchingQualifier(cell, qualifier)
684            && cell.getTimestamp() == ts) {
685          return true;
686        }
687      }
688    } else if (!ignoreValue && ignoreTS) {
689      for (Cell cell : list) {
690        if (CellUtil.matchingFamily(cell, family) && CellUtil.matchingQualifier(cell, qualifier)
691            && CellUtil.matchingValue(cell, value)) {
692          return true;
693        }
694      }
695    } else {
696      for (Cell cell : list) {
697        if (CellUtil.matchingFamily(cell, family) &&
698            CellUtil.matchingQualifier(cell, qualifier)) {
699          return true;
700        }
701      }
702    }
703    return false;
704  }
705
706  /**
707   * @param row Row to check
708   * @throws IllegalArgumentException Thrown if <code>row</code> is empty or null or
709   * &gt; {@link HConstants#MAX_ROW_LENGTH}
710   * @return <code>row</code>
711   */
712  static byte [] checkRow(final byte [] row) {
713    return checkRow(row, 0, row == null? 0: row.length);
714  }
715
716  /**
717   * @param row Row to check
718   * @param offset
719   * @param length
720   * @throws IllegalArgumentException Thrown if <code>row</code> is empty or null or
721   * &gt; {@link HConstants#MAX_ROW_LENGTH}
722   * @return <code>row</code>
723   */
724  static byte [] checkRow(final byte [] row, final int offset, final int length) {
725    if (row == null) {
726      throw new IllegalArgumentException("Row buffer is null");
727    }
728    if (length == 0) {
729      throw new IllegalArgumentException("Row length is 0");
730    }
731    if (length > HConstants.MAX_ROW_LENGTH) {
732      throw new IllegalArgumentException("Row length " + length + " is > " +
733        HConstants.MAX_ROW_LENGTH);
734    }
735    return row;
736  }
737
738  static void checkRow(ByteBuffer row) {
739    if (row == null) {
740      throw new IllegalArgumentException("Row buffer is null");
741    }
742    if (row.remaining() == 0) {
743      throw new IllegalArgumentException("Row length is 0");
744    }
745    if (row.remaining() > HConstants.MAX_ROW_LENGTH) {
746      throw new IllegalArgumentException("Row length " + row.remaining() + " is > " +
747          HConstants.MAX_ROW_LENGTH);
748    }
749  }
750
751  Mutation add(Cell cell) throws IOException {
752    //Checking that the row of the kv is the same as the mutation
753    // TODO: It is fraught with risk if user pass the wrong row.
754    // Throwing the IllegalArgumentException is more suitable I'd say.
755    if (!CellUtil.matchingRows(cell, this.row)) {
756      throw new WrongRowIOException("The row in " + cell.toString() +
757        " doesn't match the original one " +  Bytes.toStringBinary(this.row));
758    }
759
760    byte[] family;
761
762    if (cell instanceof IndividualBytesFieldCell) {
763      family = cell.getFamilyArray();
764    } else {
765      family = CellUtil.cloneFamily(cell);
766    }
767
768    if (family == null || family.length == 0) {
769      throw new IllegalArgumentException("Family cannot be null");
770    }
771
772    if (cell instanceof ExtendedCell) {
773      getCellList(family).add(cell);
774    } else {
775      getCellList(family).add(new CellWrapper(cell));
776    }
777    return this;
778  }
779
780  /**
781   * get a CellBuilder instance that already has relevant Type and Row set.
782   * @param cellBuilderType e.g CellBuilderType.SHALLOW_COPY
783   * @return CellBuilder which already has relevant Type and Row set.
784   */
785  public abstract CellBuilder getCellBuilder(CellBuilderType cellBuilderType);
786
787  /**
788   * get a CellBuilder instance that already has relevant Type and Row set.
789   * the default CellBuilderType is CellBuilderType.SHALLOW_COPY
790   * @return CellBuilder which already has relevant Type and Row set.
791   */
792  public CellBuilder getCellBuilder() {
793    return getCellBuilder(CellBuilderType.SHALLOW_COPY);
794  }
795
796  /**
797   * get a CellBuilder instance that already has relevant Type and Row set.
798   * @param cellBuilderType e.g CellBuilderType.SHALLOW_COPY
799   * @param cellType e.g Cell.Type.Put
800   * @return CellBuilder which already has relevant Type and Row set.
801     */
802  protected CellBuilder getCellBuilder(CellBuilderType cellBuilderType, Cell.Type cellType) {
803    CellBuilder builder = CellBuilderFactory.create(cellBuilderType).setRow(row).setType(cellType);
804    return new CellBuilder() {
805      @Override
806      public CellBuilder setRow(byte[] row) {
807        return this;
808      }
809
810      @Override
811      public CellBuilder setType(Cell.Type type) {
812        return this;
813      }
814
815      @Override
816      public CellBuilder setRow(byte[] row, int rOffset, int rLength) {
817        return this;
818      }
819
820      @Override
821      public CellBuilder setFamily(byte[] family) {
822        builder.setFamily(family);
823        return this;
824      }
825
826      @Override
827      public CellBuilder setFamily(byte[] family, int fOffset, int fLength) {
828        builder.setFamily(family, fOffset, fLength);
829        return this;
830      }
831
832      @Override
833      public CellBuilder setQualifier(byte[] qualifier) {
834        builder.setQualifier(qualifier);
835        return this;
836      }
837
838      @Override
839      public CellBuilder setQualifier(byte[] qualifier, int qOffset, int qLength) {
840        builder.setQualifier(qualifier, qOffset, qLength);
841        return this;
842      }
843
844      @Override
845      public CellBuilder setTimestamp(long timestamp) {
846        builder.setTimestamp(timestamp);
847        return this;
848      }
849
850      @Override
851      public CellBuilder setValue(byte[] value) {
852        builder.setValue(value);
853        return this;
854      }
855
856      @Override
857      public CellBuilder setValue(byte[] value, int vOffset, int vLength) {
858        builder.setValue(value, vOffset, vLength);
859        return this;
860      }
861
862      @Override
863      public Cell build() {
864        return builder.build();
865      }
866
867      @Override
868      public CellBuilder clear() {
869        builder.clear();
870        // reset the row and type
871        builder.setRow(row);
872        builder.setType(cellType);
873        return this;
874      }
875    };
876  }
877
878  private static final class CellWrapper implements ExtendedCell {
879    private static final long FIXED_OVERHEAD = ClassSize.align(
880      ClassSize.OBJECT              // object header
881        + KeyValue.TIMESTAMP_SIZE       // timestamp
882        + Bytes.SIZEOF_LONG             // sequence id
883        + 1 * ClassSize.REFERENCE);     // references to cell
884    private final Cell cell;
885    private long sequenceId;
886    private long timestamp;
887
888    CellWrapper(Cell cell) {
889      assert !(cell instanceof ExtendedCell);
890      this.cell = cell;
891      this.sequenceId = cell.getSequenceId();
892      this.timestamp = cell.getTimestamp();
893    }
894
895    @Override
896    public void setSequenceId(long seqId) {
897      sequenceId = seqId;
898    }
899
900    @Override
901    public void setTimestamp(long ts) {
902      timestamp = ts;
903    }
904
905    @Override
906    public void setTimestamp(byte[] ts) {
907      timestamp = Bytes.toLong(ts);
908    }
909
910    @Override
911    public long getSequenceId() {
912      return sequenceId;
913    }
914
915    @Override
916    public byte[] getValueArray() {
917      return cell.getValueArray();
918    }
919
920    @Override
921    public int getValueOffset() {
922      return cell.getValueOffset();
923    }
924
925    @Override
926    public int getValueLength() {
927      return cell.getValueLength();
928    }
929
930    @Override
931    public byte[] getTagsArray() {
932      return cell.getTagsArray();
933    }
934
935    @Override
936    public int getTagsOffset() {
937      return cell.getTagsOffset();
938    }
939
940    @Override
941    public int getTagsLength() {
942      return cell.getTagsLength();
943    }
944
945    @Override
946    public byte[] getRowArray() {
947      return cell.getRowArray();
948    }
949
950    @Override
951    public int getRowOffset() {
952      return cell.getRowOffset();
953    }
954
955    @Override
956    public short getRowLength() {
957      return cell.getRowLength();
958    }
959
960    @Override
961    public byte[] getFamilyArray() {
962      return cell.getFamilyArray();
963    }
964
965    @Override
966    public int getFamilyOffset() {
967      return cell.getFamilyOffset();
968    }
969
970    @Override
971    public byte getFamilyLength() {
972      return cell.getFamilyLength();
973    }
974
975    @Override
976    public byte[] getQualifierArray() {
977      return cell.getQualifierArray();
978    }
979
980    @Override
981    public int getQualifierOffset() {
982      return cell.getQualifierOffset();
983    }
984
985    @Override
986    public int getQualifierLength() {
987      return cell.getQualifierLength();
988    }
989
990    @Override
991    public long getTimestamp() {
992      return timestamp;
993    }
994
995    @Override
996    public byte getTypeByte() {
997      return cell.getTypeByte();
998    }
999
1000    @Override
1001    public Optional<Tag> getTag(byte type) {
1002      return PrivateCellUtil.getTag(cell, type);
1003    }
1004
1005    @Override
1006    public Iterator<Tag> getTags() {
1007      return PrivateCellUtil.tagsIterator(cell);
1008    }
1009
1010    @Override
1011    public byte[] cloneTags() {
1012      return PrivateCellUtil.cloneTags(cell);
1013    }
1014
1015    private long heapOverhead() {
1016      return FIXED_OVERHEAD
1017        + ClassSize.ARRAY // row
1018        + getFamilyLength() == 0 ? 0 : ClassSize.ARRAY
1019        + getQualifierLength() == 0 ? 0 : ClassSize.ARRAY
1020        + getValueLength() == 0 ? 0 : ClassSize.ARRAY
1021        + getTagsLength() == 0 ? 0 : ClassSize.ARRAY;
1022    }
1023
1024    @Override
1025    public long heapSize() {
1026      return heapOverhead()
1027        + ClassSize.align(getRowLength())
1028        + ClassSize.align(getFamilyLength())
1029        + ClassSize.align(getQualifierLength())
1030        + ClassSize.align(getValueLength())
1031        + ClassSize.align(getTagsLength());
1032    }
1033  }
1034}