/**
 * 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.tajo.storage.hbase;

import com.google.common.base.Optional;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Sets;
import net.minidev.json.JSONObject;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.security.UserProvider;
import org.apache.hadoop.mapreduce.JobID;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.task.JobContextImpl;
import org.apache.tajo.*;
import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.statistics.TableStats;
import org.apache.tajo.common.TajoDataTypes.Type;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.datum.Datum;
import org.apache.tajo.datum.TextDatum;
import org.apache.tajo.exception.*;
import org.apache.tajo.plan.LogicalPlan;
import org.apache.tajo.plan.expr.*;
import org.apache.tajo.plan.logical.CreateTableNode;
import org.apache.tajo.plan.logical.LogicalNode;
import org.apache.tajo.plan.logical.NodeType;
import org.apache.tajo.plan.rewrite.LogicalPlanRewriteRuleContext;
import org.apache.tajo.plan.verifier.SyntaxErrorUtil;
import org.apache.tajo.storage.*;
import org.apache.tajo.storage.fragment.Fragment;
import org.apache.tajo.util.*;

import javax.annotation.Nullable;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.*;

/**
 * Tablespace for HBase table.
 */
public class HBaseTablespace extends Tablespace {
  private static final Log LOG = LogFactory.getLog(HBaseTablespace.class);

  public static final StorageProperty HBASE_STORAGE_PROPERTIES =
      new StorageProperty("hbase", false, true, false, false);
  public static final FormatProperty HFILE_FORMAT_PROPERTIES = new FormatProperty(true, false, true);
  public static final FormatProperty PUT_MODE_PROPERTIES = new FormatProperty(true, true, false);
  public static final List<byte []> EMPTY_START_ROW_KEY = Arrays.asList(new byte [0]);
  public static final List<byte []> EMPTY_END_ROW_KEY   = Arrays.asList(new byte [0]);

  private Configuration hbaseConf;

  private final static SortedInsertRewriter REWRITE_RULE = new SortedInsertRewriter();

  private Map<HConnectionKey, HConnection> connMap = new HashMap<HConnectionKey, HConnection>();

  public HBaseTablespace(String spaceName, URI uri, JSONObject config) {
    super(spaceName, uri, config);
  }

  @Override
  public void storageInit() throws IOException {
    this.hbaseConf = HBaseConfiguration.create(conf);
    String zkQuorum = extractQuorum(uri);
    String [] splits = zkQuorum.split(":");
    hbaseConf.set(HConstants.ZOOKEEPER_QUORUM, splits[0]);
    hbaseConf.set(HConstants.ZOOKEEPER_CLIENT_PORT, splits[1]);
  }

  public Configuration getHbaseConf() {
    return hbaseConf;
  }

  @Override
  public long getTableVolume(TableDesc table, Optional<EvalNode> filter) {
    long totalVolume = 0;
    try {
      for (HBaseFragment f : getRawSplits("", table, filter.orNull())) {
        if (f.getLength() > 0) {
          totalVolume += f.getLength();
        }
      }
    } catch (TajoException e) {
      throw new TajoRuntimeException(e);
    } catch (Throwable ioe) {
      throw new TajoInternalError(ioe);
    }
    return totalVolume;
  }

  @Override
  public void close() {
    synchronized (connMap) {
      for (HConnection eachConn: connMap.values()) {
        try {
          eachConn.close();
        } catch (Exception e) {
          LOG.error(e.getMessage(), e);
        }
      }
    }
  }

  @Override
  public void createTable(TableDesc tableDesc, boolean ifNotExists) throws TajoException, IOException {
    createTable(tableDesc.getUri(), tableDesc.getMeta(), tableDesc.getSchema(), tableDesc.isExternal(), ifNotExists);
    TableStats stats = new TableStats();
    stats.setNumRows(TajoConstants.UNKNOWN_ROW_NUMBER);
    tableDesc.setStats(stats);
  }

  private void createTable(URI uri, TableMeta tableMeta, Schema schema,
                           boolean isExternal, boolean ifNotExists) throws TajoException, IOException {
    String hbaseTableName = tableMeta.getOption(HBaseStorageConstants.META_TABLE_KEY, "");
    if (hbaseTableName == null || hbaseTableName.trim().isEmpty()) {
      throw new MissingTablePropertyException(HBaseStorageConstants.META_TABLE_KEY, "hbase");
    }
    TableName hTableName = TableName.valueOf(hbaseTableName);

    String mappedColumns = tableMeta.getOption(HBaseStorageConstants.META_COLUMNS_KEY, "");
    if (mappedColumns != null && mappedColumns.split(",").length > schema.size()) {
      throw new InvalidTablePropertyException(HBaseStorageConstants.META_COLUMNS_KEY,
          "mapping column pairs must be more than number of columns in the schema");
    }

    ColumnMapping columnMapping = new ColumnMapping(schema, tableMeta.getOptions());
    int numRowKeys = 0;
    boolean[] isRowKeyMappings = columnMapping.getIsRowKeyMappings();
    for (boolean isRowKeyMapping : isRowKeyMappings) {
      if (isRowKeyMapping) {
        numRowKeys++;
      }
    }
    if (numRowKeys > 1) {
      for (int i = 0; i < isRowKeyMappings.length; i++) {
        if (isRowKeyMappings[i] && schema.getColumn(i).getDataType().getType() != Type.TEXT) {
          throw SyntaxErrorUtil.makeSyntaxError("Key field type should be TEXT type.");
        }
      }
    }

    for (int i = 0; i < isRowKeyMappings.length; i++) {
      if (columnMapping.getIsColumnKeys()[i] && schema.getColumn(i).getDataType().getType() != Type.TEXT) {
        throw SyntaxErrorUtil.makeSyntaxError("Column key field('<cfname>:key:') type should be TEXT type.");
      }
      if (columnMapping.getIsColumnValues()[i] && schema.getColumn(i).getDataType().getType() != Type.TEXT) {
        throw SyntaxErrorUtil.makeSyntaxError("Column value field(('<cfname>:value:') type should be TEXT type.");
      }
    }

    HBaseAdmin hAdmin =  new HBaseAdmin(getHbaseConf());

    try {
      if (isExternal) {
        // If tajo table is external table, only check validation.
        if (mappedColumns == null || mappedColumns.isEmpty()) {
          throw new MissingTablePropertyException(HBaseStorageConstants.META_COLUMNS_KEY, hbaseTableName);
        }
        if (!hAdmin.tableExists(hTableName)) {
          throw new UnavailableTableLocationException(hbaseTableName, "the table does not exist");
        }
        HTableDescriptor hTableDescriptor = hAdmin.getTableDescriptor(hTableName);
        Set<String> tableColumnFamilies = new HashSet<>();
        for (HColumnDescriptor eachColumn : hTableDescriptor.getColumnFamilies()) {
          tableColumnFamilies.add(eachColumn.getNameAsString());
        }

        Collection<String> mappingColumnFamilies = columnMapping.getColumnFamilyNames();
        if (mappingColumnFamilies.isEmpty()) {
          throw new MissingTablePropertyException(HBaseStorageConstants.META_COLUMNS_KEY, hbaseTableName);
        }

        for (String eachMappingColumnFamily : mappingColumnFamilies) {
          if (!tableColumnFamilies.contains(eachMappingColumnFamily)) {
            throw new IOException("There is no " + eachMappingColumnFamily + " column family in " + hbaseTableName);
          }
        }
      } else {
        if (hAdmin.tableExists(hbaseTableName)) {
          if (ifNotExists) {
            return;
          } else {
            throw new IOException("HBase table [" + hbaseTableName + "] already exists.");
          }
        }
        // Creating hbase table
        HTableDescriptor hTableDescriptor = parseHTableDescriptor(tableMeta, schema);

        byte[][] splitKeys = getSplitKeys(conf, hbaseTableName, schema, tableMeta);
        if (splitKeys == null) {
          hAdmin.createTable(hTableDescriptor);
        } else {
          hAdmin.createTable(hTableDescriptor, splitKeys);
        }
      }
    } finally {
      hAdmin.close();
    }
  }

  /**
   * Returns initial region split keys.
   *
   * @param conf
   * @param schema
   * @param meta
   * @return
   * @throws java.io.IOException
   */
  private byte[][] getSplitKeys(TajoConf conf, String hbaseTableName, Schema schema, TableMeta meta)
      throws MissingTablePropertyException, InvalidTablePropertyException, IOException {

    String splitRowKeys = meta.getOption(HBaseStorageConstants.META_SPLIT_ROW_KEYS_KEY, "");
    String splitRowKeysFile = meta.getOption(HBaseStorageConstants.META_SPLIT_ROW_KEYS_FILE_KEY, "");

    if ((splitRowKeys == null || splitRowKeys.isEmpty()) &&
        (splitRowKeysFile == null || splitRowKeysFile.isEmpty())) {
      return null;
    }

    ColumnMapping columnMapping = new ColumnMapping(schema, meta.getOptions());
    boolean[] isBinaryColumns = columnMapping.getIsBinaryColumns();
    boolean[] isRowKeys = columnMapping.getIsRowKeyMappings();

    boolean rowkeyBinary = false;
    int numRowKeys = 0;
    Column rowKeyColumn = null;
    for (int i = 0; i < isBinaryColumns.length; i++) {
      if (isBinaryColumns[i] && isRowKeys[i]) {
        rowkeyBinary = true;
      }
      if (isRowKeys[i]) {
        numRowKeys++;
        rowKeyColumn = schema.getColumn(i);
      }
    }

    if (rowkeyBinary && numRowKeys > 1) {
      throw new InvalidTablePropertyException("If rowkey is mapped to multi column and a rowkey is binary, " +
          "Multiple region for creation is not support.", hbaseTableName);
    }

    if (splitRowKeys != null && !splitRowKeys.isEmpty()) {
      String[] splitKeyTokens = splitRowKeys.split(",");
      byte[][] splitKeys = new byte[splitKeyTokens.length][];
      for (int i = 0; i < splitKeyTokens.length; i++) {
        if (numRowKeys == 1 && rowkeyBinary) {
          splitKeys[i] = HBaseBinarySerializerDeserializer.serialize(rowKeyColumn, new TextDatum(splitKeyTokens[i]));
        } else {
          splitKeys[i] = HBaseTextSerializerDeserializer.serialize(rowKeyColumn, new TextDatum(splitKeyTokens[i]));
        }
      }
      return splitKeys;
    }

    if (splitRowKeysFile != null && !splitRowKeysFile.isEmpty()) {
      // If there is many split keys, Tajo allows to define in the file.
      Path path = new Path(splitRowKeysFile);
      FileSystem fs = path.getFileSystem(conf);

      if (!fs.exists(path)) {
        throw new MissingTablePropertyException("hbase.split.rowkeys.file=" + path.toString() + " not exists.",
            hbaseTableName);
      }

      SortedSet<String> splitKeySet = new TreeSet<>();
      BufferedReader reader = null;

      try {
        reader = new BufferedReader(new InputStreamReader(fs.open(path)));
        String line = null;
        while ( (line = reader.readLine()) != null ) {
          if (line.isEmpty()) {
            continue;
          }
          splitKeySet.add(line);
        }
      } finally {
        if (reader != null) {
          reader.close();
        }
      }

      if (splitKeySet.isEmpty()) {
        return null;
      }

      byte[][] splitKeys = new byte[splitKeySet.size()][];
      int index = 0;
      for (String eachKey: splitKeySet) {
        if (numRowKeys == 1 && rowkeyBinary) {
          splitKeys[index++] = HBaseBinarySerializerDeserializer.serialize(rowKeyColumn, new TextDatum(eachKey));
        } else {
          splitKeys[index++] = HBaseTextSerializerDeserializer.serialize(rowKeyColumn, new TextDatum(eachKey));
        }
      }

      return splitKeys;
    }

    return null;
  }

  /**
   * It extracts quorum addresses from a Hbase Tablespace URI.
   * For example, consider an example URI 'hbase:zk://host1:2171,host2:2172,host3:2173/table1'.
   * <code>extractQuorum</code> will extract only 'host1:2171,host2:2172,host3:2173'.
   *
   * @param uri Hbase Tablespace URI
   * @return Quorum addresses
   */
  static String extractQuorum(URI uri) {
    String uriStr = uri.toString();
    int start = uriStr.indexOf("/") + 2;
    int pathIndex = uriStr.lastIndexOf("/");

    if (pathIndex < start) {
      return uriStr.substring(start);
    } else {
      return uriStr.substring(start, pathIndex);
    }
  }

  /**
   * Creates HTableDescription using table meta data.
   *
   * @param tableMeta
   * @param schema
   * @return
   * @throws java.io.IOException
   */
  public static HTableDescriptor parseHTableDescriptor(TableMeta tableMeta, Schema schema)
      throws MissingTablePropertyException, InvalidTablePropertyException {

    String hbaseTableName = tableMeta.getOption(HBaseStorageConstants.META_TABLE_KEY, "");
    if (hbaseTableName == null || hbaseTableName.trim().isEmpty()) {
      throw new MissingTablePropertyException(HBaseStorageConstants.META_TABLE_KEY, hbaseTableName);
    }
    TableName hTableName = TableName.valueOf(hbaseTableName);

    ColumnMapping columnMapping = new ColumnMapping(schema, tableMeta.getOptions());

    HTableDescriptor hTableDescriptor = new HTableDescriptor(hTableName);

    Collection<String> columnFamilies = columnMapping.getColumnFamilyNames();
    //If 'columns' attribute is empty, Tajo table columns are mapped to all HBase table column.
    if (columnFamilies.isEmpty()) {
      for (Column eachColumn: schema.getRootColumns()) {
        columnFamilies.add(eachColumn.getSimpleName());
      }
    }

    for (String eachColumnFamily: columnFamilies) {
      hTableDescriptor.addFamily(new HColumnDescriptor(eachColumnFamily));
    }

    return hTableDescriptor;
  }

  @Override
  public void purgeTable(TableDesc tableDesc) throws IOException, TajoException {
    HBaseAdmin hAdmin =  new HBaseAdmin(hbaseConf);

    try {
      HTableDescriptor hTableDesc = parseHTableDescriptor(tableDesc.getMeta(), tableDesc.getSchema());
      LOG.info("Deleting hbase table: " + new String(hTableDesc.getName()));
      hAdmin.disableTable(hTableDesc.getName());
      hAdmin.deleteTable(hTableDesc.getName());
    } finally {
      hAdmin.close();
    }
  }

  @Override
  public URI getTableUri(String databaseName, String tableName) {
    return URI.create(uri.toString() + "/" + tableName);
  }

  /**
   * Returns columns which are mapped to the rowkey of the hbase table.
   *
   * @param tableDesc
   * @return
   * @throws java.io.IOException
   */
  private Column[] getIndexableColumns(TableDesc tableDesc) throws
      MissingTablePropertyException, InvalidTablePropertyException {

    ColumnMapping columnMapping = new ColumnMapping(tableDesc.getSchema(), tableDesc.getMeta().getOptions());
    boolean[] isRowKeyMappings = columnMapping.getIsRowKeyMappings();
    int[] rowKeyIndexes = columnMapping.getRowKeyFieldIndexes();

    Column indexColumn = null;
    for (int i = 0; i < isRowKeyMappings.length; i++) {
      if (isRowKeyMappings[i]) {
        if (columnMapping.getNumRowKeys() == 1 ||
            rowKeyIndexes[i] == 0) {
          indexColumn = tableDesc.getSchema().getColumn(i);
        }
      }
    }
    return new Column[]{indexColumn};
  }

  private Pair<List<byte []>, List<byte []>> getSelectedKeyRange(
      ColumnMapping columnMap,
      List<IndexPredication> predicates) {

    final List<byte[]> startRows;
    final List<byte[]> stopRows;

    if (predicates != null && !predicates.isEmpty()) {
      startRows = new ArrayList<>();
      stopRows = new ArrayList<>();

      // indexPredications is Disjunctive set
      for (IndexPredication pred: predicates) {
        if (pred.getStartValue() != null) {
          startRows.add(serialize(columnMap, pred, pred.getStartValue()));
        } else {
          startRows.add(HConstants.EMPTY_START_ROW);
        }

        if (pred.getStopValue() != null) {
          stopRows.add(serialize(columnMap, pred, pred.getStopValue()));
        } else {
          stopRows.add(HConstants.EMPTY_START_ROW);
        }
      }
    } else {
      startRows = EMPTY_START_ROW_KEY;
      stopRows  = EMPTY_END_ROW_KEY;
    }

    return new Pair(startRows, stopRows);
  }

  private boolean isEmptyRegion(org.apache.hadoop.hbase.util.Pair<byte[][], byte[][]> tableRange) {
    return tableRange == null || tableRange.getFirst() == null || tableRange.getFirst().length == 0;
  }

  private long getRegionSize(RegionSizeCalculator calculator, byte [] regionName) {
    long regionSize = calculator.getRegionSize(regionName);
    if (regionSize == 0) {
      return TajoConstants.UNKNOWN_LENGTH;
    } else {
      return regionSize;
    }
  }

  private List<HBaseFragment> createEmptyFragment(TableDesc table, String sourceId, HTable htable,
                                                  RegionSizeCalculator sizeCalculator) throws IOException {
    HRegionLocation regLoc = htable.getRegionLocation(HConstants.EMPTY_BYTE_ARRAY, false);
    if (null == regLoc) {
      throw new IOException("Expecting at least one region.");
    }

    HBaseFragment fragment = new HBaseFragment(
        table.getUri(),
        sourceId, htable.getName().getNameAsString(),
        HConstants.EMPTY_BYTE_ARRAY,
        HConstants.EMPTY_BYTE_ARRAY,
        regLoc.getHostname());

    fragment.setLength(getRegionSize(sizeCalculator, regLoc.getRegionInfo().getRegionName()));
    return ImmutableList.of(fragment);
  }

  private Collection<HBaseFragment> convertRangeToFragment(
      TableDesc table, String inputSourceId, HTable htable, RegionSizeCalculator sizeCalculator,
      org.apache.hadoop.hbase.util.Pair<byte[][], byte[][]> tableRange,
      Pair<List<byte[]>, List<byte[]>> selectedRange) throws IOException {

    final Map<byte[], HBaseFragment> fragmentMap = new HashMap<>();

    for (int i = 0; i < tableRange.getFirst().length; i++) {
      HRegionLocation location = htable.getRegionLocation(tableRange.getFirst()[i], false);
      if (location == null) {
        throw new IOException("Can't find the region of the key: " + Bytes.toStringBinary(tableRange.getFirst()[i]));
      }

      final byte[] regionStartKey = tableRange.getFirst()[i];
      final byte[] regionStopKey = tableRange.getSecond()[i];

      int startRowsSize = selectedRange.getFirst().size();
      for (int j = 0; j < startRowsSize; j++) {
        byte[] startRow = selectedRange.getFirst().get(j);
        byte[] stopRow = selectedRange.getSecond().get(j);
        // determine if the given start an stop key fall into the region
        if ((startRow.length == 0 || regionStopKey.length == 0 || Bytes.compareTo(startRow, regionStopKey) < 0)
            && (stopRow.length == 0 || Bytes.compareTo(stopRow, regionStartKey) > 0)) {
          final byte[] fragmentStart = (startRow.length == 0 || Bytes.compareTo(regionStartKey, startRow) >= 0) ?
              regionStartKey : startRow;

          final byte[] fragmentStop = (stopRow.length == 0 || Bytes.compareTo(regionStopKey, stopRow) <= 0) &&
              regionStopKey.length > 0 ? regionStopKey : stopRow;

          if (fragmentMap.containsKey(regionStartKey)) {
            final HBaseFragment prevFragment = fragmentMap.get(regionStartKey);
            if (Bytes.compareTo(fragmentStart, prevFragment.getStartRow()) < 0) {
              prevFragment.setStartRow(fragmentStart);
            }
            if (Bytes.compareTo(fragmentStop, prevFragment.getStopRow()) > 0) {
              prevFragment.setStopRow(fragmentStop);
            }
          } else {

            final HBaseFragment fragment = new HBaseFragment(table.getUri(),
                inputSourceId,
                htable.getName().getNameAsString(),
                fragmentStart,
                fragmentStop,
                location.getHostname());

            fragment.setLength(getRegionSize(sizeCalculator, location.getRegionInfo().getRegionName()));
            fragmentMap.put(regionStartKey, fragment);

            if (LOG.isDebugEnabled()) {
              LOG.debug("getFragments: fragment -> " + i + " -> " + fragment);
            }
          }
        }
      }
    }

    return fragmentMap.values();
  }

  @Override
  public List<Fragment> getSplits(String inputSourceId,
                                  TableDesc table,
                                  @Nullable EvalNode filterCondition) throws IOException, TajoException {
    return (List<Fragment>) (List) getRawSplits(inputSourceId, table, filterCondition);
  }

  private List<HBaseFragment> getRawSplits(String inputSourceId,
                                           TableDesc table,
                                           @Nullable EvalNode filterCondition) throws IOException, TajoException {
    final ColumnMapping columnMapping = new ColumnMapping(table.getSchema(), table.getMeta().getOptions());

    try (final HTable htable = new HTable(hbaseConf, table.getMeta().getOption(HBaseStorageConstants.META_TABLE_KEY))) {
      final RegionSizeCalculator sizeCalculator = new RegionSizeCalculator(htable);
      final org.apache.hadoop.hbase.util.Pair<byte[][], byte[][]> tableRange = htable.getStartEndKeys();
      if (isEmptyRegion(tableRange)) {
        return createEmptyFragment(table, inputSourceId, htable, sizeCalculator);
      }

      final Pair<List<byte []>, List<byte []>> selectedRange = getSelectedKeyRange(
          columnMapping,
          getIndexPredications(columnMapping, table, filterCondition));

      // region startkey -> HBaseFragment
      List<HBaseFragment> fragments = new ArrayList<>(convertRangeToFragment(table, inputSourceId, htable, sizeCalculator, tableRange, selectedRange));
      Collections.sort(fragments);
      if (!fragments.isEmpty()) {
        fragments.get(fragments.size() - 1).setLast(true);
      }

      return ImmutableList.copyOf(fragments);
    }
  }

  private byte[] serialize(ColumnMapping columnMapping,
                           IndexPredication indexPredication, Datum datum) {
    if (columnMapping.getIsBinaryColumns()[indexPredication.getColumnId()]) {
      return HBaseBinarySerializerDeserializer.serialize(indexPredication.getColumn(), datum);
    } else {
      return HBaseTextSerializerDeserializer.serialize(indexPredication.getColumn(), datum);
    }
  }

  @Override
  public Appender getAppenderForInsertRow(OverridableConf queryContext,
                                          TaskAttemptId taskAttemptId,
                                          TableMeta meta,
                                          Schema schema,
                                          Path workDir) throws IOException {
    return new HBasePutAppender(conf, uri, taskAttemptId, schema, meta, workDir);
  }

  @Override
  public Appender getAppender(OverridableConf queryContext,
                              TaskAttemptId taskAttemptId, TableMeta meta, Schema schema, Path workDir)
      throws IOException {
    if ("true".equalsIgnoreCase(queryContext.get(HBaseStorageConstants.INSERT_PUT_MODE, "false"))) {
      return new HBasePutAppender(conf, uri, taskAttemptId, schema, meta, workDir);
    } else {
      return super.getAppender(queryContext, taskAttemptId, meta, schema, workDir);
    }
  }

  public HConnection getConnection() throws IOException {
    synchronized(connMap) {
      HConnectionKey key = new HConnectionKey(hbaseConf);
      HConnection conn = connMap.get(key);
      if (conn == null) {
        conn = HConnectionManager.createConnection(hbaseConf);
        connMap.put(key, conn);
      }

      return conn;
    }
  }

  static class HConnectionKey {
    final static String[] CONNECTION_PROPERTIES = new String[] {
        HConstants.ZOOKEEPER_QUORUM, HConstants.ZOOKEEPER_ZNODE_PARENT,
        HConstants.ZOOKEEPER_CLIENT_PORT,
        HConstants.ZOOKEEPER_RECOVERABLE_WAITTIME,
        HConstants.HBASE_CLIENT_PAUSE, HConstants.HBASE_CLIENT_RETRIES_NUMBER,
        HConstants.HBASE_RPC_TIMEOUT_KEY,
        HConstants.HBASE_META_SCANNER_CACHING,
        HConstants.HBASE_CLIENT_INSTANCE_ID,
        HConstants.RPC_CODEC_CONF_KEY };

    private Map<String, String> properties;
    private String username;

    HConnectionKey(Configuration conf) {
      Map<String, String> m = new HashMap<>();
      if (conf != null) {
        for (String property : CONNECTION_PROPERTIES) {
          String value = conf.get(property);
          if (value != null) {
            m.put(property, value);
          }
        }
      }
      this.properties = Collections.unmodifiableMap(m);

      try {
        UserProvider provider = UserProvider.instantiate(conf);
        User currentUser = provider.getCurrent();
        if (currentUser != null) {
          username = currentUser.getName();
        }
      } catch (IOException ioe) {
        LOG.warn("Error obtaining current user, skipping username in HConnectionKey", ioe);
      }
    }

    @Override
    public int hashCode() {
      final int prime = 31;
      int result = 1;
      if (username != null) {
        result = username.hashCode();
      }
      for (String property : CONNECTION_PROPERTIES) {
        String value = properties.get(property);
        if (value != null) {
          result = prime * result + value.hashCode();
        }
      }

      return result;
    }

    @Override
    public boolean equals(Object obj) {
      if (this == obj)
        return true;
      if (obj == null)
        return false;
      if (getClass() != obj.getClass())
        return false;
      HConnectionKey that = (HConnectionKey) obj;
      if (this.username != null && !this.username.equals(that.username)) {
        return false;
      } else if (this.username == null && that.username != null) {
        return false;
      }
      if (this.properties == null) {
        if (that.properties != null) {
          return false;
        }
      } else {
        if (that.properties == null) {
          return false;
        }
        for (String property : CONNECTION_PROPERTIES) {
          String thisValue = this.properties.get(property);
          String thatValue = that.properties.get(property);
          // noinspection StringEquality
          if (thisValue == thatValue) {
            continue;
          }
          if (thisValue == null || !thisValue.equals(thatValue)) {
            return false;
          }
        }
      }
      return true;
    }

    @Override
    public String toString() {
      return "HConnectionKey{ properties=" + properties + ", username='" + username + '\'' + '}';
    }
  }

  public List<IndexPredication> getIndexPredications(ColumnMapping columnMapping,
                                                     TableDesc tableDesc,
                                                     @Nullable EvalNode filterCondition)
      throws IOException, MissingTablePropertyException, InvalidTablePropertyException {

    List<IndexPredication> indexPredications = new ArrayList<IndexPredication>();
    Column[] indexableColumns = getIndexableColumns(tableDesc);
    if (indexableColumns != null && indexableColumns.length == 1) {
      // Currently supports only single index column.
      List<Set<EvalNode>> indexablePredicateList = findIndexablePredicateSet(filterCondition, indexableColumns);
      for (Set<EvalNode> eachEvalSet: indexablePredicateList) {
        Pair<Datum, Datum> indexPredicationValues = getIndexablePredicateValue(columnMapping, eachEvalSet);
        if (indexPredicationValues != null) {
          IndexPredication indexPredication = new IndexPredication();
          indexPredication.setColumn(indexableColumns[0]);
          indexPredication.setColumnId(tableDesc.getLogicalSchema().getColumnId(indexableColumns[0].getQualifiedName()));
          indexPredication.setStartValue(indexPredicationValues.getFirst());
          indexPredication.setStopValue(indexPredicationValues.getSecond());

          indexPredications.add(indexPredication);
        }
      }
    }
    return indexPredications;
  }

  public List<Set<EvalNode>> findIndexablePredicateSet(@Nullable EvalNode qual,
                                                       Column[] indexableColumns) throws IOException {
    List<Set<EvalNode>> indexablePredicateList = new ArrayList<>();

    // if a query statement has a search condition, try to find indexable predicates
    if (indexableColumns != null && qual != null) {
      EvalNode[] disjunctiveForms = AlgebraicUtil.toDisjunctiveNormalFormArray(qual);

      // add qualifier to schema for qual
      for (Column column : indexableColumns) {
        for (EvalNode disjunctiveExpr : disjunctiveForms) {
          EvalNode[] conjunctiveForms = AlgebraicUtil.toConjunctiveNormalFormArray(disjunctiveExpr);
          Set<EvalNode> indexablePredicateSet = Sets.newHashSet();
          for (EvalNode conjunctiveExpr : conjunctiveForms) {
            if (checkIfIndexablePredicateOnTargetColumn(conjunctiveExpr, column)) {
              indexablePredicateSet.add(conjunctiveExpr);
            }
          }
          if (!indexablePredicateSet.isEmpty()) {
            indexablePredicateList.add(indexablePredicateSet);
          }
        }
      }
    }

    return indexablePredicateList;
  }

  private boolean checkIfIndexablePredicateOnTargetColumn(EvalNode evalNode, Column targetColumn) {
    if (checkIfIndexablePredicate(evalNode) || checkIfConjunctiveButOneVariable(evalNode)) {
      Set<Column> variables = EvalTreeUtil.findUniqueColumns(evalNode);
      // if it contains only single variable matched to a target column
      return variables.size() == 1 && variables.contains(targetColumn);
    } else {
      return false;
    }
  }

  /**
   *
   * @param evalNode The expression to be checked
   * @return true if an conjunctive expression, consisting of indexable expressions
   */
  private boolean checkIfConjunctiveButOneVariable(EvalNode evalNode) {
    if (evalNode.getType() == EvalType.AND) {
      BinaryEval orEval = (BinaryEval) evalNode;
      boolean indexable =
          checkIfIndexablePredicate(orEval.getLeftExpr()) &&
              checkIfIndexablePredicate(orEval.getRightExpr());

      boolean sameVariable =
          EvalTreeUtil.findUniqueColumns(orEval.getLeftExpr())
              .equals(EvalTreeUtil.findUniqueColumns(orEval.getRightExpr()));

      return indexable && sameVariable;
    } else {
      return false;
    }
  }

  /**
   * Check if an expression consists of one variable and one constant and
   * the expression is a comparison operator.
   *
   * @param evalNode The expression to be checked
   * @return true if an expression consists of one variable and one constant
   * and the expression is a comparison operator. Other, false.
   */
  private boolean checkIfIndexablePredicate(EvalNode evalNode) {
    return AlgebraicUtil.containSingleVar(evalNode) && isIndexableOperator(evalNode);
  }

  public static boolean isIndexableOperator(EvalNode expr) {
    return expr.getType() == EvalType.EQUAL ||
        expr.getType() == EvalType.LEQ ||
        expr.getType() == EvalType.LTH ||
        expr.getType() == EvalType.GEQ ||
        expr.getType() == EvalType.GTH ||
        expr.getType() == EvalType.BETWEEN;
  }

  public Pair<Datum, Datum> getIndexablePredicateValue(ColumnMapping columnMapping,
                                                       Set<EvalNode> evalNodes) {
    Datum startDatum = null;
    Datum endDatum = null;
    for (EvalNode evalNode: evalNodes) {
      if (evalNode instanceof BinaryEval) {
        BinaryEval binaryEval = (BinaryEval) evalNode;
        EvalNode left = binaryEval.getLeftExpr();
        EvalNode right = binaryEval.getRightExpr();

        Datum constValue = null;
        if (left.getType() == EvalType.CONST) {
          constValue = ((ConstEval) left).getValue();
        } else if (right.getType() == EvalType.CONST) {
          constValue = ((ConstEval) right).getValue();
        }

        if (constValue != null) {
          if (evalNode.getType() == EvalType.EQUAL ||
              evalNode.getType() == EvalType.GEQ ||
              evalNode.getType() == EvalType.GTH) {
            if (startDatum != null) {
              if (constValue.compareTo(startDatum) > 0) {
                startDatum = constValue;
              }
            } else {
              startDatum = constValue;
            }
          }

          if (evalNode.getType() == EvalType.EQUAL ||
              evalNode.getType() == EvalType.LEQ ||
              evalNode.getType() == EvalType.LTH) {
            if (endDatum != null) {
              if (constValue.compareTo(endDatum) < 0) {
                endDatum = constValue;
              }
            } else {
              endDatum = constValue;
            }
          }
        }
      } else if (evalNode instanceof BetweenPredicateEval) {
        BetweenPredicateEval betweenEval = (BetweenPredicateEval) evalNode;
        if (betweenEval.getBegin().getType() == EvalType.CONST && betweenEval.getEnd().getType() == EvalType.CONST) {
          Datum value = ((ConstEval) betweenEval.getBegin()).getValue();
          if (startDatum != null) {
            if (value.compareTo(startDatum) > 0) {
              startDatum = value;
            }
          } else {
            startDatum = value;
          }

          value = ((ConstEval) betweenEval.getEnd()).getValue();
          if (endDatum != null) {
            if (value.compareTo(endDatum) < 0) {
              endDatum = value;
            }
          } else {
            endDatum = value;
          }
        }
      }
    }

    if (endDatum != null && columnMapping != null && columnMapping.getNumRowKeys() > 1) {
      endDatum = new TextDatum(endDatum.asChars() +
          new String(new char[]{columnMapping.getRowKeyDelimiter(), Character.MAX_VALUE}));
    }
    if (startDatum != null || endDatum != null) {
      return new Pair<>(startDatum, endDatum);
    } else {
      return null;
    }
  }

  @Override
  public Path commitTable(OverridableConf queryContext, ExecutionBlockId finalEbId,
                          LogicalPlan plan, Schema schema,
                          TableDesc tableDesc) throws IOException {
    if (tableDesc == null) {
      throw new IOException("TableDesc is null while calling loadIncrementalHFiles: " + finalEbId);
    }

    Path stagingDir = new Path(queryContext.get(QueryVars.STAGING_DIR));
    Path stagingResultDir = new Path(stagingDir, TajoConstants.RESULT_DIR_NAME);

    Configuration hbaseConf = HBaseConfiguration.create(this.hbaseConf);
    hbaseConf.set("hbase.loadincremental.threads.max", "2");

    JobContextImpl jobContext = new JobContextImpl(hbaseConf,
        new JobID(finalEbId.getQueryId().toString(), finalEbId.getId()));

    FileOutputCommitter committer = new FileOutputCommitter(stagingResultDir, jobContext);
    Path jobAttemptPath = committer.getJobAttemptPath(jobContext);
    FileSystem fs = jobAttemptPath.getFileSystem(queryContext.getConf());
    if (!fs.exists(jobAttemptPath) || fs.listStatus(jobAttemptPath) == null) {
      LOG.warn("No query attempt file in " + jobAttemptPath);
      return stagingResultDir;
    }
    committer.commitJob(jobContext);

    // insert into table
    String tableName = tableDesc.getMeta().getOption(HBaseStorageConstants.META_TABLE_KEY);

    HTable htable = new HTable(hbaseConf, tableName);
    try {
      LoadIncrementalHFiles loadIncrementalHFiles = null;
      try {
        loadIncrementalHFiles = new LoadIncrementalHFiles(hbaseConf);
      } catch (Exception e) {
        LOG.error(e.getMessage(), e);
        throw new IOException(e.getMessage(), e);
      }
      loadIncrementalHFiles.doBulkLoad(stagingResultDir, htable);

      return stagingResultDir;
    } finally {
      htable.close();
    }
  }

  @Override
  public TupleRange[] getInsertSortRanges(OverridableConf queryContext, TableDesc tableDesc,
                                          Schema inputSchema, SortSpec[] sortSpecs, TupleRange dataRange)
      throws IOException {
    try {
      int[] sortKeyIndexes = new int[sortSpecs.length];
      for (int i = 0; i < sortSpecs.length; i++) {
        sortKeyIndexes[i] = inputSchema.getColumnId(sortSpecs[i].getSortKey().getQualifiedName());
      }

      ColumnMapping columnMapping = new ColumnMapping(tableDesc.getSchema(), tableDesc.getMeta().getOptions());

      HTable htable = new HTable(hbaseConf, columnMapping.getHbaseTableName());
      try {
        byte[][] endKeys = htable.getEndKeys();
        if (endKeys.length == 1) {
          return new TupleRange[]{dataRange};
        }
        List<TupleRange> tupleRanges = new ArrayList<>(endKeys.length);

        TupleComparator comparator = new BaseTupleComparator(inputSchema, sortSpecs);
        Tuple previousTuple = dataRange.getStart();

        for (byte[] eachEndKey : endKeys) {
          VTuple endTuple = new VTuple(sortSpecs.length);
          byte[][] rowKeyFields;
          if (sortSpecs.length > 1) {
            byte[][] splitValues = BytesUtils.splitPreserveAllTokens(
                eachEndKey, columnMapping.getRowKeyDelimiter(), columnMapping.getNumColumns());
            if (splitValues.length == sortSpecs.length) {
              rowKeyFields = splitValues;
            } else {
              rowKeyFields = new byte[sortSpecs.length][];
              for (int j = 0; j < sortSpecs.length; j++) {
                if (j < splitValues.length) {
                  rowKeyFields[j] = splitValues[j];
                } else {
                  rowKeyFields[j] = null;
                }
              }
            }

          } else {
            rowKeyFields = new byte[1][];
            rowKeyFields[0] = eachEndKey;
          }

          for (int i = 0; i < sortSpecs.length; i++) {
            if (columnMapping.getIsBinaryColumns()[sortKeyIndexes[i]]) {
              endTuple.put(i,
                      HBaseBinarySerializerDeserializer.deserialize(inputSchema.getColumn(sortKeyIndexes[i]),
                              rowKeyFields[i]));
            } else {
              endTuple.put(i,
                      HBaseTextSerializerDeserializer.deserialize(inputSchema.getColumn(sortKeyIndexes[i]),
                              rowKeyFields[i]));
            }
          }
          tupleRanges.add(new TupleRange(sortSpecs, previousTuple, endTuple));
          previousTuple = endTuple;
        }

        // Last region endkey is empty. Tajo ignores empty key, so endkey is replaced with max data value.
        if (comparator.compare(dataRange.getEnd(), tupleRanges.get(tupleRanges.size() - 1).getStart()) >= 0) {
          tupleRanges.get(tupleRanges.size() - 1).setEnd(dataRange.getEnd());
        } else {
          tupleRanges.remove(tupleRanges.size() - 1);
        }
        return tupleRanges.toArray(new TupleRange[]{});
      } finally {
        htable.close();
      }
    } catch (Throwable t) {
      LOG.error(t.getMessage(), t);
      throw new IOException(t.getMessage(), t);
    }
  }

  @Override
  public void rewritePlan(OverridableConf context, LogicalPlan plan) throws TajoException {
    if (REWRITE_RULE.isEligible(new LogicalPlanRewriteRuleContext(context, plan))) {
      REWRITE_RULE.rewrite(new LogicalPlanRewriteRuleContext(context, plan));
    }
  }

  @Override
  public StorageProperty getProperty() {
    return HBASE_STORAGE_PROPERTIES;
  }

  @Override
  public FormatProperty getFormatProperty(TableMeta meta) {
    KeyValueSet tableProperty = meta.getOptions();
    if (tableProperty.isTrue(HBaseStorageConstants.INSERT_PUT_MODE) ||
        tableProperty.isTrue(StorageConstants.INSERT_DIRECTLY)) {
      return PUT_MODE_PROPERTIES;
    } else {
      return HFILE_FORMAT_PROPERTIES;
    }
  }

  public void prepareTable(LogicalNode node) throws TajoException, IOException {
    if (node.getType() == NodeType.CREATE_TABLE) {
      CreateTableNode cNode = (CreateTableNode)node;
      if (!cNode.isExternal()) {
        TableMeta tableMeta = new TableMeta(cNode.getStorageType(), cNode.getOptions());
        createTable(
            ((CreateTableNode) node).getUri(), tableMeta, cNode.getTableSchema(),
            cNode.isExternal(), cNode.isIfNotExists());
      }
    }
  }

  @Override
  public void rollbackTable(LogicalNode node) throws IOException, TajoException {
    if (node.getType() == NodeType.CREATE_TABLE) {
      CreateTableNode cNode = (CreateTableNode)node;
      if (cNode.isExternal()) {
        return;
      }

      HBaseAdmin hAdmin =  new HBaseAdmin(this.hbaseConf);
      TableMeta tableMeta = new TableMeta(cNode.getStorageType(), cNode.getOptions());
      try {
        HTableDescriptor hTableDesc = parseHTableDescriptor(tableMeta, cNode.getTableSchema());
        LOG.info("Delete table cause query failed:" + new String(hTableDesc.getName()));
        hAdmin.disableTable(hTableDesc.getName());
        hAdmin.deleteTable(hTableDesc.getName());
      } finally {
        hAdmin.close();
      }
    }
  }

  @Override
  public URI getStagingUri(OverridableConf context, String queryId, TableMeta meta) throws IOException {
    if (meta.getOptions().isTrue(HBaseStorageConstants.INSERT_PUT_MODE)) {
      throw new IOException("Staging phase is not supported in this storage.");
    } else {
      return TablespaceManager.getDefault().getStagingUri(context, queryId, meta);
    }
  }

  public URI prepareStagingSpace(TajoConf conf, String queryId, OverridableConf context,
                                 TableMeta meta) throws IOException {
    if (!meta.getOptions().isTrue(HBaseStorageConstants.INSERT_PUT_MODE)) {
      return TablespaceManager.getDefault().prepareStagingSpace(conf, queryId, context, meta);
    } else {
      throw new IOException("Staging phase is not supported in this storage.");
    }
  }

  @Override
  public void verifySchemaToWrite(TableDesc tableDesc, Schema outSchema) throws TajoException {
    if (tableDesc != null) {
      Schema tableSchema = tableDesc.getSchema();
      if (tableSchema.size() != outSchema.size()) {
        throw SyntaxErrorUtil.makeSyntaxError("Target columns and projected columns are mismatched to each other");
      }

      for (int i = 0; i < tableSchema.size(); i++) {
        if (!tableSchema.getColumn(i).getDataType().equals(outSchema.getColumn(i).getDataType())) {
          final Column tableColumn = tableSchema.getColumn(i);
          final Column outColumn = outSchema.getColumn(i);
          throw new DataTypeMismatchException(
              tableColumn.getQualifiedName(),
              tableColumn.getDataType().getType().name(),
              outColumn.getQualifiedName(),
              outColumn.getDataType().getType().name());
        }
      }
    }
  }
}