/**
 * 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.hcatalog.api;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.MetaStoreUtils;
import org.apache.hadoop.hive.metastore.TableType;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Order;
import org.apache.hadoop.hive.metastore.api.SerDeInfo;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat;
import org.apache.hadoop.hive.ql.io.RCFileInputFormat;
import org.apache.hadoop.hive.ql.io.RCFileOutputFormat;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.HiveStorageHandler;
import org.apache.hadoop.hive.ql.metadata.HiveUtils;
import org.apache.hadoop.hive.serde2.columnar.ColumnarSerDe;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hcatalog.common.HCatException;
import org.apache.hcatalog.data.schema.HCatFieldSchema;
import org.apache.hcatalog.data.schema.HCatSchemaUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The Class HCatCreateTableDesc for defining attributes for a new table.
 */
@SuppressWarnings("deprecation")
public class HCatCreateTableDesc{

    private static final Logger LOG = LoggerFactory.getLogger(HCatCreateTableDesc.class);

    private String tableName;
    private String dbName;
    private boolean isExternal;
    private String comment;
    private String location;
    private List<HCatFieldSchema> cols;
    private List<HCatFieldSchema> partCols;
    private List<String> bucketCols;
    private int numBuckets;
    private List<Order> sortCols;
    private Map<String, String> tblProps;
    private boolean ifNotExists;
    private String fileFormat;
    private String inputformat;
    private String outputformat;
    private String serde;
    private String storageHandler;

    private HCatCreateTableDesc(String dbName, String tableName, List<HCatFieldSchema> columns){
        this.dbName = dbName;
        this.tableName = tableName;
        this.cols = columns;
    }

    /**
     * Creates a builder for defining attributes.
     *
     * @param dbName the db name
     * @param tableName the table name
     * @param columns the columns
     * @return the builder
     */
    public static Builder create(String dbName, String tableName, List<HCatFieldSchema> columns){
        return new Builder(dbName, tableName, columns);
    }

    Table toHiveTable(HiveConf conf) throws HCatException{

        Table newTable = new Table();
        newTable.setDbName(dbName);
        newTable.setTableName(tableName);
        if (tblProps != null) {
            newTable.setParameters(tblProps);
        }

        if (isExternal) {
            newTable.putToParameters("EXTERNAL", "TRUE");
            newTable.setTableType(TableType.EXTERNAL_TABLE.toString());
        } else {
            newTable.setTableType(TableType.MANAGED_TABLE.toString());
        }

        StorageDescriptor sd = new StorageDescriptor();
        sd.setSerdeInfo(new SerDeInfo());
        if (location != null) {
            sd.setLocation(location);
        }
        if (this.comment != null) {
            newTable.putToParameters("comment", comment);
        }
        if (!StringUtils.isEmpty(fileFormat)) {
            sd.setInputFormat(inputformat);
            sd.setOutputFormat(outputformat);
            if (serde != null) {
                sd.getSerdeInfo().setSerializationLib(serde);
            } else {
                LOG.info("Using LazySimpleSerDe for table " + tableName);
                sd.getSerdeInfo()
                        .setSerializationLib(
                                org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe.class
                                        .getName());
            }
        } else {
            try {
                LOG.info("Creating instance of storage handler to get input/output, serder info.");
                HiveStorageHandler sh = HiveUtils.getStorageHandler(conf,
                        storageHandler);
                sd.setInputFormat(sh.getInputFormatClass().getName());
                sd.setOutputFormat(sh.getOutputFormatClass().getName());
                sd.getSerdeInfo().setSerializationLib(
                        sh.getSerDeClass().getName());
                newTable.putToParameters(
                        org.apache.hadoop.hive.metastore.api.Constants.META_TABLE_STORAGE,
                        storageHandler);
            } catch (HiveException e) {
                throw new HCatException(
                        "Exception while creating instance of storage handler",
                        e);
            }
        }
        newTable.setSd(sd);
        if (this.partCols != null) {
            ArrayList<FieldSchema> hivePtnCols = new ArrayList<FieldSchema>();
            for (HCatFieldSchema fs : this.partCols) {
                hivePtnCols.add(HCatSchemaUtils.getFieldSchema(fs));
            }
            newTable.setPartitionKeys(hivePtnCols);
        }

        if (this.cols != null) {
            ArrayList<FieldSchema> hiveTblCols = new ArrayList<FieldSchema>();
            for (HCatFieldSchema fs : this.cols) {
                hiveTblCols.add(HCatSchemaUtils.getFieldSchema(fs));
            }
            newTable.getSd().setCols(hiveTblCols);
        }

        if (this.bucketCols != null) {
            newTable.getSd().setBucketCols(bucketCols);
            newTable.getSd().setNumBuckets(numBuckets);
        }

        if (this.sortCols != null) {
            newTable.getSd().setSortCols(sortCols);
        }

        newTable.setCreateTime((int) (System.currentTimeMillis() / 1000));
        newTable.setLastAccessTimeIsSet(false);
        return newTable;
    }

      /**
       * Gets the if not exists.
       *
       * @return the if not exists
       */
      public boolean getIfNotExists() {
          return this.ifNotExists;
      }

     /**
      * Gets the table name.
      *
      * @return the table name
      */
     public String getTableName() {
          return this.tableName;
      }

      /**
       * Gets the cols.
       *
       * @return the cols
       */
      public List<HCatFieldSchema> getCols() {
         return this.cols;
      }

      /**
       * Gets the partition cols.
       *
       * @return the partition cols
       */
      public List<HCatFieldSchema> getPartitionCols() {
          return this.partCols;
      }

      /**
       * Gets the bucket cols.
       *
       * @return the bucket cols
       */
      public List<String> getBucketCols() {
          return this.bucketCols;
      }

      public int getNumBuckets() {
          return this.numBuckets;
      }

      /**
       * Gets the comments.
       *
       * @return the comments
       */
      public String getComments() {
          return this.comment;
      }

      /**
       * Gets the storage handler.
       *
       * @return the storage handler
       */
      public String getStorageHandler() {
          return this.storageHandler;
      }

      /**
       * Gets the location.
       *
       * @return the location
       */
      public String getLocation() {
          return this.location;
      }

      /**
       * Gets the external.
       *
       * @return the external
       */
      public boolean getExternal() {
          return this.isExternal;
      }

      /**
       * Gets the sort cols.
       *
       * @return the sort cols
       */
      public List<Order> getSortCols() {
          return this.sortCols;
      }

      /**
       * Gets the tbl props.
       *
       * @return the tbl props
       */
      public Map<String, String> getTblProps() {
          return this.tblProps;
      }

      /**
       * Gets the file format.
       *
       * @return the file format
       */
      public String getFileFormat(){
          return this.fileFormat;
      }

      /**
       * Gets the database name.
       *
       * @return the database name
       */
      public String getDatabaseName() {
          return this.dbName;
      }

      @Override
    public String toString() {
        return "HCatCreateTableDesc ["
                + (tableName != null ? "tableName=" + tableName + ", " : "tableName=null")
                + (dbName != null ? "dbName=" + dbName + ", " : "dbName=null")
                + "isExternal="
                + isExternal
                + ", "
                + (comment != null ? "comment=" + comment + ", " : "comment=null")
                + (location != null ? "location=" + location + ", " : "location=null")
                + (cols != null ? "cols=" + cols + ", " : "cols=null")
                + (partCols != null ? "partCols=" + partCols + ", " : "partCols=null")
                + (bucketCols != null ? "bucketCols=" + bucketCols + ", " : "bucketCols=null")
                + "numBuckets="
                + numBuckets
                + ", "
                + (sortCols != null ? "sortCols=" + sortCols + ", " : "sortCols=null")
                + (tblProps != null ? "tblProps=" + tblProps + ", " : "tblProps=null")
                + "ifNotExists="
                + ifNotExists
                + ", "
                + (fileFormat != null ? "fileFormat=" + fileFormat + ", " : "fileFormat=null")
                + (inputformat != null ? "inputformat=" + inputformat + ", "
                        : "inputformat=null")
                + (outputformat != null ? "outputformat=" + outputformat + ", "
                        : "outputformat=null")
                + (serde != null ? "serde=" + serde + ", " : "serde=null")
                + (storageHandler != null ? "storageHandler=" + storageHandler
                        : "storageHandler=null") + "]";
    }

    public static class Builder{

          private String tableName;
          private boolean isExternal;
          private List<HCatFieldSchema> cols;
          private List<HCatFieldSchema> partCols;
          private List<String> bucketCols;
          private List<Order> sortCols;
          private int numBuckets;
          private String comment;
          private String fileFormat;
          private String location;
          private String storageHandler;
          private Map<String, String> tblProps;
          private boolean ifNotExists;
          private String dbName;


          private Builder(String dbName, String tableName, List<HCatFieldSchema> columns) {
              this.dbName = dbName;
              this.tableName = tableName;
              this.cols = columns;
          }


          /**
           * If not exists.
           *
           * @param ifNotExists If set to true, hive will not throw exception, if a
           * table with the same name already exists.
           * @return the builder
           */
          public Builder ifNotExists(boolean ifNotExists) {
            this.ifNotExists = ifNotExists;
            return this;
          }


          /**
           * Partition cols.
           *
           * @param partCols the partition cols
           * @return the builder
           */
          public Builder partCols(ArrayList<HCatFieldSchema> partCols) {
              this.partCols = partCols;
              return this;
          }


          /**
           * Bucket cols.
           *
           * @param bucketCols the bucket cols
           * @return the builder
           */
          public Builder bucketCols(ArrayList<String> bucketCols, int buckets) {
            this.bucketCols = bucketCols;
            this.numBuckets = buckets;
            return this;
          }

          /**
           * Storage handler.
           *
           * @param storageHandler the storage handler
           * @return the builder
           */
          public Builder storageHandler(String storageHandler) {
            this.storageHandler = storageHandler;
            return this;
          }

          /**
           * Location.
           *
           * @param location the location
           * @return the builder
           */
          public Builder location(String location) {
            this.location = location;
            return this;
          }

          /**
           * Comments.
           *
           * @param comment the comment
           * @return the builder
           */
          public Builder comments(String comment) {
            this.comment = comment;
            return this;
          }

          /**
           * Checks if is table external.
           *
           * @param isExternal the is external
           * @return the builder
           */
          public Builder isTableExternal(boolean isExternal) {
            this.isExternal = isExternal;
            return this;
          }

          /**
           * Sort cols.
           *
           * @param sortCols the sort cols
           * @return the builder
           */
          public Builder sortCols(ArrayList<Order> sortCols) {
            this.sortCols = sortCols;
            return this;
          }

          /**
           * Tbl props.
           *
           * @param tblProps the tbl props
           * @return the builder
           */
          public Builder tblProps(Map<String, String> tblProps) {
            this.tblProps = tblProps;
            return this;
          }

          /**
           * File format.
           *
           * @param format the format
           * @return the builder
           */
          public Builder fileFormat(String format){
              this.fileFormat = format;
              return this;
          }

          /**
           * Builds the HCatCreateTableDesc.
           *
           * @return HCatCreateTableDesc
           * @throws HCatException
           */
          public HCatCreateTableDesc build() throws HCatException {
              if(this.dbName == null){
                LOG.info("Database name found null. Setting db to :"
                        + MetaStoreUtils.DEFAULT_DATABASE_NAME);
                this.dbName = MetaStoreUtils.DEFAULT_DATABASE_NAME;
              }
              HCatCreateTableDesc desc = new HCatCreateTableDesc(this.dbName,
                      this.tableName, this.cols);
              desc.ifNotExists = this.ifNotExists;
              desc.isExternal = this.isExternal;
              desc.comment = this.comment;
              desc.partCols = this.partCols;
              desc.bucketCols = this.bucketCols;
              desc.numBuckets = this.numBuckets;
              desc.location = this.location;
              desc.tblProps = this.tblProps;
              desc.sortCols = this.sortCols;
              desc.serde = null;
              if (!StringUtils.isEmpty(fileFormat)) {
                  desc.fileFormat = fileFormat;
                  if ("SequenceFile".equalsIgnoreCase(fileFormat)) {
                      desc.inputformat = SequenceFileInputFormat.class.getName();
                      desc.outputformat = SequenceFileOutputFormat.class
                              .getName();
                  } else if ("RCFile".equalsIgnoreCase(fileFormat)) {
                      desc.inputformat = RCFileInputFormat.class.getName();
                      desc.outputformat = RCFileOutputFormat.class.getName();
                      desc.serde = ColumnarSerDe.class.getName();
                  }
                  desc.storageHandler = StringUtils.EMPTY;
              } else if (!StringUtils.isEmpty(storageHandler)) {
                  desc.storageHandler = storageHandler;
              } else {
                  desc.fileFormat = "TextFile";
                  LOG.info("Using text file format for the table.");
                  desc.inputformat = TextInputFormat.class.getName();
                  LOG.info("Table input format:" + desc.inputformat);
                  desc.outputformat = IgnoreKeyTextOutputFormat.class
                          .getName();
                  LOG.info("Table output format:" + desc.outputformat);
              }
              return desc;
          }
      }
}