/**
 * 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.sqoop.mapreduce.hcat;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

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.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.shims.HadoopShims;
import org.apache.hadoop.hive.shims.HadoopShims.HCatHadoopShims;
import org.apache.hadoop.hive.shims.ShimLoader;
import org.apache.hadoop.io.ArrayWritable;
import org.apache.hadoop.io.DefaultStringifier;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.util.Shell;
import org.apache.hadoop.util.StringUtils;
import org.apache.hive.hcatalog.common.HCatConstants;
import org.apache.hive.hcatalog.common.HCatUtil;
import org.apache.hive.hcatalog.data.DefaultHCatRecord;
import org.apache.hive.hcatalog.data.schema.HCatFieldSchema;
import org.apache.hive.hcatalog.data.schema.HCatSchema;
import org.apache.hive.hcatalog.mapreduce.HCatInputFormat;
import org.apache.hive.hcatalog.mapreduce.HCatOutputFormat;
import org.apache.hive.hcatalog.mapreduce.OutputJobInfo;
import org.apache.sqoop.config.ConfigurationConstants;
import org.apache.sqoop.config.ConfigurationHelper;
import org.apache.sqoop.hive.HiveTypes;
import org.apache.sqoop.manager.ConnManager;
import org.apache.sqoop.util.Executor;
import org.apache.sqoop.util.LoggingAsyncSink;
import org.apache.sqoop.util.SubprocessSecurityManager;

import com.cloudera.sqoop.SqoopOptions;
import com.cloudera.sqoop.lib.DelimiterSet;
import com.cloudera.sqoop.util.ExitSecurityException;

/**
 * Utility methods for the HCatalog support for Sqoop.
 */
public final class SqoopHCatUtilities {
  public static final String DEFHCATDB = "default";
  public static final String HIVESITEXMLPATH = "/conf/hive-site.xml";
  public static final String HCATSHAREDIR = "share/hcatalog";
  public static final String DEFLIBDIR = "lib";
  public static final String TEXT_FORMAT_IF_CLASS =
    "org.apache.hadoop.mapred.TextInputFormat";
  public static final String TEXT_FORMAT_OF_CLASS =
    "org.apache.hadoop.mapred.TextOutputFormat";
  public static final String TEXT_FORMAT_SERDE_CLASS =
    "org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe";
  public static final String HCAT_DB_OUTPUT_COLTYPES_JAVA =
    "sqoop.hcat.db.output.coltypes.java";
  public static final String HCAT_DB_OUTPUT_COLTYPES_SQL =
    "sqoop.hcat.db.output.coltypes.sql";
  public static final String HCAT_CLI_MAIN_CLASS =
    "org.apache.hive.hcatalog.cli.HCatCli";
  public static final String HCAT_DEF_STORAGE_STANZA = "stored as rcfile";
  public static final String HIVE_DELIMITERS_TO_REPLACE_PROP =
    "sqoop.hive.delims.to.replace";
  public static final String HIVE_DELIMITERS_REPLACEMENT_PROP =
    "sqoop.hive.delims.replacement";
  public static final String HIVE_DELIMITERS_REPLACEMENT_ENABLED_PROP =
    "sqoop.hive.delims.replacement.enabled";
  public static final String HCAT_STATIC_PARTITION_KEY_PROP =
    "sqoop.hcat.partition.key";
  public static final String HCAT_FIELD_POSITIONS_PROP =
    "sqoop.hcat.field.positions";
  public static final String DEBUG_HCAT_IMPORT_MAPPER_PROP =
    "sqoop.hcat.debug.import.mapper";
  public static final String DEBUG_HCAT_EXPORT_MAPPER_PROP =
    "sqoop.hcat.debug.export.mapper";
  private static final String HCATCMD = Shell.WINDOWS ? "hcat.py" : "hcat";
  private static final int MAX_HIVE_CHAR_PREC = 65535;
  private static final int MAX_HIVE_DECIMAL_PREC = 65;
  private static final int MAX_HIVE_DECIMAL_SCALE = 30;
  private SqoopOptions options;
  private ConnManager connManager;
  private String hCatTableName;
  private String hCatDatabaseName;
  private Configuration configuration;
  private Job hCatJob;
  private HCatSchema hCatOutputSchema;
  private HCatSchema hCatPartitionSchema;
  private HCatSchema projectedSchema;
  private boolean configured;

  private String hCatQualifiedTableName;
  private List<String> hCatStaticPartitionKeys;
  private List<String> hCatStaticPartitionValues;
  private List<String> hCatDynamicPartitionKeys;
  // DB stuff
  private String[] dbColumnNames;
  private String dbTableName;
  private LCKeyMap<List<Integer>> dbColumnInfo;

  private int[] hCatFieldPositions; // For each DB column, HCat position

  private HCatSchema hCatFullTableSchema;
  private List<String> hCatFullTableSchemaFieldNames;
  private LCKeyMap<String> userHiveMapping;

  // For testing support
  private static Class<? extends InputFormat> inputFormatClass =
    SqoopHCatExportFormat.class;

  private static Class<? extends OutputFormat> outputFormatClass =
    HCatOutputFormat.class;

  private static Class<? extends Mapper> exportMapperClass =
    SqoopHCatExportMapper.class;

  private static Class<? extends Mapper> importMapperClass =
    SqoopHCatImportMapper.class;

  private static Class<? extends Writable> importValueClass =
    DefaultHCatRecord.class;

  private static boolean testMode = false;

  public static boolean isHCatView(final SqoopOptions opts)  {

    String hCatDatabaseName = getHCatDbNameFromOptions(opts);

    String hCatTableName = getHCatTableNameFromOptions(opts);

    Configuration conf = opts.getConf();
    HiveConf hiveConf;
    try {
      if (conf != null) {
        hiveConf = HCatUtil.getHiveConf(conf);
      } else {
        hiveConf = new HiveConf(HCatInputFormat.class);
      }
      HiveMetaStoreClient client = HCatUtil.getHiveClient(hiveConf);
      Table table = HCatUtil.getTable(client, hCatDatabaseName, hCatTableName);
      return table.isView();
    }
    catch(Exception e){
      LOG.warn("We were not able to determine if "+hCatDatabaseName+ ":"+hCatTableName+ "is view or table.");
      LOG.info("isHCatView threw an exception:", e);
      return false;
    }
  }

  private static String getHCatTableNameFromOptions(SqoopOptions opts) {
    String optHCTabName = opts.getHCatTableName();
    return optHCTabName.toLowerCase();
  }

  private static String getHCatDbNameFromOptions(SqoopOptions opts) {
    String hCatDatabaseName = opts.getHCatDatabaseName() != null ? opts
        .getHCatDatabaseName() : DEFHCATDB;
    hCatDatabaseName = hCatDatabaseName.toLowerCase();
    return hCatDatabaseName;
  }

  static class IntArrayWritable extends ArrayWritable {
    public IntArrayWritable() {
      super(IntWritable.class);
    }
  }

  /**
   * A Map using String as key type that ignores case of its key and stores the
   * key in lower case.
   */
  private static class LCKeyMap<V> extends HashMap<String, V> {

    private static final long serialVersionUID = -6751510232323094216L;

    @Override
    public V put(String key, V value) {
      return super.put(key.toLowerCase(), value);
    }

    @Override
    public V get(Object key) {
      return super.get(((String) key).toLowerCase());
    }

    @Override
    public void putAll(Map<? extends String, ? extends V> map) {
      for(Map.Entry<? extends String, ? extends V> entry: map.entrySet()) {
        put(entry.getKey(), entry.getValue());
      }
    }
  }

  /**
   * A Map using String as key type that ignores case of its key and stores the
   * key in upper case.
   */
  public class UCKeyMap<V> extends HashMap<String, V> {

    private static final long serialVersionUID = -6751510232323094216L;

    @Override
    public V put(String key, V value) {
      return super.put(key.toUpperCase(), value);
    }

    @Override
    public V get(Object key) {
      return super.get(((String) key).toUpperCase());
    }

    @Override
    public void putAll(Map<? extends String, ? extends V> map) {
      for(Map.Entry<? extends String, ? extends V> entry: map.entrySet()) {
        put(entry.getKey(), entry.getValue());
      }
    }
  }

  /**
   * A class to hold the instance. For guaranteeing singleton creation using JMM
   * semantics.
   */
  public static final class Holder {
    @SuppressWarnings("synthetic-access")
    public static final SqoopHCatUtilities INSTANCE = new SqoopHCatUtilities();

    private Holder() {
    }
  }

  public static SqoopHCatUtilities instance() {
    return Holder.INSTANCE;
  }

  private SqoopHCatUtilities() {
    configured = false;
  }

  public static final Log LOG = LogFactory.getLog(SqoopHCatUtilities.class
    .getName());

  public boolean isConfigured() {
    return configured;
  }

  private void checkHomeDirs(SqoopOptions opts) {

    LOG.info("Configuring HCatalog specific details for job");

    String home = opts.getHiveHome();

    if (home == null || home.length() == 0) {
      LOG.warn("Hive home is not set. job may fail if needed jar files "
        + "are not found correctly.  Please set HIVE_HOME in"
        + " sqoop-env.sh or provide --hive-home option.  HIVE_HOME defaulted "
        + " to " + SqoopOptions.getHiveHomeDefault());
    }

    home = opts.getHCatHome();
    if (home == null || home.length() == 0) {
      LOG.warn("HCatalog home is not set. job may fail if needed jar "
        + "files are not found correctly.  Please set HCAT_HOME in"
        + " sqoop-env.sh or provide --hcatalog-home option.  "
        + "HCAT_HOME defaulted to " + SqoopOptions.getHCatHomeDefault());
    }
  }

  public void configureHCat(final SqoopOptions opts, final Job job,
    final ConnManager connMgr, final String dbTable,
    final Configuration config) throws IOException {
    if (configured) {
      LOG.info("Ignoring configuration request for HCatalog info");
      return;
    }
    options = opts;
    checkHomeDirs(opts);
    connManager = connMgr;
    dbTableName = dbTable;
    configuration = config;
    hCatJob = job;
    hCatDatabaseName = getHCatDbNameFromOptions(opts);

    String optHCTabName = options.getHCatTableName();
    hCatTableName = getHCatTableNameFromOptions(opts);

    if (!hCatTableName.equals(optHCTabName)) {
      LOG.warn("Provided HCatalog table name " + optHCTabName
        + " will be mapped to  " + hCatTableName);
    }

    StringBuilder sb = new StringBuilder();
    sb.append(hCatDatabaseName);
    sb.append('.').append(hCatTableName);
    hCatQualifiedTableName = sb.toString();

    String principalID = System
      .getProperty(HCatConstants.HCAT_METASTORE_PRINCIPAL);
    if (principalID != null) {
      configuration.set(HCatConstants.HCAT_METASTORE_PRINCIPAL, principalID);
    }
    hCatStaticPartitionKeys = new ArrayList<String>();
    hCatStaticPartitionValues = new ArrayList<String>();
    String partKeysString = options.getHCatalogPartitionKeys();
    String partKeysVals = options.getHCatalogPartitionValues();
    // Already validated
    if (partKeysString != null) {
      String[] keys = partKeysString.split(",");
      for (int i = 0; i < keys.length; ++i) {
        String k = keys[i].trim();
        hCatStaticPartitionKeys.add(k);
      }
      String[] vals = partKeysVals.split(",");
      for (int i = 0; i < vals.length; ++i) {
        String v = vals[i].trim();
        hCatStaticPartitionValues.add(v);
      }
    } else {
      partKeysString = options.getHivePartitionKey();
      if (partKeysString != null) {
        hCatStaticPartitionKeys.add(partKeysString);
      }
      partKeysVals = options.getHivePartitionValue();
      hCatStaticPartitionValues.add(partKeysVals);

    }
    Properties userMapping = options.getMapColumnHive();
    userHiveMapping = new LCKeyMap<String>();
    for (Object o : userMapping.keySet()) {
      String v = (String) userMapping.get(o);
      userHiveMapping.put((String) o, v);
    }
    // Get the partition key filter if needed
    Map<String, String> filterMap = getHCatSPFilterMap();
    String filterStr = getHCatSPFilterStr();
    initDBColumnInfo();
    if (options.doCreateHCatalogTable()) {
      LOG.info("Creating HCatalog table " + hCatQualifiedTableName
        + " for import");
      createHCatTable(false);
    } else if (options.doDropAndCreateHCatalogTable()) {
      LOG.info("Dropping and Creating HCatalog table "
        + hCatQualifiedTableName + " for import");
      createHCatTable(true);
    }
    // For serializing the schema to conf
    HCatInputFormat hif = HCatInputFormat.setInput(hCatJob, hCatDatabaseName,
      hCatTableName);
    // For serializing the schema to conf
    if (filterStr != null) {
      LOG.info("Setting hCatInputFormat filter to " + filterStr);
      hif.setFilter(filterStr);
    }

    hCatFullTableSchema = HCatInputFormat.getTableSchema(configuration);
    hCatFullTableSchemaFieldNames = hCatFullTableSchema.getFieldNames();

    LOG.info("HCatalog full table schema fields = "
      + Arrays.toString(hCatFullTableSchema.getFieldNames().toArray()));

    if (filterMap != null) {
      LOG.info("Setting hCatOutputFormat filter to " + filterStr);
    }

    HCatOutputFormat.setOutput(hCatJob,
      OutputJobInfo.create(hCatDatabaseName, hCatTableName, filterMap));
    hCatOutputSchema = HCatOutputFormat.getTableSchema(configuration);
    List<HCatFieldSchema> hCatPartitionSchemaFields =
      new ArrayList<HCatFieldSchema>();
    int totalFieldsCount = hCatFullTableSchema.size();
    int dataFieldsCount = hCatOutputSchema.size();
    if (totalFieldsCount > dataFieldsCount) {
      for (int i = dataFieldsCount; i < totalFieldsCount; ++i) {
        hCatPartitionSchemaFields.add(hCatFullTableSchema.get(i));
      }
    }

    hCatPartitionSchema = new HCatSchema(hCatPartitionSchemaFields);
    for (HCatFieldSchema hfs : hCatPartitionSchemaFields) {
      if (hfs.getType() != HCatFieldSchema.Type.STRING) {
        throw new IOException("The table provided "
          + getQualifiedHCatTableName()
          + " uses unsupported  partitioning key type  for column "
          + hfs.getName() + " : " + hfs.getTypeString() + ".  Only string "
          + "fields are allowed in partition columns in HCatalog");
      }

    }
    LOG.info("HCatalog table partitioning key fields = "
      + Arrays.toString(hCatPartitionSchema.getFieldNames().toArray()));

    List<HCatFieldSchema> outputFieldList = new ArrayList<HCatFieldSchema>();
    for (String col : dbColumnNames) {
      try {
        HCatFieldSchema hfs = hCatFullTableSchema.get(col);
        if (hfs == null) {
          throw new IOException("Database column " + col + " not found in "
              + " hcatalog table.");
        }
      } catch (Exception e) {
        throw new IOException("Caught Exception checking database column " + col + " in "
            + " hcatalog table.", e);
      }
      boolean skip=false;
      if (hCatStaticPartitionKeys != null) {
        for (String key : hCatStaticPartitionKeys) {
          if (col.equals(key)) {
            skip=true;
            break;
          }
        }
      }
      if (skip) {
        continue;
      }
      outputFieldList.add(hCatFullTableSchema.get(col));
    }

    projectedSchema = new HCatSchema(outputFieldList);

    LOG.info("HCatalog projected schema fields = "
      + Arrays.toString(projectedSchema.getFieldNames().toArray()));

    validateStaticPartitionKey();
    validateHCatTableFieldTypes();

    HCatOutputFormat.setSchema(configuration, hCatFullTableSchema);

    addJars(hCatJob, options);
    config.setBoolean(DEBUG_HCAT_IMPORT_MAPPER_PROP,
      Boolean.getBoolean(DEBUG_HCAT_IMPORT_MAPPER_PROP));
    config.setBoolean(DEBUG_HCAT_EXPORT_MAPPER_PROP,
      Boolean.getBoolean(DEBUG_HCAT_EXPORT_MAPPER_PROP));
    configured = true;
  }

  public void validateDynamicPartitionKeysMapping() throws IOException {
    // Now validate all partition columns are in the database column list
    StringBuilder missingKeys = new StringBuilder();

    for (String s : hCatDynamicPartitionKeys) {
      boolean found = false;
      for (String c : dbColumnNames) {
        if (s.equals(c)) {
          found = true;
          break;
        }
      }
      if (!found) {
        missingKeys.append(',').append(s);
      }
    }
    if (missingKeys.length() > 0) {
      throw new IOException("Dynamic partition keys are not "
        + "present in the database columns.   Missing keys = "
        + missingKeys.substring(1));
    }
  }

  public void validateHCatTableFieldTypes() throws IOException {
    StringBuilder sb = new StringBuilder();
    boolean hasComplexFields = false;
    for (HCatFieldSchema hfs : projectedSchema.getFields()) {
      if (hfs.isComplex()) {
        sb.append('.').append(hfs.getName());
        hasComplexFields = true;
      }
    }

    if (hasComplexFields) {
      String unsupportedFields = sb.substring(1);
      throw new IOException("The HCatalog table provided "
        + getQualifiedHCatTableName() + " has complex field types ("
        + unsupportedFields + ").  They are currently not supported");
    }

  }

  /**
   * Get the column names to import.
   */
  private void initDBColumnInfo() throws IOException {
    String[] colNames = options.getColumns();
    if (null == colNames) {
      if (null != dbTableName) {
        colNames = connManager.getColumnNames(dbTableName);
      } else if (options.getCall() != null) {
        // Read procedure arguments from metadata
        colNames = connManager.getColumnNamesForProcedure(this.options
          .getCall());
      } else {
        colNames = connManager.getColumnNamesForQuery(options.getSqlQuery());
      }
    } else {
      String[] fixedColNames = new String[colNames.length];
      for (int i = 0; i < colNames.length; ++i) {
        String userColName = colNames[i];

        // Remove surrounding quotes if any
        int len = userColName.length();
        if (len > 2 && userColName.charAt(0) == '"' &&  userColName.charAt(len -1) == '"') {
          userColName = userColName.substring(1, len -1);
        }

        fixedColNames[i] = userColName;
      }
      colNames = fixedColNames;
    }

    dbColumnNames = new String[colNames.length];

    for (int i = 0; i < colNames.length; ++i) {
      dbColumnNames[i] = colNames[i].toLowerCase();
    }

    LCKeyMap<List<Integer>> colInfo = new LCKeyMap<List<Integer>>();
    if (dbTableName != null) {
        colInfo.putAll(connManager.getColumnInfo(dbTableName));
    } else if (options.getCall() != null) {
        // Read procedure arguments from metadata
        colInfo.putAll(connManager.getColumnInfoForProcedure(this.options
          .getCall()));
    } else {
        colInfo.putAll(connManager.getColumnInfoForQuery(options
          .getSqlQuery()));
    }

    if (options.getColumns() == null) {
      dbColumnInfo = colInfo;
    } else {
      dbColumnInfo = new LCKeyMap<List<Integer>>();
      // prune column types based on projection
      for (String col : dbColumnNames) {
        List<Integer> info = colInfo.get(col);
        if (info == null) {
          throw new IOException("Projected column " + col
            + " not in list of columns from database");
        }
        dbColumnInfo.put(col, info);
      }
    }
    LOG.info("Database column names projected : "
      + Arrays.toString(dbColumnNames));
    StringBuilder sb = new StringBuilder(1024);
    for (String col : dbColumnInfo.keySet()) {
      sb.append('\t').append(col).append(" : [");
      List<Integer> info = dbColumnInfo.get(col);
      sb.append("Type : ").append(info.get(0)).append(',');
      sb.append("Precision : ").append(info.get(1)).append(',');
      sb.append("Scale : ").append(info.get(2)).append(']');
      sb.append('\n');
    }
    LOG.info("Database column name - info map :\n" + sb.toString());
  }

  public static StringBuilder escHCatObj(String objectName) {
    StringBuilder sb = new StringBuilder(64);
    sb.append('`').append(objectName).append('`');
    return sb;
  }

  private void createHCatTable(boolean dropIfExists) throws IOException {
    StringBuilder sb = new StringBuilder();
    if (dropIfExists) {
      sb.append("drop table ").
              append(escHCatObj(hCatDatabaseName)).append('.').
              append(escHCatObj(hCatTableName)).append(";\n");
    }
    sb.append("create table ").
      append(escHCatObj(hCatDatabaseName)).append('.');
    sb.append(escHCatObj(hCatTableName)).append(" (\n\t");
    boolean first = true;
    for (String col : dbColumnNames) {
      String type = userHiveMapping.get(col);
      int prec = -1;
      int scale = -1;
      if (type == null) {
        type = connManager.toHCatType(dbColumnInfo.get(col).get(0));
      }
      if (type.equals("char") || type.equals("varchar")) {
        prec = dbColumnInfo.get(col).get(1);
        if (prec > MAX_HIVE_CHAR_PREC) {
          LOG.warn("Truncating precison of column " + col + "  from " + prec
            + " to " + MAX_HIVE_CHAR_PREC);
          prec = MAX_HIVE_CHAR_PREC;
        }
      } else if (type.equals("decimal")) {
          prec = dbColumnInfo.get(col).get(1);
          if (prec > MAX_HIVE_DECIMAL_PREC) {
            LOG.warn("Truncating precison of column " + col + "  from " + prec
              + " to " + MAX_HIVE_DECIMAL_PREC);
            prec = MAX_HIVE_DECIMAL_PREC;
          }

          scale = dbColumnInfo.get(col).get(2);
          if (scale > MAX_HIVE_DECIMAL_SCALE) {
            LOG.warn("Truncating precison of column " + col + "  from " + scale
              + " to " + MAX_HIVE_DECIMAL_SCALE);
            scale = MAX_HIVE_DECIMAL_SCALE;
          }
      }
      boolean skip=false;
      if (hCatStaticPartitionKeys != null) {
        for (String key : hCatStaticPartitionKeys) {
          if (col.equals(key)) {
            skip=true;
            break;
          }
        }
      }
      if (skip) {
        continue;
      }
      if (first) {
        first = false;
      } else {
        sb.append(",\n\t");
      }
      sb.append(escHCatObj(col)).append(' ').append(type);
      if (prec > 0) {
        sb.append('(').append(prec);
        if (scale > 0) {
          sb.append(',').append(scale);
        }
        sb.append(')');
      }
    }
    sb.append(")\n");
    first = true;
    if (hCatStaticPartitionKeys != null && hCatStaticPartitionKeys.size() > 0) {
      sb.append("partitioned by (\n\t");
      for (String key : hCatStaticPartitionKeys) {
        if (first) {
          first = false;
        } else {
          sb.append(", ");
        }
        sb.append(escHCatObj(key)).append(" string");
      }
      sb.append(")\n");
    }
    String storageStanza = options.getHCatStorageStanza();
    if (storageStanza == null) {
      sb.append(HCAT_DEF_STORAGE_STANZA);
    } else {
      sb.append(storageStanza);
    }
    String createStatement = sb.toString();
    LOG.info("HCatalog Create table statement: \n\n" + createStatement);
    // Always launch as an external program so that logging is not messed
    // up by the use of inline hive CLI except in tests
    // We prefer external HCAT client.
    launchHCatCli(createStatement);
  }


  private void validateFieldAndColumnMappings() throws IOException {
    // Check that all explicitly mapped columns are present
    for (Object column : userHiveMapping.keySet()) {
      boolean found = false;
      for (String c : dbColumnNames) {
        if (c.equalsIgnoreCase((String) column)) {
          found = true;
          break;
        }
      }

      if (!found) {
        throw new IllegalArgumentException("Column " + column
          + " not found while mapping database columns to hcatalog columns");
      }
    }

    hCatFieldPositions = new int[dbColumnNames.length];

    Arrays.fill(hCatFieldPositions, -1);

    for (int indx = 0; indx < dbColumnNames.length; ++indx) {
      boolean userMapped = false;
      String col = dbColumnNames[indx];
      List<Integer> colInfo = dbColumnInfo.get(col);
      String hCatColType = userHiveMapping.get(col);
      if (hCatColType == null) {
        LOG.debug("No user defined type mapping for HCatalog field " + col);
        hCatColType = connManager.toHCatType(colInfo.get(0));
      } else {
        LOG.debug("Found type mapping for HCatalog filed " + col);
        userMapped = true;
      }
      if (null == hCatColType) {
        throw new IOException("HCat does not support the SQL type for column "
          + col);
      }

      boolean found = false;
      for (String tf : hCatFullTableSchemaFieldNames) {
        if (tf.equals(col)) {
          found = true;
          break;
        }
      }

      if (!found) {
        throw new IOException("Database column " + col + " not found in "
          + "hcatalog table schema or partition schema");
      }
      if (!userMapped) {
        HCatFieldSchema hCatFS = hCatFullTableSchema.get(col);
        if (!hCatFS.getTypeString().equals(hCatColType)) {
          LOG.warn("The HCatalog field " + col + " has type "
            + hCatFS.getTypeString() + ".  Expected = " + hCatColType
            + " based on database column type : "
            + sqlTypeString(colInfo.get(0)));
          LOG.warn("The Sqoop job can fail if types are not "
            + " assignment compatible");
        }
      }

      if (HiveTypes.isHiveTypeImprovised(colInfo.get(0))) {
        LOG.warn("Column " + col + " had to be cast to a less precise type "
          + hCatColType + " in hcatalog");
      }
      hCatFieldPositions[indx] = hCatFullTableSchemaFieldNames.indexOf(col);
      if (hCatFieldPositions[indx] < 0) {
        throw new IOException("The HCatalog field " + col
          + " could not be found");
      }
    }

    IntWritable[] positions = new IntWritable[hCatFieldPositions.length];
    for (int i = 0; i < hCatFieldPositions.length; ++i) {
      positions[i] = new IntWritable(hCatFieldPositions[i]);
    }

    DefaultStringifier.storeArray(configuration, positions,
      HCAT_FIELD_POSITIONS_PROP);
  }

  private String getHCatSPFilterStr() {
    if (hCatStaticPartitionKeys != null && hCatStaticPartitionKeys.size() > 0) {
      StringBuilder filter = new StringBuilder();
      boolean first = true;
      for (int i = 0; i < hCatStaticPartitionKeys.size(); ++i) {
        String key = hCatStaticPartitionKeys.get(i);
        String val = hCatStaticPartitionValues.get(i);
        if (first) {
          first = false;
        } else {
          filter.append(" AND ");
        }
        filter.append(key).append('=').append('\'')
          .append(val).append('\'');
      }
      return filter.toString();
    }
    return null;
  }

  private Map<String, String> getHCatSPFilterMap() {
    if (hCatStaticPartitionKeys != null && hCatStaticPartitionKeys.size() > 0) {
      Map<String, String> filter = new HashMap<String, String>();
      for (int i = 0; i < hCatStaticPartitionKeys.size(); ++i) {
        String key = hCatStaticPartitionKeys.get(i);
        String val = hCatStaticPartitionValues.get(i);
        filter.put(key, val);
      }
      return filter;
    }
    return null;
  }

  private void validateStaticPartitionKey() throws IOException {
    // check the static partition key from command line
    List<HCatFieldSchema> partFields = hCatPartitionSchema.getFields();

    if (hCatStaticPartitionKeys != null
        && hCatStaticPartitionKeys.size() > 0) {
      for (String key : hCatStaticPartitionKeys) {
        boolean found = false;
        for (HCatFieldSchema hfs : partFields) {
          if (hfs.getName().equals(key)) {
            found = true;
            break;
          }
        }
        if (!found) {
          throw new IOException("The provided hive partition key "
          + key + " is not part of the partition "
          + " keys for table " + getQualifiedHCatTableName());
        }
      }
    }
    hCatDynamicPartitionKeys = new ArrayList<String>();
    hCatDynamicPartitionKeys.addAll(hCatPartitionSchema.getFieldNames());
    if (hCatStaticPartitionKeys != null
        && hCatStaticPartitionKeys.size() > 0) {
      for (String key : hCatStaticPartitionKeys) {
        hCatDynamicPartitionKeys.remove(key);
      }
    }
    configuration.setStrings(HCAT_STATIC_PARTITION_KEY_PROP,
       hCatStaticPartitionKeys.toArray(
         new String[hCatStaticPartitionKeys.size()]));
  }

  public static void configureImportOutputFormat(SqoopOptions opts, Job job,
    ConnManager connMgr, String dbTable, Configuration config)
    throws IOException {

    LOG.info("Configuring HCatalog for import job");
    SqoopHCatUtilities.instance().configureHCat(opts, job, connMgr, dbTable,
      job.getConfiguration());
    LOG.info("Validating dynamic partition keys");
    SqoopHCatUtilities.instance().validateFieldAndColumnMappings();
    SqoopHCatUtilities.instance().validateDynamicPartitionKeysMapping();
    job.setOutputFormatClass(getOutputFormatClass());
    IntWritable[] delimChars = new IntWritable[5];
    String hiveReplacement = "";
    LOG.debug("Hive delimiters will be fixed during import");
    DelimiterSet delims = opts.getOutputDelimiters();
    if (!opts.explicitOutputDelims()) {
      delims = DelimiterSet.HIVE_DELIMITERS;
    }
    delimChars = new IntWritable[] {
      new IntWritable(delims.getFieldsTerminatedBy()),
      new IntWritable(delims.getLinesTerminatedBy()),
      new IntWritable(delims.getEnclosedBy()),
      new IntWritable(delims.getEscapedBy()),
      new IntWritable(delims.isEncloseRequired() ? 1 : 0), };
    hiveReplacement = opts.getHiveDelimsReplacement();
    if (hiveReplacement == null) {
      hiveReplacement = "";
    }

    LOG.debug("Setting hive delimiters information");
    DefaultStringifier.storeArray(config, delimChars,
      HIVE_DELIMITERS_TO_REPLACE_PROP);
    config.set(HIVE_DELIMITERS_REPLACEMENT_PROP, hiveReplacement);
    if (opts.doHiveDropDelims() || opts.getHiveDelimsReplacement() != null) {
      LOG.debug("Enabling hive delimter replacement");
      config.set(HIVE_DELIMITERS_REPLACEMENT_ENABLED_PROP, "true");
    } else {
      LOG.debug("Disabling hive delimter replacement");
      config.set(HIVE_DELIMITERS_REPLACEMENT_ENABLED_PROP, "false");
    }
  }

  public static void configureExportInputFormat(SqoopOptions opts, Job job,
    ConnManager connMgr, String dbTable, Configuration config)
    throws IOException {

    LOG.info("Configuring HCatalog for export job");
    SqoopHCatUtilities hCatUtils = SqoopHCatUtilities.instance();
    hCatUtils
      .configureHCat(opts, job, connMgr, dbTable, job.getConfiguration());
    job.setInputFormatClass(getInputFormatClass());
    Map<String, List<Integer>> dbColInfo = hCatUtils.getDbColumnInfo();
    MapWritable columnTypesJava = new MapWritable();
    Properties mapColumnJava = opts.getMapColumnJava();
    for (Map.Entry<String, List<Integer>> e : dbColInfo.entrySet()) {
      Text columnName = new Text(e.getKey());
      Text columnText = null;
      if (mapColumnJava.containsKey(e.getKey())) {
        columnText = new Text(mapColumnJava.getProperty(e.getKey()));
      } else {
        columnText = new Text(connMgr.toJavaType(dbTable, e.getKey(),
          e.getValue().get(0)));
      }
      columnTypesJava.put(columnName, columnText);
    }
    MapWritable columnTypesSql = new MapWritable();
    for (Map.Entry<String, List<Integer>> e : dbColInfo.entrySet()) {
      Text columnName = new Text(e.getKey());
      IntWritable sqlType = new IntWritable(e.getValue().get(0));
      columnTypesSql.put(columnName, sqlType);
    }
    DefaultStringifier.store(config, columnTypesJava,
      SqoopHCatUtilities.HCAT_DB_OUTPUT_COLTYPES_JAVA);
    DefaultStringifier.store(config, columnTypesSql,
      SqoopHCatUtilities.HCAT_DB_OUTPUT_COLTYPES_SQL);
  }

  /**
   * Add the Hive and HCatalog jar files to local classpath and dist cache.
   * @throws IOException
   */
  public static void addJars(Job job, SqoopOptions options) throws IOException {

    if (ConfigurationHelper.isLocalJobTracker(job.getConfiguration())) {
      LOG.info("Not adding hcatalog jars to distributed cache in local mode");
      return;
    }
    if (options.isSkipDistCache()) {
      LOG.info("Not adding hcatalog jars to distributed cache as requested");
      return;
    }
    Configuration conf = job.getConfiguration();
    String hiveHome = null;
    String hCatHome = null;
    FileSystem fs = FileSystem.getLocal(conf);
    if (options != null) {
      hiveHome = options.getHiveHome();
    }
    if (hiveHome == null) {
      hiveHome = SqoopOptions.getHiveHomeDefault();
    }
    if (options != null) {
      hCatHome = options.getHCatHome();
    }
    if (hCatHome == null) {
      hCatHome = SqoopOptions.getHCatHomeDefault();
    }
    LOG.info("HCatalog job : Hive Home = " + hiveHome);
    LOG.info("HCatalog job:  HCatalog Home = " + hCatHome);

    conf.addResource(hiveHome + HIVESITEXMLPATH);

    // Add these to the 'tmpjars' array, which the MR JobSubmitter
    // will upload to HDFS and put in the DistributedCache libjars.
    List<String> libDirs = new ArrayList<String>();
    libDirs.add(hCatHome + File.separator + HCATSHAREDIR);
    libDirs.add(hCatHome + File.separator + DEFLIBDIR);
    libDirs.add(hiveHome + File.separator + DEFLIBDIR);
    Set<String> localUrls = new HashSet<String>();
    // Add any libjars already specified
    localUrls
      .addAll(conf
        .getStringCollection(
        ConfigurationConstants.MAPRED_DISTCACHE_CONF_PARAM));
    for (String dir : libDirs) {
      LOG.info("Adding jar files under " + dir + " to distributed cache");
      addDirToCache(new File(dir), fs, localUrls, false);
    }

    // Recursively add all hcatalog storage handler jars
    // The HBase storage handler is getting deprecated post Hive+HCat merge
    String hCatStorageHandlerDir = hCatHome + File.separator
      + "share/hcatalog/storage-handlers";
    LOG.info("Adding jar files under " + hCatStorageHandlerDir
      + " to distributed cache (recursively)");

    addDirToCache(new File(hCatStorageHandlerDir), fs, localUrls, true);

    String tmpjars = conf
      .get(ConfigurationConstants.MAPRED_DISTCACHE_CONF_PARAM);
    StringBuilder sb = new StringBuilder(1024);
    if (null != tmpjars) {
      sb.append(tmpjars);
      sb.append(",");
    }
    sb.append(StringUtils.arrayToString(localUrls.toArray(new String[0])));
    conf.set(ConfigurationConstants.MAPRED_DISTCACHE_CONF_PARAM, sb.toString());
  }

  /**
   * Add the .jar elements of a directory to the DCache classpath, optionally
   * recursively.
   */
  private static void addDirToCache(File dir, FileSystem fs,
    Set<String> localUrls, boolean recursive) {
    if (dir == null) {
      return;
    }

    File[] fileList = dir.listFiles();

    if (fileList == null) {
      LOG.warn("No files under " + dir
        + " to add to distributed cache for hcatalog job");
      return;
    }

    for (File libFile : dir.listFiles()) {
      if (libFile.exists() && !libFile.isDirectory()
        && libFile.getName().endsWith("jar")) {
        Path p = new Path(libFile.toString());
        if (libFile.canRead()) {
          String qualified = p.makeQualified(fs).toString();
          LOG.info("Adding to job classpath: " + qualified);
          localUrls.add(qualified);
        } else {
          LOG.warn("Ignoring unreadable file " + libFile);
        }
      }
      if (recursive && libFile.isDirectory()) {
        addDirToCache(libFile, fs, localUrls, recursive);
      }
    }
  }

  public static boolean isHadoop1() {
    String version = org.apache.hadoop.util.VersionInfo.getVersion();
    if (version.matches("\\b0\\.20\\..+\\b")
      || version.matches("\\b1\\.\\d\\.\\d")) {
      return true;
    }
    return false;
  }
  /**
  * Resolve a database-specific type to HCat data type. Largely follows Sqoop's
  * hive translation.
  * @param sqlType
  *          sql type
  * @return hcat type
  */
 public static String toHCatType(int sqlType) {
   switch (sqlType) {

   // Ideally TINYINT and SMALLINT should be mapped to their
   // HCat equivalents tinyint and smallint respectively
   // But the Sqoop Java type conversion has them mapped to Integer
   // Even though the referenced Java doc clearly recommends otherwise.
   // Changing this now can cause many of the sequence file usages to
   // break as value class implementations will change. So, we
   // just use the same behavior here.
     case Types.SMALLINT:
     case Types.TINYINT:
     case Types.INTEGER:
       return "int";

     case Types.VARCHAR:
       return "varchar";
     case Types.CHAR:
       return "char";
     case Types.LONGVARCHAR:
     case Types.NVARCHAR:
     case Types.NCHAR:
     case Types.LONGNVARCHAR:
     case Types.DATE:
     case Types.TIME:
     case Types.TIMESTAMP:
     case Types.CLOB:
       return "string";

     case Types.FLOAT:
     case Types.REAL:
       return "float";

     case Types.NUMERIC:
     case Types.DECIMAL:
       return "decimal";

     case Types.DOUBLE:
       return "double";

     case Types.BIT:
     case Types.BOOLEAN:
       return "boolean";

     case Types.BIGINT:
       return "bigint";

     case Types.BINARY:
     case Types.VARBINARY:
     case Types.BLOB:
     case Types.LONGVARBINARY:
       return "binary";

     default:
       throw new IllegalArgumentException(
         "Cannot convert SQL type to HCatalog type " + sqlType);
   }
 }


  public void invokeOutputCommitterForLocalMode(Job job) throws IOException {
    if (ConfigurationHelper.isLocalJobTracker(job.getConfiguration()) && isHadoop1()) {
      // HCatalog 0.11- do have special class HCatHadoopShims, however this
      // class got merged into Hive Shim layer in 0.12+. Following method will
      // try to find correct implementation via reflection.
      HadoopShims shims = ShimLoader.getHadoopShims();
      HCatHadoopShims hcatShims = shims.getHCatShim();
      try {
        hcatShims.commitJob(new HCatOutputFormat(), job);
      } catch (Exception e) {
        throw new RuntimeException("Can't explicitly commit job", e);
      }
    }
  }

  public void launchHCatCli(String cmdLine)
    throws IOException {
    String tmpFileName = null;


    String tmpDir = System.getProperty("java.io.tmpdir");
    if (options != null) {
      tmpDir = options.getTempDir();
    }
    tmpFileName =
      new File(tmpDir, "hcat-script-"
        + System.currentTimeMillis()).getAbsolutePath();

    writeHCatScriptFile(tmpFileName, cmdLine);
    // Create the argv for the HCatalog Cli Driver.
    String[] argArray = new String[2];
    argArray[0] = "-f";
    argArray[1] = tmpFileName;
    String argLine = StringUtils.join(",", Arrays.asList(argArray));

    if (testMode) {
      LOG.debug("Executing HCatalog CLI in-process with " + argLine);
      executeHCatProgramInProcess(argArray);
    } else {
      LOG.info("Executing external HCatalog CLI process with args :" + argLine);
      executeExternalHCatProgram(Executor.getCurEnvpStrings(), argArray);
    }
  }

  public void writeHCatScriptFile(String fileName, String contents)
    throws IOException {
    BufferedWriter w = null;
    try {
      FileOutputStream fos = new FileOutputStream(fileName);
      w = new BufferedWriter(new OutputStreamWriter(fos));
      w.write(contents, 0, contents.length());
    } catch (IOException ioe) {
      LOG.error("Error writing HCatalog load-in script", ioe);
      throw ioe;
    } finally {
      if (null != w) {
        try {
          w.close();
        } catch (IOException ioe) {
          LOG.warn("IOException closing stream to HCatalog script", ioe);
        }
      }
    }
  }

  /**
   * Execute HCat via an external 'bin/hcat' process.
   * @param env
   *          the environment strings to pass to any subprocess.
   * @throws IOException
   *           if HCatalog did not exit successfully.
   */
  public void executeExternalHCatProgram(List<String> env, String[] cmdLine)
    throws IOException {
    // run HCat command with the given args
    String hCatProgram = getHCatPath();
    ArrayList<String> args = new ArrayList<String>();
    if (Shell.WINDOWS) {
      // windows depends on python to be available
      args.add("python");
    }
    args.add(hCatProgram);
    if (cmdLine != null && cmdLine.length > 0) {
      for (String s : cmdLine) {
        args.add(s);
      }
    }
    LoggingAsyncSink logSink = new LoggingAsyncSink(LOG);
    int ret = Executor.exec(args.toArray(new String[0]),
      env.toArray(new String[0]), logSink, logSink);
    if (0 != ret) {
      throw new IOException("HCat exited with status " + ret);
    }
  }

  public void executeHCatProgramInProcess(String[] argv) throws IOException {
    SubprocessSecurityManager subprocessSM = null;

    try {
      Class<?> cliDriverClass = Class.forName(HCAT_CLI_MAIN_CLASS);
      subprocessSM = new SubprocessSecurityManager();
      subprocessSM.install();
      Method mainMethod = cliDriverClass.getMethod("main", argv.getClass());
      mainMethod.invoke(null, (Object) argv);
    } catch (ClassNotFoundException cnfe) {
      throw new IOException("HCatalog class not found", cnfe);
    } catch (NoSuchMethodException nsme) {
      throw new IOException("Could not access HCatCli.main()", nsme);
    } catch (IllegalAccessException iae) {
      throw new IOException("Could not access HatCli.main()", iae);
    } catch (InvocationTargetException ite) {
      // This may have been the ExitSecurityException triggered by the
      // SubprocessSecurityManager.
      Throwable cause = ite.getCause();
      if (cause instanceof ExitSecurityException) {
        ExitSecurityException ese = (ExitSecurityException) cause;
        int status = ese.getExitStatus();
        if (status != 0) {
          throw new IOException("HCatCli  exited with status=" + status);
        }
      } else {
        throw new IOException("Exception thrown from HCatCli", ite);
      }
    } finally {
      if (null != subprocessSM) {
        subprocessSM.uninstall();
      }
    }
  }

  /**
   * @return the filename of the hcat executable to run to do the import
   */
  public String getHCatPath() {
    String hCatHome = null;
    if (options == null) {
      hCatHome = SqoopOptions.getHCatHomeDefault();
    } else {
      hCatHome = options.getHCatHome();
    }

    if (null == hCatHome) {
      return null;
    }

    Path p = new Path(hCatHome);
    p = new Path(p, "bin");
    p = new Path(p, HCATCMD);
    String hCatBinStr = p.toString();
    if (new File(hCatBinStr).canExecute()) {
      return hCatBinStr;
    } else {
      return null;
    }
  }

  public static boolean isTestMode() {
    return testMode;
  }

  public static void setTestMode(boolean mode) {
    testMode = mode;
  }

  public static Class<? extends InputFormat> getInputFormatClass() {
    return inputFormatClass;
  }

  public static Class<? extends OutputFormat> getOutputFormatClass() {
    return outputFormatClass;
  }

  public static void setInputFormatClass(Class<? extends InputFormat> clz) {
    inputFormatClass = clz;
  }

  public static void setOutputFormatClass(Class<? extends OutputFormat> clz) {
    outputFormatClass = clz;
  }

  public static Class<? extends Mapper> getImportMapperClass() {
    return importMapperClass;
  }

  public static Class<? extends Mapper> getExportMapperClass() {
    return exportMapperClass;
  }

  public static void setExportMapperClass(Class<? extends Mapper> clz) {
    exportMapperClass = clz;
  }

  public static void setImportMapperClass(Class<? extends Mapper> clz) {
    importMapperClass = clz;
  }

  public static Class<? extends Writable> getImportValueClass() {
    return importValueClass;
  }

  public static void setImportValueClass(Class<? extends Writable> clz) {
    importValueClass = clz;
  }


  public String getDatabaseTable() {
    return dbTableName;
  }

  public String getHCatTableName() {
    return hCatTableName;
  }

  public String getHCatDatabaseName() {
    return hCatDatabaseName;
  }

  public String getQualifiedHCatTableName() {
    return hCatQualifiedTableName;
  }

  public List<String> getHCatDynamicPartitionKeys() {
    return hCatDynamicPartitionKeys;
  }

  public List<String> getHCatStaticPartitionKey() {
    return hCatStaticPartitionKeys;
  }

  public String[] getDBColumnNames() {
    return dbColumnNames;
  }

  public HCatSchema getHCatOutputSchema() {
    return hCatOutputSchema;
  }

  public void setHCatOutputSchema(HCatSchema schema) {
    hCatOutputSchema = schema;
  }

  public HCatSchema getHCatPartitionSchema() {
    return hCatPartitionSchema;
  }

  public void setHCatPartitionSchema(HCatSchema schema) {
    hCatPartitionSchema = schema;
  }

  public void setHCatStaticPartitionKey(List<String> keys) {
    hCatStaticPartitionKeys = keys;
  }

  public void setHCatDynamicPartitionKeys(List<String> keys) {
    hCatDynamicPartitionKeys = keys;
  }

  public String[] getDbColumnNames() {
    return dbColumnNames;
  }

  public void setDbColumnNames(String[] names) {
    dbColumnNames = names;
  }

  public Map<String, List<Integer>> getDbColumnInfo() {
    return dbColumnInfo;
  }

  public String gethCatTableName() {
    return hCatTableName;
  }

  public String gethCatDatabaseName() {
    return hCatDatabaseName;
  }

  public String gethCatQualifiedTableName() {
    return hCatQualifiedTableName;
  }

  public void setConfigured(boolean value) {
    configured = value;
  }

  public static String sqlTypeString(int sqlType) {
    switch (sqlType) {
      case Types.BIT:
        return "BIT";
      case Types.TINYINT:
        return "TINYINT";
      case Types.SMALLINT:
        return "SMALLINT";
      case Types.INTEGER:
        return "INTEGER";
      case Types.BIGINT:
        return "BIGINT";
      case Types.FLOAT:
        return "FLOAT";
      case Types.REAL:
        return "REAL";
      case Types.DOUBLE:
        return "DOUBLE";
      case Types.NUMERIC:
        return "NUMERIC";
      case Types.DECIMAL:
        return "DECIMAL";
      case Types.CHAR:
        return "CHAR";
      case Types.VARCHAR:
        return "VARCHAR";
      case Types.LONGVARCHAR:
        return "LONGVARCHAR";
      case Types.DATE:
        return "DATE";
      case Types.TIME:
        return "TIME";
      case Types.TIMESTAMP:
        return "TIMESTAMP";
      case Types.BINARY:
        return "BINARY";
      case Types.VARBINARY:
        return "VARBINARY";
      case Types.LONGVARBINARY:
        return "LONGVARBINARY";
      case Types.NULL:
        return "NULL";
      case Types.OTHER:
        return "OTHER";
      case Types.JAVA_OBJECT:
        return "JAVA_OBJECT";
      case Types.DISTINCT:
        return "DISTINCT";
      case Types.STRUCT:
        return "STRUCT";
      case Types.ARRAY:
        return "ARRAY";
      case Types.BLOB:
        return "BLOB";
      case Types.CLOB:
        return "CLOB";
      case Types.REF:
        return "REF";
      case Types.DATALINK:
        return "DATALINK";
      case Types.BOOLEAN:
        return "BOOLEAN";
      case Types.ROWID:
        return "ROWID";
      case Types.NCHAR:
        return "NCHAR";
      case Types.NVARCHAR:
        return "NVARCHAR";
      case Types.LONGNVARCHAR:
        return "LONGNVARCHAR";
      case Types.NCLOB:
        return "NCLOB";
      case Types.SQLXML:
        return "SQLXML";
      default:
        return "<UNKNOWN>";
    }
  }
}