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

import static org.apache.sqoop.manager.JdbcDrivers.ORACLE;

import java.io.IOException;
import java.lang.reflect.Method;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.TimeZone;
import java.util.TreeMap;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.sqoop.manager.oracle.OracleUtils;
import org.apache.sqoop.util.LoggingUtils;

import com.cloudera.sqoop.SqoopOptions;
import com.cloudera.sqoop.SqoopOptions.UpdateMode;
import com.cloudera.sqoop.mapreduce.ExportBatchOutputFormat;
import com.cloudera.sqoop.mapreduce.JdbcExportJob;
import com.cloudera.sqoop.mapreduce.JdbcUpsertExportJob;
import com.cloudera.sqoop.mapreduce.OracleUpsertOutputFormat;
import com.cloudera.sqoop.mapreduce.db.OracleDataDrivenDBInputFormat;
import com.cloudera.sqoop.util.ExportException;
import com.cloudera.sqoop.util.ImportException;

/**
 * Manages connections to Oracle databases.
 * Requires the Oracle JDBC driver.
 */
public class OracleManager
    extends com.cloudera.sqoop.manager.GenericJdbcManager {

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

  /**
   * ORA-00942: Table or view does not exist. Indicates that the user does
   * not have permissions.
   */
  public static final int ERROR_TABLE_OR_VIEW_DOES_NOT_EXIST = 942;

  /**
   * This is a catalog view query to list the databases. For Oracle we map the
   * concept of a database to a schema, and a schema is identified by a user.
   * In order for the catalog view DBA_USERS be visible to the user who executes
   * this query, they must have the DBA privilege.
   */
  public static final String QUERY_LIST_DATABASES =
    "SELECT USERNAME FROM DBA_USERS";

  /**
   * Query to list all tables visible to the current user. Note that this list
   * does not identify the table owners which is required in order to
   * ensure that the table can be operated on for import/export purposes.
   */
  public static final String QUERY_LIST_TABLES =
    "SELECT TABLE_NAME FROM ALL_TABLES WHERE OWNER = ?";

  /**
   * Query to list all columns of the given table. Even if the user has the
   * privileges to access table objects from another schema, this query will
   * limit it to explore tables only from within the active schema.
   */
  public static final String QUERY_COLUMNS_FOR_TABLE =
          "SELECT COLUMN_NAME FROM ALL_TAB_COLUMNS WHERE "
        + "OWNER = ? AND TABLE_NAME = ? ORDER BY COLUMN_ID";

  /**
   * Query to find the primary key column name for a given table. This query
   * is restricted to the current schema.
   */
  public static final String QUERY_PRIMARY_KEY_FOR_TABLE =
    "SELECT ALL_CONS_COLUMNS.COLUMN_NAME FROM ALL_CONS_COLUMNS, "
     + "ALL_CONSTRAINTS WHERE ALL_CONS_COLUMNS.CONSTRAINT_NAME = "
     + "ALL_CONSTRAINTS.CONSTRAINT_NAME AND "
     + "ALL_CONSTRAINTS.CONSTRAINT_TYPE = 'P' AND "
     + "ALL_CONS_COLUMNS.TABLE_NAME = ? AND "
     + "ALL_CONS_COLUMNS.OWNER = ?";

  /**
   * Query to get the current user for the DB session.   Used in case of
   * wallet logins.
   */
  public static final String QUERY_GET_SESSIONUSER =
     "SELECT USER FROM DUAL";

  // Configuration key to use to set the session timezone.
  public static final String ORACLE_TIMEZONE_KEY = "oracle.sessionTimeZone";

  // Oracle XE does a poor job of releasing server-side resources for
  // closed connections. So we actually want to cache connections as
  // much as possible. This is especially important for JUnit tests which
  // may need to make 60 or more connections (serially), since each test
  // uses a different OracleManager instance.
  private static class ConnCache {

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

    private static class CacheKey {
      private final String connectString;
      private final String username;

      public CacheKey(String connect, String user) {
        this.connectString = connect;
        this.username = user; // note: may be null.
      }

      @Override
      public boolean equals(Object o) {
        if (o instanceof CacheKey) {
          CacheKey k = (CacheKey) o;
          if (null == username) {
            return k.username == null && k.connectString.equals(connectString);
          } else {
            return k.username.equals(username)
                && k.connectString.equals(connectString);
          }
        } else {
          return false;
        }
      }

      @Override
      public int hashCode() {
        if (null == username) {
          return connectString.hashCode();
        } else {
          return username.hashCode() ^ connectString.hashCode();
        }
      }

      @Override
      public String toString() {
        return connectString + "/" + username;
      }
    }

    private Map<CacheKey, Connection> connectionMap;

    public ConnCache() {
      LOG.debug("Instantiated new connection cache.");
      connectionMap = new HashMap<CacheKey, Connection>();
    }

    /**
     * @return a Connection instance that can be used to connect to the
     * given database, if a previously-opened connection is available in
     * the cache. Returns null if none is available in the map.
     */
    public synchronized Connection getConnection(String connectStr,
        String username) throws SQLException {
      CacheKey key = new CacheKey(connectStr, username);
      Connection cached = connectionMap.get(key);
      if (null != cached) {
        connectionMap.remove(key);
        if (cached.isReadOnly()) {
          // Read-only mode? Don't want it.
          cached.close();
        }

        if (cached.isClosed()) {
          // This connection isn't usable.
          return null;
        }

        cached.rollback(); // Reset any transaction state.
        cached.clearWarnings();

        LOG.debug("Got cached connection for " + key);
      }

      return cached;
    }

    /**
     * Returns a connection to the cache pool for future use. If a connection
     * is already cached for the connectstring/username pair, then this
     * connection is closed and discarded.
     */
    public synchronized void recycle(String connectStr, String username,
        Connection conn) throws SQLException {

      CacheKey key = new CacheKey(connectStr, username);
      Connection existing = connectionMap.get(key);
      if (null != existing) {
        // Cache is already full for this entry.
        LOG.debug("Discarding additional connection for " + key);
        conn.close();
        return;
      }

      // Put it in the map for later use.
      LOG.debug("Caching released connection for " + key);
      connectionMap.put(key, conn);
    }

    @Override
    protected synchronized void finalize() throws Throwable {
      for (Connection c : connectionMap.values()) {
        c.close();
      }

      super.finalize();
    }
  }

  private static final ConnCache CACHE;
  static {
    CACHE = new ConnCache();
  }

  public OracleManager(final SqoopOptions opts) {
    super(ORACLE.getDriverClass(), opts);
  }

  public void close() throws SQLException {
    release(); // Release any open statements associated with the connection.
    if (hasOpenConnection()) {
      // Release our open connection back to the cache.
      CACHE.recycle(options.getConnectString(), options.getUsername(),
          getConnection());
      discardConnection(false);
    }
  }

  protected String getColNamesQuery(String tableName) {
    // SqlManager uses "tableName AS t" which doesn't work in Oracle.
    String query =  "SELECT t.* FROM " + escapeTableName(tableName)
            + " t WHERE 1=0";

    LOG.debug("Using column names query: " + query);
    return query;
  }

  /**
   * Create a connection to the database; usually used only from within
   * getConnection(), which enforces a singleton guarantee around the
   * Connection object.
   *
   * Oracle-specific driver uses READ_COMMITTED which is the weakest
   * semantics Oracle supports.
   */
  protected Connection makeConnection() throws SQLException {

    Connection connection;
    String driverClass = getDriverClass();

    try {
      Class.forName(driverClass);
    } catch (ClassNotFoundException cnfe) {
      throw new RuntimeException("Could not load db driver class: "
          + driverClass);
    }

    String username = options.getUsername();
    String password = options.getPassword();
    String connectStr = options.getConnectString();

    try {
      connection = CACHE.getConnection(connectStr, username);
    } catch (SQLException e) {
      connection = null;
      LOG.debug("Cached connecion has expired.");
    }

    if (null == connection) {
      // Couldn't pull one from the cache. Get a new one.
      LOG.debug("Creating a new connection for "
              + connectStr + ", using username: " + username);
      Properties connectionParams = options.getConnectionParams();
      if (connectionParams != null && connectionParams.size() > 0) {
        LOG.debug("User specified connection params. "
                  + "Using properties specific API for making connection.");

        Properties props = new Properties();
        if (username != null) {
          props.put("user", username);
        }

        if (password != null) {
          props.put("password", password);
        }

        props.putAll(connectionParams);
        connection = DriverManager.getConnection(connectStr, props);
      } else {
        LOG.debug("No connection paramenters specified. "
                + "Using regular API for making connection.");
        if (username == null) {
          connection = DriverManager.getConnection(connectStr);
        } else {
          connection = DriverManager.getConnection(
                              connectStr, username, password);
        }
      }
    }

    // We only use this for metadata queries. Loosest semantics are okay.
    connection.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);

    // Setting session time zone
    setSessionTimeZone(connection);

    // Rest of the Sqoop code expects that the connection will have be running
    // without autoCommit, so we need to explicitly set it to false. This is
    // usually done directly by SqlManager in the makeConnection method, but
    // since we are overriding it, we have to do it ourselves.
    connection.setAutoCommit(false);

    return connection;
  }

  public static String getSessionUser(Connection conn) {
    Statement stmt = null;
    ResultSet rset = null;
    String user = null;
    try {
      stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY,
              ResultSet.CONCUR_READ_ONLY);
      rset = stmt.executeQuery(QUERY_GET_SESSIONUSER);

      if (rset.next()) {
        user = rset.getString(1);
      }
      conn.commit();
    } catch (SQLException e) {
      try {
        conn.rollback();
      } catch (SQLException ex) {
        LoggingUtils.logAll(LOG, "Failed to rollback transaction", ex);
      }
    } finally {
      if (rset != null) {
        try {
          rset.close();
        } catch (SQLException ex) {
          LoggingUtils.logAll(LOG, "Failed to close resultset", ex);
        }
      }
      if (stmt != null) {
        try {
          stmt.close();
        } catch (SQLException ex) {
          LoggingUtils.logAll(LOG, "Failed to close statement", ex);
        }
      }
    }
    if (user == null) {
      throw new RuntimeException("Unable to get current session user");
    }
    return user;
  }

  /**
   * Set session time zone.
   * @param conn      Connection object
   * @throws          SQLException instance
   */
  private void setSessionTimeZone(Connection conn) throws SQLException {
    // Need to use reflection to call the method setSessionTimeZone on the
    // OracleConnection class because oracle specific java libraries are not
    // accessible in this context.
    Method methodSession;
    Method methodDefaultTimezone;
    try {
      methodSession = conn.getClass().getMethod(
              "setSessionTimeZone", new Class [] {String.class});
      methodDefaultTimezone = conn.getClass().getMethod("setDefaultTimeZone", TimeZone.class);
    } catch (Exception ex) {
      LOG.error("Could not find method setSessionTimeZone in "
        + conn.getClass().getName(), ex);
      // rethrow SQLException
      throw new SQLException(ex);
    }

    // Need to set the time zone in order for Java to correctly access the
    // column "TIMESTAMP WITH LOCAL TIME ZONE".  The user may have set this in
    // the configuration as 'oracle.sessionTimeZone'.
    String clientTimeZoneStr = options.getConf().get(ORACLE_TIMEZONE_KEY,
        "GMT");
    TimeZone timeZone = TimeZone.getTimeZone(clientTimeZoneStr);
    TimeZone.setDefault(timeZone);
    try {
      methodSession.setAccessible(true);
      methodSession.invoke(conn, clientTimeZoneStr);
      methodDefaultTimezone.setAccessible(true);
      methodDefaultTimezone.invoke(conn, timeZone);
      LOG.info("Time zone has been set to " + clientTimeZoneStr);
    } catch (Exception ex) {
      LOG.warn("Time zone " + clientTimeZoneStr
               + " could not be set on Oracle database.");
      LOG.info("Setting default time zone: GMT");
      try {
        // Per the documentation at:
        // http://download-west.oracle.com/docs/cd/B19306_01
        //     /server.102/b14225/applocaledata.htm#i637736
        // The "GMT" timezone is guaranteed to exist in the available timezone
        // regions, whereas others (e.g., "UTC") are not.
        methodSession.invoke(conn, "GMT");
        methodDefaultTimezone.invoke(conn, "GMT");
        TimeZone.setDefault(TimeZone.getTimeZone("GMT"));
      } catch (Exception ex2) {
        LOG.error("Could not set time zone for oracle connection", ex2);
        // rethrow SQLException
        throw new SQLException(ex);
      }
    }
  }

  @Override
  public void importTable(
          com.cloudera.sqoop.manager.ImportJobContext context)
      throws IOException, ImportException {
    context.setConnManager(this);
    // Specify the Oracle-specific DBInputFormat for import.
    context.setInputFormat(OracleDataDrivenDBInputFormat.class);
    super.importTable(context);
  }

  /**
   * Export data stored in HDFS into a table in a database.
   */
  public void exportTable(com.cloudera.sqoop.manager.ExportJobContext context)
      throws IOException, ExportException {
    context.setConnManager(this);
    JdbcExportJob exportJob = new JdbcExportJob(context,
            null, null, ExportBatchOutputFormat.class);
    exportJob.runExport();
  }

  @Override
  /**
   * {@inheritDoc}
   */
  public void upsertTable(com.cloudera.sqoop.manager.ExportJobContext context)
      throws IOException, ExportException {
    context.setConnManager(this);
    JdbcUpsertExportJob exportJob =
      new JdbcUpsertExportJob(context, OracleUpsertOutputFormat.class);
    exportJob.runExport();
  }

  @Override
  /**
   * {@inheritDoc}
   */
  public void configureDbOutputColumns(SqoopOptions options) {
    if (options.getUpdateMode() == UpdateMode.UpdateOnly) {
      super.configureDbOutputColumns(options);
    } else {
      // We're in upsert mode. We need to explicitly set
      // the database output column ordering in the codeGenerator.
      Set<String> updateKeys = new LinkedHashSet<String>();
      Set<String> updateKeysUppercase = new HashSet<String>();
      String updateKeyValue = options.getUpdateKeyCol();
      StringTokenizer stok = new StringTokenizer(updateKeyValue, ",");
      while (stok.hasMoreTokens()) {
        String nextUpdateColumn = stok.nextToken().trim();
        if (nextUpdateColumn.length() > 0) {
          updateKeys.add(nextUpdateColumn);
          updateKeysUppercase.add(nextUpdateColumn.toUpperCase());
        }  else {
          throw new RuntimeException("Invalid update key column value specified"
              + ": '" + updateKeyValue + "'");
        }
      }

      String [] allColNames = getColumnNames(options.getTableName());
      List<String> dbOutCols = new ArrayList<String>();
      dbOutCols.addAll(updateKeys);
      for (String col : allColNames) {
        if (!updateKeysUppercase.contains(col.toUpperCase())) {
          dbOutCols.add(col); // add update columns to the output order list.
        }
      }
      for (String col : allColNames) {
        dbOutCols.add(col); // add insert columns to the output order list.
      }
      options.setDbOutputColumns(dbOutCols.toArray(
          new String[dbOutCols.size()]));
    }
  }

  @Override
  public ResultSet readTable(String tableName, String[] columns)
      throws SQLException {
    if (columns == null) {
      columns = getColumnNames(tableName);
    }

    StringBuilder sb = new StringBuilder();
    sb.append("SELECT ");
    boolean first = true;
    for (String col : columns) {
      if (!first) {
        sb.append(", ");
      }
      sb.append(escapeColName(col));
      first = false;
    }
    sb.append(" FROM ");
    sb.append(escapeTableName(tableName));

    String sqlCmd = sb.toString();
    LOG.debug("Reading table with command: " + sqlCmd);
    return execute(sqlCmd);
  }

  private Map<String, String> columnTypeNames;

  /**
   * Resolve a database-specific type to the Java type that should contain it.
   * @param tableName   table name
   * @param colName  column name
   * @return the name of a Java type to hold the sql datatype, or null if none.
   */
  private String toDbSpecificJavaType(String tableName, String colName) {
    if (columnTypeNames == null) {
      columnTypeNames = getColumnTypeNames(tableName, options.getCall(),
        options.getSqlQuery());
    }

    String colTypeName = columnTypeNames.get(colName);
    if (colTypeName != null) {
      if (colTypeName.equalsIgnoreCase("BINARY_FLOAT")) {
        return "Float";
      }
      if (colTypeName.equalsIgnoreCase("FLOAT")) {
        return "Float";
      }
      if (colTypeName.equalsIgnoreCase("BINARY_DOUBLE")) {
        return "Double";
      }
      if (colTypeName.equalsIgnoreCase("DOUBLE")) {
        return "Double";
      }
      if (colTypeName.toUpperCase().startsWith("TIMESTAMP")) {
        return "java.sql.Timestamp";
      }
    }
    return null;
  }

  /**
   * Resolve a database-specific type to the Hive type that should contain it.
   * @param tableName   table name
   * @param colName  column name
   * @return the name of a Hive type to hold the sql datatype, or null if none.
   */
  private String toDbSpecificHiveType(String tableName, String colName) {
    if (columnTypeNames == null) {
      columnTypeNames = getColumnTypeNames(tableName, options.getCall(),
        options.getSqlQuery());
    }
    LOG.debug("Column Types and names returned = ("
      + StringUtils.join(columnTypeNames.keySet(), ",")
      + ")=>("
      + StringUtils.join(columnTypeNames.values(), ",")
      + ")");

    String colTypeName = columnTypeNames.get(colName);
    if (colTypeName != null) {
      if (colTypeName.equalsIgnoreCase("BINARY_FLOAT")) {
        return "FLOAT";
      }
      if (colTypeName.equalsIgnoreCase("BINARY_DOUBLE")) {
        return "DOUBLE";
      }
      if (colTypeName.toUpperCase().startsWith("TIMESTAMP")) {
        return "STRING";
      }
    }
    return null;
  }

  /**
   * Return java type for SQL type.
   * @param tableName   table name
   * @param columnName  column name
   * @param sqlType     sql type
   * @return            java type
   */
  @Override
  public String toJavaType(String tableName, String columnName, int sqlType) {
    String javaType = super.toJavaType(tableName, columnName, sqlType);
    if (javaType == null) {
      javaType = toDbSpecificJavaType(tableName, columnName);
    }
    return javaType;
  }

  /**
   * Return hive type for SQL type.
   * @param tableName   table name
   * @param columnName  column name
   * @param sqlType     sql data type
   * @return            hive type
   */
  @Override
  public String toHiveType(String tableName, String columnName, int sqlType) {
    String hiveType = super.toHiveType(tableName, columnName, sqlType);
    if (hiveType == null) {
      hiveType = toDbSpecificHiveType(tableName, columnName);
    }
    return hiveType;
  }

  @Override
  protected void finalize() throws Throwable {
    close();
    super.finalize();
  }

  @Override
  protected String getCurTimestampQuery() {
    return "SELECT SYSDATE FROM dual";
  }

  @Override
  public String timestampToQueryString(Timestamp ts) {
    return "TO_TIMESTAMP('" + ts + "', 'YYYY-MM-DD HH24:MI:SS.FF')";
  }

  @Override
  public String datetimeToQueryString(String datetime, int columnType) {
    if (columnType == Types.TIMESTAMP) {
      return "TO_TIMESTAMP('" + datetime + "', 'YYYY-MM-DD HH24:MI:SS.FF')";
    } else if (columnType == Types.DATE) {
      // converting timestamp of the form 2012-11-11 11:11:11.00 to
      // date of the form 2011:11:11 11:11:11
      datetime = datetime.split("\\.")[0];
      return "TO_DATE('" + datetime + "', 'YYYY-MM-DD HH24:MI:SS')";
    } else {
      String msg = "Column type is neither timestamp nor date!";
      LOG.error(msg);
      throw new RuntimeException(msg);
    }
  }

  @Override
  public boolean supportsStagingForExport() {
    return true;
  }

  /**
   * The concept of database in Oracle is mapped to schemas. Each schema
   * is identified by the corresponding username.
   */
  @Override
  public String[] listDatabases() {
    Connection conn = null;
    Statement stmt = null;
    ResultSet rset = null;
    List<String> databases = new ArrayList<String>();

    try {
      conn = getConnection();
      stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY,
              ResultSet.CONCUR_READ_ONLY);
      rset = stmt.executeQuery(QUERY_LIST_DATABASES);

      while (rset.next()) {
        databases.add(rset.getString(1));
      }
      conn.commit();
    } catch (SQLException e) {
      try {
        conn.rollback();
      } catch (SQLException ex) {
        LoggingUtils.logAll(LOG, "Failed to rollback transaction", ex);
      }

      if (e.getErrorCode() == ERROR_TABLE_OR_VIEW_DOES_NOT_EXIST) {
        LOG.error("The catalog view DBA_USERS was not found. "
            + "This may happen if the user does not have DBA privileges. "
            + "Please check privileges and try again.");
        LOG.debug("Full trace for ORA-00942 exception", e);
      } else {
        LoggingUtils.logAll(LOG, "Failed to list databases", e);
      }
    } finally {
      if (rset != null) {
        try {
          rset.close();
        } catch (SQLException ex) {
          LoggingUtils.logAll(LOG, "Failed to close resultset", ex);
        }
      }
      if (stmt != null) {
        try {
          stmt.close();
        } catch (SQLException ex) {
          LoggingUtils.logAll(LOG, "Failed to close statement", ex);
        }
      }

      try {
        close();
      } catch (SQLException ex) {
        LoggingUtils.logAll(LOG, "Unable to discard connection", ex);
      }
    }

    return databases.toArray(new String[databases.size()]);
  }

  @Override
  public String[] listTables() {
    Connection conn = null;
    PreparedStatement pStmt = null;
    ResultSet rset = null;
    List<String> tables = new ArrayList<String>();
    String tableOwner = null;


    try {
      conn = getConnection();
      tableOwner = getSessionUser(conn);
      pStmt = conn.prepareStatement(QUERY_LIST_TABLES,
          ResultSet.TYPE_FORWARD_ONLY,
              ResultSet.CONCUR_READ_ONLY);
      pStmt.setString(1, tableOwner);

      rset = pStmt.executeQuery();

      while (rset.next()) {
        tables.add(rset.getString(1));
      }
      conn.commit();
    } catch (SQLException e) {
      try {
        conn.rollback();
      } catch (SQLException ex) {
        LoggingUtils.logAll(LOG, "Failed to rollback transaction", ex);
      }
      LoggingUtils.logAll(LOG, "Failed to list tables", e);
    } finally {
      if (rset != null) {
        try {
          rset.close();
        } catch (SQLException ex) {
          LoggingUtils.logAll(LOG, "Failed to close resultset", ex);
        }
      }
      if (pStmt != null) {
        try {
          pStmt.close();
        } catch (SQLException ex) {
          LoggingUtils.logAll(LOG, "Failed to close statement", ex);
        }
      }

      try {
        close();
      } catch (SQLException ex) {
        LoggingUtils.logAll(LOG, "Unable to discard connection", ex);
      }
    }

    return tables.toArray(new String[tables.size()]);
  }

  @Override
  public String[] getColumnNamesForProcedure(String procedureName) {
    List<String> ret = new ArrayList<String>();
    try {
      DatabaseMetaData metaData = this.getConnection().getMetaData();
      ResultSet results = metaData.getProcedureColumns(null, null,
        procedureName, null);
      if (null == results) {
        return null;
      }

      try {
        while (results.next()) {
          if (results.getInt("COLUMN_TYPE")
          != DatabaseMetaData.procedureColumnReturn) {
            int index = results.getInt("ORDINAL_POSITION");
            if (index < 0) {
              continue; // actually the return type
            }
            for (int i = ret.size(); i < index; ++i) {
              ret.add(null);
            }
            String name = results.getString("COLUMN_NAME");
            if (index == ret.size()) {
              ret.add(name);
            } else {
              ret.set(index, name);
            }
          }
        }
        String[] result = ret.toArray(new String[ret.size()]);
        LOG.debug("getColumnsNamesForProcedure returns "
          + StringUtils.join(ret, ","));
        return result;
      } finally {
        results.close();
        getConnection().commit();
      }
    } catch (SQLException e) {
      LoggingUtils.logAll(LOG, "Error reading procedure metadata: ", e);
      throw new RuntimeException("Can't fetch column names for procedure.", e);
    }
  }

  @Override
  public Map<String, Integer>
    getColumnTypesForProcedure(String procedureName) {
    Map<String, Integer> ret = new TreeMap<String, Integer>();
    try {
      DatabaseMetaData metaData = this.getConnection().getMetaData();
      ResultSet results = metaData.getProcedureColumns(null, null,
        procedureName, null);
      if (null == results) {
        return null;
      }

      try {
        while (results.next()) {
          if (results.getInt("COLUMN_TYPE")
          != DatabaseMetaData.procedureColumnReturn) {
            int index = results.getInt("ORDINAL_POSITION");
            if (index < 0) {
              continue; // actually the return type
            }
            // we don't care if we get several rows for the
            // same ORDINAL_POSITION (e.g. like H2 gives us)
            // as we'll just overwrite the entry in the map:
            ret.put(
              results.getString("COLUMN_NAME"),
              results.getInt("DATA_TYPE"));
          }
        }
        LOG.debug("Columns returned = " + StringUtils.join(ret.keySet(), ","));
        LOG.debug("Types returned = " + StringUtils.join(ret.values(), ","));
        return ret.isEmpty() ? null : ret;
      } finally {
        results.close();
        getConnection().commit();
      }
    } catch (SQLException sqlException) {
      LoggingUtils.logAll(LOG, "Error reading primary key metadata: "
        + sqlException.toString(), sqlException);
      return null;
    }
  }

  @Override
  public Map<String, String>
    getColumnTypeNamesForProcedure(String procedureName) {
    Map<String, String> ret = new TreeMap<String, String>();
    try {
      DatabaseMetaData metaData = this.getConnection().getMetaData();
      ResultSet results = metaData.getProcedureColumns(null, null,
        procedureName, null);
      if (null == results) {
        return null;
      }

      try {
        while (results.next()) {
          if (results.getInt("COLUMN_TYPE")
          != DatabaseMetaData.procedureColumnReturn) {
            int index = results.getInt("ORDINAL_POSITION");
            if (index < 0) {
              continue; // actually the return type
            }
            // we don't care if we get several rows for the
            // same ORDINAL_POSITION (e.g. like H2 gives us)
            // as we'll just overwrite the entry in the map:
            ret.put(
              results.getString("COLUMN_NAME"),
              results.getString("TYPE_NAME"));
          }
        }
        LOG.debug("Columns returned = " + StringUtils.join(ret.keySet(), ","));
        LOG.debug(
          "Type names returned = " + StringUtils.join(ret.values(), ","));
        return ret.isEmpty() ? null : ret;
      } finally {
        results.close();
        getConnection().commit();
      }
    } catch (SQLException sqlException) {
      LoggingUtils.logAll(LOG, "Error reading primary key metadata: "
        + sqlException.toString(), sqlException);
      return null;
    }
  }

  @Override
  public String escapeColName(String colName) {
    return OracleUtils.escapeIdentifier(colName, options.isOracleEscapingDisabled());
  }

  @Override
  public String escapeTableName(String tableName) {
    return OracleUtils.escapeIdentifier(tableName, options.isOracleEscapingDisabled());
  }

  @Override
  public boolean escapeTableNameOnExport() {
    return true;
  }

  @Override
  public String[] getColumnNames(String tableName) {
    Connection conn = null;
    PreparedStatement pStmt = null;
    ResultSet rset = null;
    List<String> columns = new ArrayList<String>();

    String tableOwner = null;
    String shortTableName = tableName;
    int qualifierIndex = tableName.indexOf('.');
    if (qualifierIndex != -1) {
      tableOwner = tableName.substring(0, qualifierIndex);
      shortTableName = tableName.substring(qualifierIndex + 1);
    }

    try {
      conn = getConnection();

      if (tableOwner == null) {
        tableOwner = getSessionUser(conn);
      }

      pStmt = conn.prepareStatement(QUERY_COLUMNS_FOR_TABLE,
                  ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);

      pStmt.setString(1, tableOwner);

      pStmt.setString(2, shortTableName);
      rset = pStmt.executeQuery();

      while (rset.next()) {
          columns.add(rset.getString(1));
      }
      conn.commit();
    } catch (SQLException e) {
      try {
        conn.rollback();
      } catch (SQLException ex) {
        LoggingUtils.logAll(LOG, "Failed to rollback transaction", ex);
      }
      LoggingUtils.logAll(LOG, "Failed to list columns", e);
    } finally {
      if (rset != null) {
        try {
          rset.close();
        } catch (SQLException ex) {
          LoggingUtils.logAll(LOG, "Failed to close resultset", ex);
        }
      }
      if (pStmt != null) {
        try {
          pStmt.close();
        } catch (SQLException ex) {
          LoggingUtils.logAll(LOG, "Failed to close statement", ex);
        }
      }

      try {
        close();
      } catch (SQLException ex) {
        LoggingUtils.logAll(LOG, "Unable to discard connection", ex);
      }
    }

    return filterSpecifiedColumnNames(columns.toArray(new String[columns.size()]));
  }

  @Override
  public String getPrimaryKey(String tableName) {
    Connection conn = null;
    PreparedStatement pStmt = null;
    ResultSet rset = null;
    List<String> columns = new ArrayList<String>();

    String tableOwner = null;
    String shortTableName = tableName;
    int qualifierIndex = tableName.indexOf('.');
    if (qualifierIndex != -1) {
      tableOwner = tableName.substring(0, qualifierIndex);
      shortTableName = tableName.substring(qualifierIndex + 1);
    }

    try {
      conn = getConnection();

      if (tableOwner == null) {
        tableOwner = getSessionUser(conn);
      }

      pStmt = conn.prepareStatement(QUERY_PRIMARY_KEY_FOR_TABLE,
                  ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
      pStmt.setString(1, shortTableName);
      pStmt.setString(2, tableOwner);
      rset = pStmt.executeQuery();

      while (rset.next()) {
        columns.add(rset.getString(1));
      }
      conn.commit();
    } catch (SQLException e) {
      try {
        if (conn != null) {
          conn.rollback();
        }
      } catch (SQLException ex) {
        LoggingUtils.logAll(LOG, "Failed to rollback transaction", ex);
      }
      LoggingUtils.logAll(LOG, "Failed to list columns", e);
    } finally {
      if (rset != null) {
        try {
          rset.close();
        } catch (SQLException ex) {
          LoggingUtils.logAll(LOG, "Failed to close resultset", ex);
        }
      }
      if (pStmt != null) {
        try {
          pStmt.close();
        } catch (SQLException ex) {
          LoggingUtils.logAll(LOG, "Failed to close statement", ex);
        }
      }

      try {
        close();
      } catch (SQLException ex) {
        LoggingUtils.logAll(LOG, "Unable to discard connection", ex);
      }
    }

    if (columns.size() == 0) {
      // Table has no primary key
      return null;
    }

    if (columns.size() > 1) {
      // The primary key is multi-column primary key. Warn the user.
      // TODO select the appropriate column instead of the first column based
      // on the datatype - giving preference to numerics over other types.
      LOG.warn("The table " + tableName + " "
          + "contains a multi-column primary key. Sqoop will default to "
          + "the column " + columns.get(0) + " only for this job.");
    }

    return columns.get(0);
  }

  @Override
  public String getInputBoundsQuery(String splitByCol, String sanitizedQuery) {
      /*
       * The default input bounds query generated by DataDrivenImportJob
       * is of the form:
       *  SELECT MIN(splitByCol), MAX(splitByCol) FROM (sanitizedQuery) AS t1
       *
       * This works for most databases but not Oracle since Oracle does not
       * allow the use of "AS" to project the subquery as a table. Instead the
       * correct format for use with Oracle is as follows:
       *  SELECT MIN(splitByCol), MAX(splitByCol) FROM (sanitizedQuery) t1
       */
      return "SELECT MIN(" + splitByCol + "), MAX(" + splitByCol + ") FROM ("
                   + sanitizedQuery + ") t1";
  }
}