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