/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.hbase.client;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.UUID;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellBuilder;
import org.apache.hadoop.hbase.CellBuilderType;
import org.apache.hadoop.hbase.ExtendedCell;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.io.HeapSize;
import org.apache.hadoop.hbase.security.access.Permission;
import org.apache.hadoop.hbase.security.visibility.CellVisibility;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.yetus.audience.InterfaceAudience;

/**
 * Used to perform Put operations for a single row.
 * <p>
 * To perform a Put, instantiate a Put object with the row to insert to, and for each column to be
 * inserted, execute {@link #addColumn(byte[], byte[], byte[]) add} or
 * {@link #addColumn(byte[], byte[], long, byte[]) add} if setting the timestamp.
 */
@InterfaceAudience.Public
public class Put extends Mutation implements HeapSize {
  /**
   * Create a Put operation for the specified row.
   * @param row row key
   */
  public Put(byte[] row) {
    this(row, HConstants.LATEST_TIMESTAMP);
  }

  /**
   * Create a Put operation for the specified row, using a given timestamp.
   * @param row row key; we make a copy of what we are passed to keep local.
   * @param ts  timestamp
   */
  public Put(byte[] row, long ts) {
    this(row, 0, row.length, ts);
  }

  /**
   * We make a copy of the passed in row key to keep local.
   */
  public Put(byte[] rowArray, int rowOffset, int rowLength) {
    this(rowArray, rowOffset, rowLength, HConstants.LATEST_TIMESTAMP);
  }

  /**
   * @param row row key; we make a copy of what we are passed to keep local.
   * @param ts  timestamp
   */
  public Put(ByteBuffer row, long ts) {
    if (ts < 0) {
      throw new IllegalArgumentException("Timestamp cannot be negative. ts=" + ts);
    }
    checkRow(row);
    this.row = new byte[row.remaining()];
    row.get(this.row);
    this.ts = ts;
  }

  /**
   * @param row row key; we make a copy of what we are passed to keep local.
   */
  public Put(ByteBuffer row) {
    this(row, HConstants.LATEST_TIMESTAMP);
  }

  /**
   * We make a copy of the passed in row key to keep local.
   */
  public Put(byte[] rowArray, int rowOffset, int rowLength, long ts) {
    checkRow(rowArray, rowOffset, rowLength);
    checkTimestamp(ts);
    this.row = Bytes.copy(rowArray, rowOffset, rowLength);
    this.ts = ts;
  }

  /**
   * Create a Put operation for an immutable row key.
   * @param row            row key
   * @param rowIsImmutable whether the input row is immutable. Set to true if the caller can
   *                       guarantee that the row will not be changed for the Put duration.
   */
  public Put(byte[] row, boolean rowIsImmutable) {
    this(row, HConstants.LATEST_TIMESTAMP, rowIsImmutable);
  }

  /**
   * Create a Put operation for an immutable row key, using a given timestamp.
   * @param row            row key
   * @param ts             timestamp
   * @param rowIsImmutable whether the input row is immutable. Set to true if the caller can
   *                       guarantee that the row will not be changed for the Put duration.
   */
  public Put(byte[] row, long ts, boolean rowIsImmutable) {
    // Check and set timestamp
    checkTimestamp(ts);
    this.ts = ts;

    // Deal with row according to rowIsImmutable
    checkRow(row);
    if (rowIsImmutable) { // Row is immutable
      this.row = row; // Do not make a local copy, but point to the provided byte array directly
    } else { // Row is not immutable
      this.row = Bytes.copy(row, 0, row.length); // Make a local copy
    }
  }

  /**
   * Copy constructor. Creates a Put operation cloned from the specified Put.
   * @param putToCopy put to copy
   */
  public Put(Put putToCopy) {
    super(putToCopy);
  }

  /**
   * Construct the Put with user defined data. NOTED: 1) all cells in the familyMap must have the
   * Type.Put 2) the row of each cell must be same with passed row.
   * @param row       row. CAN'T be null
   * @param ts        timestamp
   * @param familyMap the map to collect all cells internally. CAN'T be null
   */
  public Put(byte[] row, long ts, NavigableMap<byte[], List<Cell>> familyMap) {
    super(row, ts, familyMap);
  }

  /**
   * Add the specified column and value to this Put operation.
   * @param family    family name
   * @param qualifier column qualifier
   * @param value     column value
   */
  public Put addColumn(byte[] family, byte[] qualifier, byte[] value) {
    return addColumn(family, qualifier, this.ts, value);
  }

  /**
   * Add the specified column and value, with the specified timestamp as its version to this Put
   * operation.
   * @param family    family name
   * @param qualifier column qualifier
   * @param ts        version timestamp
   * @param value     column value
   */
  public Put addColumn(byte[] family, byte[] qualifier, long ts, byte[] value) {
    checkTimestamp(ts);
    List<ExtendedCell> list = getCellList(family);
    KeyValue kv = createPutKeyValue(family, qualifier, ts, value);
    list.add(kv);
    return this;
  }

  /**
   * Add the specified column and value, with the specified timestamp as its version to this Put
   * operation.
   * @param family    family name
   * @param qualifier column qualifier
   * @param ts        version timestamp
   * @param value     column value
   */
  public Put addColumn(byte[] family, ByteBuffer qualifier, long ts, ByteBuffer value) {
    checkTimestamp(ts);
    List<ExtendedCell> list = getCellList(family);
    KeyValue kv = createPutKeyValue(family, qualifier, ts, value, null);
    list.add(kv);
    return this;
  }

  /**
   * Add the specified KeyValue to this Put operation. Operation assumes that the passed KeyValue is
   * immutable and its backing array will not be modified for the duration of this Put.
   * @param cell individual cell
   * @throws java.io.IOException e
   */
  @Override
  public Put add(Cell cell) throws IOException {
    super.add(cell);
    return this;
  }

  @Override
  public Put setTimestamp(long timestamp) {
    super.setTimestamp(timestamp);
    return this;
  }

  @Override
  public Put setAttribute(String name, byte[] value) {
    return (Put) super.setAttribute(name, value);
  }

  @Override
  public Put setId(String id) {
    return (Put) super.setId(id);
  }

  @Override
  public Put setDurability(Durability d) {
    return (Put) super.setDurability(d);
  }

  @Override
  public Put setClusterIds(List<UUID> clusterIds) {
    return (Put) super.setClusterIds(clusterIds);
  }

  @Override
  public Put setCellVisibility(CellVisibility expression) {
    return (Put) super.setCellVisibility(expression);
  }

  @Override
  public Put setACL(String user, Permission perms) {
    return (Put) super.setACL(user, perms);
  }

  @Override
  public Put setACL(Map<String, Permission> perms) {
    return (Put) super.setACL(perms);
  }

  @Override
  public Put setTTL(long ttl) {
    return (Put) super.setTTL(ttl);
  }

  @Override
  public Put setPriority(int priority) {
    return (Put) super.setPriority(priority);
  }

  @Override
  public CellBuilder getCellBuilder(CellBuilderType type) {
    return getCellBuilder(type, Cell.Type.Put);
  }

}