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

import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
import java.math.BigDecimal;
import java.security.InvalidParameterException;
import java.sql.Connection;
import java.sql.Date;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;

import org.apache.hadoop.conf.Configuration;

/**
 * Contains the queries to get data dictionary information from Oracle database.
 */
public final class OraOopOracleQueries {

  private static final OraOopLog LOG = OraOopLogFactory
      .getLog(OraOopOracleQueries.class);

  private static Class<?> oracleConnectionClass;
  private static Class<?> oracleStatementClass;
  private static Class<?> oracleResultSetClass;
  private static Class<?> oracleTypesClass;
  private static Class<?> oracleDateClass;
  private static Method methSetLongAtName;
  private static Method methSetBigDecimalAtName;
  private static Method methSetStringAtName;
  private static Method methSetTimestampAtName;
  private static Method methSetBinaryDoubleAtName;
  private static Method methSetObjectAtName;
  private static Method methSetBinaryFloatAtName;
  private static Method methSetIntAtName;

  private static final Map<String, Integer> ORACLE_TYPES =
      new HashMap<String, Integer>();

  static {
    try {
      oracleStatementClass =
          Class.forName("oracle.jdbc.OraclePreparedStatement");
      methSetLongAtName =
          oracleStatementClass.getMethod("setLongAtName", String.class,
              long.class);
      methSetBigDecimalAtName =
          oracleStatementClass.getMethod("setBigDecimalAtName", String.class,
              BigDecimal.class);
      methSetStringAtName =
          oracleStatementClass.getMethod("setStringAtName", String.class,
              String.class);
      methSetTimestampAtName =
          oracleStatementClass.getMethod("setTimestampAtName", String.class,
              Timestamp.class);
      methSetBinaryDoubleAtName =
          oracleStatementClass.getMethod("setBinaryDoubleAtName", String.class,
              double.class);
      methSetObjectAtName =
          oracleStatementClass.getMethod("setObjectAtName", String.class,
              Object.class);
      methSetBinaryFloatAtName =
          oracleStatementClass.getMethod("setBinaryFloatAtName", String.class,
              float.class);
      methSetIntAtName =
          oracleStatementClass.getMethod("setIntAtName", String.class,
              int.class);

      oracleResultSetClass = Class.forName("oracle.jdbc.OracleResultSet");
      oracleDateClass = Class.forName("oracle.sql.DATE");
      oracleConnectionClass = Class.forName("oracle.jdbc.OracleConnection");
      oracleTypesClass = Class.forName("oracle.jdbc.OracleTypes");
    } catch (Exception e) {
      throw new RuntimeException(
          "Problem getting Oracle JDBC methods via reflection.", e);
    }
  }

  private OraOopOracleQueries() {
  }

  protected static void setJdbcFetchSize(Connection connection,
      org.apache.hadoop.conf.Configuration conf) {

    int fetchSize =
        conf.getInt(OraOopConstants.ORACLE_ROW_FETCH_SIZE,
            OraOopConstants.ORACLE_ROW_FETCH_SIZE_DEFAULT);
    try {
      Method methSetPrefetch =
          oracleConnectionClass.getMethod("setDefaultRowPrefetch", int.class);
      methSetPrefetch.invoke(connection, fetchSize);

      String msg =
          "The Oracle connection has had its default row fetch size set to : "
              + fetchSize;
      if (fetchSize == OraOopConstants.ORACLE_ROW_FETCH_SIZE_DEFAULT) {
        LOG.debug(msg);
      } else {
        LOG.info(msg);
      }
    } catch (Exception ex) {
      LOG.warn(
          String
              .format(
                  "Unable to configure the DefaultRowPrefetch of the "
                + "Oracle connection in %s.",
                  OraOopUtilities.getCurrentMethodName()), ex);
    }

  }

  public static void setConnectionTimeZone(Connection connection,
      Configuration conf) {
    String timeZoneString = conf.get("oracle.sessionTimeZone", "GMT");
    setConnectionTimeZone(connection, timeZoneString);
  }

  public static void setConnectionTimeZone(Connection connection,
      String timeZone) {
    TimeZone timeZoneObj = TimeZone.getTimeZone(timeZone);
    try {
      Method methSession =
          oracleConnectionClass.getMethod("setSessionTimeZone", String.class);
      Method methDefault =
          oracleConnectionClass.getMethod("setDefaultTimeZone", TimeZone.class);
      methSession.invoke(connection, timeZoneObj.getID());
      methDefault.invoke(connection, timeZoneObj);
      TimeZone.setDefault(timeZoneObj);
      LOG.info("Session Time Zone set to " + timeZoneObj.getID());
    } catch (Exception e) {
      LOG.error("Error setting time zone: " + e.getMessage());
    }
  }

  public static OracleTablePartitions getPartitions(Connection connection,
      OracleTable table) throws SQLException {

    OracleTablePartitions result = new OracleTablePartitions();

    PreparedStatement statement =
      connection
        .prepareStatement("with"
        + " partitions as"
        + "   (select table_owner, table_name, partition_name"
        +  "   from dba_tab_partitions"
        + "   where"
        + "   table_owner = ? and"
        + "   table_name = ?),"
        + " subpartitions as"
        + "  (select table_owner, table_name, partition_name, subpartition_name"
        + "  from dba_tab_subpartitions"
        + "  where"
        + "  table_owner = ? and"
        + "  table_name = ?)"
        + " select"
        + "   partitions.partition_name,"
        + "   subpartitions.subpartition_name"
        + " from partitions left outer join subpartitions on"
        + " (partitions.table_owner = subpartitions.table_owner"
        + " and partitions.table_name = subpartitions.table_name"
        + " and partitions.partition_name = subpartitions.partition_name)"
        + " order by partition_name, subpartition_name");

    statement.setString(1, table.getSchema());
    statement.setString(2, table.getName());
    statement.setString(3, table.getSchema());
    statement.setString(4, table.getName());

    ResultSet resultSet = statement.executeQuery();

    OracleTablePartition partition = null;
    while (resultSet.next()) {
      String partitionName = resultSet.getString("partition_name");
      String subPartitionName = resultSet.getString("subpartition_name");

      if (subPartitionName != null &&  !("".equals(subPartitionName))) {
        partition = new OracleTablePartition(subPartitionName, true);
        result.add(partition);
      } else {
        if (partition == null || partition.isSubPartition()
            || partition.getName() != partitionName) {
          partition = new OracleTablePartition(partitionName, false);
          result.add(partition);
        }
      }
    }

    resultSet.close();
    statement.close();

    return result;
  }

  public static int getOracleDataObjectNumber(Connection connection,
      OracleTable table) throws SQLException {

    PreparedStatement statement =
        connection.prepareStatement("SELECT data_object_id "
            + " FROM dba_objects" + " WHERE owner = ?" + " and object_name = ?"
            + " and object_type = ?");
    statement.setString(1, table.getSchema());
    statement.setString(2, table.getName());
    statement.setString(3, "TABLE");

    ResultSet resultSet = statement.executeQuery();

    resultSet.next();
    int result = resultSet.getInt("data_object_id");

    resultSet.close();
    statement.close();

    return result;
  }

  private static String getPartitionBindVars(List<String> partitionList) {
    String result = "";
    for (int i = 1; i <= partitionList.size(); i++) {
      result += (i > 1) ? "," : "";
      result += ":part" + i;
    }
    return result;
  }

  private static void bindPartitionBindVars(PreparedStatement statement,
      List<String> partitionList) throws SQLException {
    int i = 0;
    for (String partition : partitionList) {
      i++;
      OraOopOracleQueries.setStringAtName(statement, "part" + i, partition);
    }
  }

  public static List<OraOopOracleDataChunkPartition>
      getOracleDataChunksPartition(Connection connection, OracleTable table,
          List<String> partitionList) throws SQLException {
    List<OraOopOracleDataChunkPartition> result =
        new ArrayList<OraOopOracleDataChunkPartition>();
    String sql =
        "SELECT "
          + "  pl.partition_name, "
          + "  pl.is_subpartition, "
          + "  s.blocks "
          + "FROM "
          + "  (SELECT tp.table_owner, "
          + "    tp.table_name, "
          + "    NVL(tsp.subpartition_name,tp.partition_name) partition_name, "
          + "    nvl2(tsp.subpartition_name,1,0) is_subpartition "
          + "  FROM dba_tab_partitions tp, "
          + "    dba_tab_subpartitions tsp "
          + "  WHERE tp.table_owner     = :table_owner"
          + "  AND tp.table_name        = :table_name"
          + "  AND tsp.table_owner(+)   =tp.table_owner "
          + "  AND tsp.table_name(+)    =tp.table_name "
          + "  AND tsp.partition_name(+)=tp.partition_name ";

    if (partitionList != null && partitionList.size() > 0) {
      sql +=
          " AND tp.partition_name IN (" + getPartitionBindVars(partitionList)
              + ") ";
    }

    sql += "  ) pl, dba_tables t, dba_segments s "
        + "WHERE t.owner=pl.table_owner "
        + "AND t.table_name=pl.table_name "
        + "AND ( "
        + "     (t.iot_type='IOT' AND (s.owner,s.segment_name)= "
        + "                          (SELECT c.index_owner,c.index_name "
        + "                             FROM dba_constraints c "
        + "                            WHERE c.owner=pl.table_owner "
        + "                              AND c.table_name=pl.table_name "
        + "                              AND c.constraint_type='P')) "
        + "     OR (t.iot_type IS NULL "
        + "         AND s.owner=t.owner "
        + "         AND s.segment_name=t.table_name) "
        + "    ) "
        + "AND s.partition_name=pl.partition_name";

    PreparedStatement statement = connection.prepareStatement(sql);
    OraOopOracleQueries.setStringAtName(statement, "table_owner", table
        .getSchema());
    OraOopOracleQueries.setStringAtName(statement, "table_name", table
        .getName());

    if (partitionList != null && partitionList.size() > 0) {
      bindPartitionBindVars(statement, partitionList);
    }

    trace(String.format("%s SQL Query =\n%s", OraOopUtilities
        .getCurrentMethodName(), sql.replace(":table_owner", table.getSchema())
        .replace(":table_name", table.getName())));

    ResultSet resultSet = statement.executeQuery();

    while (resultSet.next()) {
      OraOopOracleDataChunkPartition dataChunk =
          new OraOopOracleDataChunkPartition(resultSet
              .getString("partition_name"), resultSet
              .getBoolean("is_subpartition"), resultSet.getLong("blocks"));
      result.add(dataChunk);
    }
    resultSet.close();
    statement.close();
    return result;
  }

  public static List<OraOopOracleDataChunkExtent> getOracleDataChunksExtent(
      Configuration conf, Connection connection, OracleTable table,
      List<String> partitionList, int numberOfChunksPerOracleDataFile)
      throws SQLException {

    List<OraOopOracleDataChunkExtent> result =
        new ArrayList<OraOopOracleDataChunkExtent>();

    String sql =
        "SELECT data_object_id, "
          + "file_id, "
          + "relative_fno, "
          + "file_batch, "
          + "MIN (start_block_id) start_block_id, "
          + "MAX (end_block_id) end_block_id, "
          + "SUM (blocks) blocks "
          + "FROM (SELECT o.data_object_id, "
          + "e.file_id, "
          + "e.relative_fno, "
          + "e.block_id start_block_id, "
          + "e.block_id + e.blocks - 1 end_block_id, "
          + "e.blocks, "
          + "CEIL ( "
          + "   SUM ( "
          + "      e.blocks) "
          + "   OVER (PARTITION BY o.data_object_id, e.file_id "
          + "         ORDER BY e.block_id ASC) "
          + "   / (SUM (e.blocks) "
          + "         OVER (PARTITION BY o.data_object_id, e.file_id) "
          + "      / :numchunks)) "
          + "   file_batch "
          + "FROM dba_extents e, dba_objects o, dba_tab_subpartitions tsp "
          + "WHERE     o.owner = :owner "
          + "AND o.object_name = :object_name "
          + "AND e.owner = :owner "
          + "AND e.segment_name = :object_name "
          + "AND o.owner = e.owner "
          + "AND o.object_name = e.segment_name "
          + "AND (o.subobject_name = e.partition_name "
          + "     OR (o.subobject_name IS NULL AND e.partition_name IS NULL)) "
          + "AND o.owner = tsp.table_owner(+) "
          + "AND o.object_name = tsp.table_name(+) "
          + "AND o.subobject_name = tsp.subpartition_name(+) ";

    if (partitionList != null && partitionList.size() > 0) {
      sql +=
          " AND case when o.object_type='TABLE SUBPARTITION' then "
          + "tsp.partition_name else o.subobject_name end IN ("
              + getPartitionBindVars(partitionList) + ") ";
    }

    sql +=
        ") " + "GROUP BY data_object_id, " + "         file_id, "
            + "         relative_fno, " + "         file_batch "
            + "ORDER BY data_object_id, " + "         file_id, "
            + "         relative_fno, " + "         file_batch";

    sql = conf.get(OraOopConstants.ORAOOP_ORACLE_DATA_CHUNKS_QUERY, sql);

    PreparedStatement statement = connection.prepareStatement(sql);
    OraOopOracleQueries.setIntAtName(statement, "numchunks",
        numberOfChunksPerOracleDataFile);
    OraOopOracleQueries.setStringAtName(statement, "owner", table.getSchema());
    OraOopOracleQueries.setStringAtName(statement, "object_name", table
        .getName());

    if (partitionList != null && partitionList.size() > 0) {
      bindPartitionBindVars(statement, partitionList);
    }

    trace(String.format("%s SQL Query =\n%s", OraOopUtilities
        .getCurrentMethodName(), sql.replace(":numchunks",
        Integer.toString(numberOfChunksPerOracleDataFile)).replace(":owner",
        table.getSchema()).replace(":object_name", table.getName())));

    ResultSet resultSet = statement.executeQuery();

    while (resultSet.next()) {
      int fileId = resultSet.getInt("relative_fno");
      int fileBatch = resultSet.getInt("file_batch");
      String dataChunkId =
          OraOopUtilities.generateDataChunkId(fileId, fileBatch);
      OraOopOracleDataChunkExtent dataChunk =
          new OraOopOracleDataChunkExtent(dataChunkId, resultSet
              .getInt("data_object_id"), resultSet.getInt("relative_fno"),
              resultSet.getLong("start_block_id"), resultSet
                  .getLong("end_block_id"));
      result.add(dataChunk);
    }

    resultSet.close();
    statement.close();

    return result;
  }

  private static void trace(String message) {

    LOG.debug(message);
  }

  public static String getOracleObjectType(Connection connection,
      OracleTable table) throws SQLException {

    PreparedStatement statement =
        connection.prepareStatement("SELECT object_type " + " FROM dba_objects"
            + " WHERE owner = ?" + " and object_name = ?");
    statement.setString(1, table.getSchema());
    statement.setString(2, table.getName());

    ResultSet resultSet = statement.executeQuery();

    String result = null;
    if (resultSet.next()) {
      result = resultSet.getString("object_type");
    }

    resultSet.close();
    statement.close();

    return result;
  }

  public static OracleVersion getOracleVersion(Connection connection)
      throws SQLException {

    String sql =
        "SELECT \n"
      + "  v.banner, \n"
      + "  rtrim(v.version)      full_version, \n"
      + "  rtrim(v.version_bit) version_bit, \n"
      + "  SUBSTR(v.version, 1, INSTR(v.version, '.', 1, 1)-1) major, \n"
      + "  SUBSTR(v.version, INSTR(v.version, '.', 1, 1) + 1, "
      + "  INSTR(v.version, '.', 1, 2) - INSTR(v.version, '.', 1, 1) - 1) "
      + "    minor, \n"
      + "  SUBSTR(v.version, INSTR(v.version, '.', 1, 2) + 1, "
      + "  INSTR(v.version, '.', 1, 3) - INSTR(v.version, '.', 1, 2) - 1) "
      + "    version, \n"
      + "  SUBSTR(v.version, INSTR(v.version, '.', 1, 3) + 1, "
      + "  INSTR(v.version, '.', 1, 4) - INSTR(v.version, '.', 1, 3) - 1) "
      + "    patch, \n"
      + "  DECODE(instr(v.banner, '64bit'), 0, 'False', 'True') isDb64bit, \n"
      + "  DECODE(instr(b.banner, 'HPUX'), 0, 'False', 'True') isHPUX \n"
      + "FROM (SELECT rownum row_num, \n"
      + "   banner,\n"
      + "   SUBSTR(SUBSTR(banner,INSTR(banner,'Release ')+8), 1) version_bit,\n"
      + "   SUBSTR(SUBSTR(banner,INSTR(banner,'Release ')+8), 1,\n"
      + "    INSTR(SUBSTR(banner,INSTR(banner,'Release ')+8),' ')) version\n"
      + "FROM v$version\n" + "  WHERE banner LIKE 'Oracle%'\n"
      + "     OR banner LIKE 'Personal Oracle%') v,\n" + "v$version b\n"
      + "  WHERE v.row_num = 1\n" + "  and b.banner like 'TNS%'\n";

    Statement statement = connection.createStatement();
    ResultSet resultSet = statement.executeQuery(sql);
    resultSet.next();
    OracleVersion result =
        new OracleVersion(resultSet.getInt("major"), resultSet.getInt("minor"),
            resultSet.getInt("version"), resultSet.getInt("patch"), resultSet
                .getString("banner"));

    resultSet.close();
    statement.close();

    return result;
  }

  public static List<OracleTable> getTables(Connection connection)
      throws SQLException {

    return getTables(connection, null, null, TableNameQueryType.Equals);
  }

  private enum GetTablesOptions {
    Owner, Table
  }

  private enum TableNameQueryType {
    Equals, Like
  }

  public static List<OracleTable>
      getTables(Connection connection, String owner) throws SQLException {

    return getTables(connection, owner, null, TableNameQueryType.Equals);
  }

  public static OracleTable getTable(Connection connection, String owner,
      String tableName) throws SQLException {

    List<OracleTable> tables =
        getTables(connection, owner, tableName, TableNameQueryType.Equals);
    if (tables.size() > 0) {
      return tables.get(0);
    }

    return null;
  }

  public static List<OracleTable> getTablesWithTableNameLike(
      Connection connection, String owner, String tableNameLike)
      throws SQLException {

    return getTables(connection, owner, tableNameLike, TableNameQueryType.Like);
  }

  private static List<OracleTable> getTables(Connection connection,
      String owner, String tableName, TableNameQueryType tableNameQueryType)
      throws SQLException {

    EnumSet<GetTablesOptions> options = EnumSet.noneOf(GetTablesOptions.class);

    if (owner != null && !owner.isEmpty()) {
      options.add(GetTablesOptions.Owner);
    }

    if (tableName != null && !tableName.isEmpty()) {
      options.add(GetTablesOptions.Table);
    }

    String sql =
        "SELECT owner, table_name " + " FROM dba_tables" + " %s %s %s %s "
            + " ORDER BY owner, table_name";

    String tableComparitor = null;
    switch (tableNameQueryType) {
      case Equals:
        tableComparitor = "=";
        break;
      case Like:
        tableComparitor = "LIKE";
        break;
      default:
        throw new RuntimeException("Operator not implemented.");
    }

    sql =
        String.format(sql, options.isEmpty() ? "" : "WHERE", options
            .contains(GetTablesOptions.Owner) ? "owner = ?" : "", options
            .containsAll(EnumSet.of(GetTablesOptions.Owner,
                GetTablesOptions.Table)) ? "AND" : "", options
            .contains(GetTablesOptions.Table) ? String.format(
            "table_name %s ?", tableComparitor) : "");

    PreparedStatement statement = connection.prepareStatement(sql);

    if (options.containsAll(EnumSet.of(GetTablesOptions.Owner,
        GetTablesOptions.Table))) {
      statement.setString(1, owner);
      statement.setString(2, tableName);
    } else {
      if (options.contains(GetTablesOptions.Owner)) {
        statement.setString(1, owner);
      } else if (options.contains(GetTablesOptions.Table)) {
        statement.setString(1, tableName);
      }
    }

    ResultSet resultSet = statement.executeQuery();

    ArrayList<OracleTable> result = new ArrayList<OracleTable>();
    while (resultSet.next()) {
      result.add(new OracleTable(resultSet.getString("owner"), resultSet
          .getString("table_name")));
    }

    resultSet.close();
    statement.close();

    return result;
  }

  public static List<String> getTableColumnNames(Connection connection,
      OracleTable table) throws SQLException {

    OracleTableColumns oracleTableColumns = getTableColumns(connection, table);
    List<String> result = new ArrayList<String>(oracleTableColumns.size());

    for (int idx = 0; idx < oracleTableColumns.size(); idx++) {
      result.add(oracleTableColumns.get(idx).getName());
    }

    return result;
  }

  public static List<String> getTableColumnNames(Connection connection,
      OracleTable table, boolean omitLobAndLongColumnsDuringImport,
      OraOopConstants.Sqoop.Tool sqoopTool, boolean onlyOraOopSupportedTypes,
      boolean omitOraOopPseudoColumns) throws SQLException {

    OracleTableColumns oracleTableColumns =
        getTableColumns(connection, table, omitLobAndLongColumnsDuringImport,
            sqoopTool, onlyOraOopSupportedTypes, omitOraOopPseudoColumns);

    List<String> result = new ArrayList<String>(oracleTableColumns.size());

    for (int idx = 0; idx < oracleTableColumns.size(); idx++) {
      result.add(oracleTableColumns.get(idx).getName());
    }

    return result;
  }

  private static OracleTableColumns getTableColumns(Connection connection,
      OracleTable table, boolean omitLobColumns, String dataTypesClause,
      HashSet<String> columnNamesToOmit) throws SQLException {

    String sql =
        "SELECT column_name, data_type " + " FROM dba_tab_columns"
            + " WHERE owner = ?" + " and table_name = ?" + " %s"
            + " ORDER BY column_id";

    sql =
        String.format(sql, dataTypesClause == null ? "" : " and "
            + dataTypesClause);

    LOG.debug(String.format("%s : sql = \n%s", OraOopUtilities
        .getCurrentMethodName(), sql));

    OracleTableColumns result = new OracleTableColumns();
    PreparedStatement statement = connection.prepareStatement(sql);
    statement.setString(1, getTableSchema(connection, table));
    statement.setString(2, table.getName());

    ResultSet resultSet = statement.executeQuery();

    while (resultSet.next()) {

      String columnName = resultSet.getString("column_name");

      if (columnNamesToOmit != null) {
        if (columnNamesToOmit.contains(columnName)) {
          continue;
        }
      }

      result.add(new OracleTableColumn(columnName, resultSet
          .getString("data_type")));
    }

    resultSet.close();
    statement.close();

    // Now get the actual JDBC data-types for these columns...
    StringBuilder columnList = new StringBuilder();
    for (int idx = 0; idx < result.size(); idx++) {
      if (idx > 0) {
        columnList.append(",");
      }
      columnList.append(OracleUtils.escapeIdentifier(result.get(idx).getName()));
    }
    sql =
        String.format("SELECT %s FROM %s WHERE 0=1", columnList.toString(),
            table.toString());
    Statement statementDesc = connection.createStatement();
    ResultSet resultSetDesc = statementDesc.executeQuery(sql);
    ResultSetMetaData metaData = resultSetDesc.getMetaData();
    for (int idx = 0; idx < metaData.getColumnCount(); idx++) {
      result.get(idx).setOracleType(metaData.getColumnType(idx + 1)); // <- JDBC
                                                                    // is
                                                                    // 1-based
    }
    resultSetDesc.close();
    statementDesc.close();

    return result;
  }

  public static OracleTableColumns getTableColumns(Connection connection,
      OracleTable table) throws SQLException {

    return getTableColumns(connection, table, false, null // <- dataTypesClause
        , null); // <-columnNamesToOmit
  }

  public static OracleTableColumns getTableColumns(Connection connection,
      OracleTable table, boolean omitLobAndLongColumnsDuringImport,
      OraOopConstants.Sqoop.Tool sqoopTool, boolean onlyOraOopSupportedTypes,
      boolean omitOraOopPseudoColumns) throws SQLException {

    String dataTypesClause = "";
    HashSet<String> columnNamesToOmit = null;

    if (onlyOraOopSupportedTypes) {

      switch (sqoopTool) {

        case UNKNOWN:
          throw new InvalidParameterException(
              "The sqoopTool parameter must not be \"UNKNOWN\".");

        case IMPORT:
          dataTypesClause =
              OraOopConstants.SUPPORTED_IMPORT_ORACLE_DATA_TYPES_CLAUSE;

          if (omitLobAndLongColumnsDuringImport) {
            LOG.info("LOB and LONG columns are being omitted from the Import.");
            dataTypesClause =
                " DATA_TYPE not in ('BLOB', 'CLOB', 'NCLOB', 'LONG') and "
                    + dataTypesClause;
          }
          break;

        case EXPORT:
          dataTypesClause =
              OraOopConstants.SUPPORTED_EXPORT_ORACLE_DATA_TYPES_CLAUSE;
          break;

        default:
          throw new InvalidParameterException("Sqoop Tool not implemented.");

      }
    }

    if (omitOraOopPseudoColumns) {

      switch (sqoopTool) {

        case EXPORT:
          if (columnNamesToOmit == null) {
            columnNamesToOmit = new HashSet<String>();
          }
          columnNamesToOmit.add(OraOopConstants.COLUMN_NAME_EXPORT_PARTITION);
          columnNamesToOmit
              .add(OraOopConstants.COLUMN_NAME_EXPORT_SUBPARTITION);
          columnNamesToOmit.add(OraOopConstants.COLUMN_NAME_EXPORT_MAPPER_ROW);
          break;
        default:
          // Only applicable for export.
          break;
      }
    }

    return getTableColumns(connection, table,
        omitLobAndLongColumnsDuringImport, dataTypesClause, columnNamesToOmit);
  }

  public static List<OracleActiveInstance> getOracleActiveInstances(
      Connection connection) throws SQLException {

    // Returns null if there are no rows in v$active_instances - which indicates
    // this Oracle database is not a RAC.
    ArrayList<OracleActiveInstance> result = null;

    Statement statement = connection.createStatement();
    ResultSet resultSet =
        statement.executeQuery("select inst_name from v$active_instances ");

    while (resultSet.next()) {
      String instName = resultSet.getString("inst_name");
      String[] nameFragments = instName.split(":");

      if (nameFragments.length != 2) {
        throw new SQLException(
            "Parsing Error: The inst_name column of v$active_instances does "
            + "not contain two values separated by a colon.");
      }

      String hostName = nameFragments[0].trim();
      String instanceName = nameFragments[1].trim();

      if (hostName.isEmpty()) {
        throw new SQLException(
            "Parsing Error: The inst_name column of v$active_instances does "
            + "not include a host name.");
      }

      if (instanceName.isEmpty()) {
        throw new SQLException(
            "Parsing Error: The inst_name column of v$active_instances does "
            + "not include an instance name.");
      }

      OracleActiveInstance instance = new OracleActiveInstance();
      instance.setHostName(hostName);
      instance.setInstanceName(instanceName);

      if (result == null) {
        result = new ArrayList<OracleActiveInstance>();
      }

      result.add(instance);
    }

    resultSet.close();
    statement.close();
    return result;
  }

  public static String getCurrentOracleInstanceName(Connection connection)
      throws SQLException {

    String result = "";

    Statement statement = connection.createStatement();
    ResultSet resultSet =
        statement.executeQuery("select instance_name from v$instance");

    if (resultSet.next()) {
      result = resultSet.getString("instance_name");
    }

    resultSet.close();
    statement.close();
    return result;
  }

  public static Object getSysDate(Connection connection) throws SQLException {
    Statement statement = connection.createStatement();
    ResultSet resultSet = statement.executeQuery("select sysdate from dual");

    resultSet.next();
    try {
      Method method = oracleResultSetClass.getMethod("getDATE", int.class);
      return method.invoke(resultSet, 1);
    } catch (Exception e) {
      if (e.getCause() instanceof SQLException) {
        throw (SQLException) e.getCause();
      } else {
        throw new RuntimeException("Could not get sysdate", e);
      }
    } finally {
      resultSet.close();
      statement.close();
    }
  }

  public static String oraDATEToString(Object date, String format) {
    try {
      Method dateMethod =
          oracleDateClass.getMethod("toText", String.class, String.class);
      return (String) dateMethod.invoke(date, format, null);
    } catch (Exception e) {
      throw new RuntimeException(String.format(
          "Unable to convert the oracle.sql.DATE value \"%s\" to text.", date
              .toString()), e);
    }
  }

  public static Object oraDATEFromString(String date, String format) {
    try {
      Method dateMethod =
          oracleDateClass.getMethod("fromText", String.class, String.class,
              String.class);
      return dateMethod.invoke(null, date, format, null);
    } catch (Exception e) {
      throw new RuntimeException(String
          .format(
              "Unable to convert the String value \"%s\" to oracle.sql.DATE.",
              date), e);
    }
  }

  public static Date oraDATEToDate(Object date) {
    try {
      Method dateMethod = oracleDateClass.getMethod("dateValue");
      return (Date) dateMethod.invoke(date);
    } catch (Exception e) {
      throw new RuntimeException("Could not get sysdate", e);
    }
  }

  public static String getSysTimeStamp(Connection connection)
      throws SQLException {

    Statement statement = connection.createStatement();
    ResultSet resultSet =
        statement.executeQuery("select systimestamp from dual");

    resultSet.next();

    try {
      Method method = oracleResultSetClass.getMethod("getTIMESTAMP", int.class);
      Object timestamp = method.invoke(resultSet, 1);
      return timestamp.toString();
    } catch (Exception e) {
      if (e.getCause() instanceof SQLException) {
        throw (SQLException) e.getCause();
      } else {
        throw new RuntimeException("Could not get sysdate", e);
      }
    } finally {
      resultSet.close();
      statement.close();
    }
  }

  public static boolean isTableAnIndexOrganizedTable(Connection connection,
      OracleTable table) throws SQLException {

    /*
     * http://ss64.com/orad/DBA_TABLES.html IOT_TYPE: If index-only table,then
     * IOT_TYPE is IOT or IOT_OVERFLOW or IOT_MAPPING else NULL
     */

    boolean result = false;

    PreparedStatement statement =
        connection.prepareStatement("select iot_type " + "from dba_tables "
            + "where owner = ? " + "and table_name = ?");
    statement.setString(1, table.getSchema());
    statement.setString(2, table.getName());
    ResultSet resultSet = statement.executeQuery();

    if (resultSet.next()) {
      String iotType = resultSet.getString("iot_type");
      result = iotType != null && !iotType.isEmpty();
    }

    resultSet.close();
    statement.close();

    return result;
  }

  public static void dropTable(Connection connection, OracleTable table)
      throws SQLException {

    String sql = String.format("DROP TABLE %s", table.toString());

    Statement statement = connection.createStatement();
    try {
      statement.execute(sql);
    } catch (SQLException ex) {
      if (ex.getErrorCode() != 942) { // ORA-00942: table or view does not exist
        throw ex;
      }
    }
    statement.close();
  }

  public static void
      exchangeSubpartition(Connection connection, OracleTable table,
          String subPartitionName, OracleTable subPartitionTable)
          throws SQLException {

    Statement statement = connection.createStatement();
    String sql =
        String.format("ALTER TABLE %s EXCHANGE SUBPARTITION %s WITH TABLE %s WITHOUT VALIDATION",
            table.toString(), subPartitionName, subPartitionTable.toString());
    statement.execute(sql);
    statement.close();
  }

  public static void createExportTableFromTemplate(Connection connection,
      OracleTable newTable, String tableStorageClause,
      OracleTable templateTable, boolean noLogging) throws SQLException {

    String sql =
        String.format("CREATE TABLE %s \n" + "%s %s \n" + "AS \n"
            + "(SELECT * FROM %s WHERE 0=1)", newTable.toString(),
            noLogging ? "NOLOGGING" : "", tableStorageClause, templateTable
                .toString());

    Statement statement = connection.createStatement();
    statement.execute(sql);
    statement.close();
  }

  private static Object oraDATEAddJulianDays(Object date, int julianDay,
      int julianSec) {
    try {
      Constructor<?> dateCon = oracleDateClass.getConstructor(byte[].class);
      Method dateBytes = oracleDateClass.getMethod("toBytes");
      Object result = dateCon.newInstance(dateBytes.invoke(date));
      Method dateAdd =
          oracleDateClass.getMethod("addJulianDays", int.class, int.class);
      result = dateAdd.invoke(result, julianDay, julianSec);
      return result;
    } catch (Exception e) {
      throw new RuntimeException("Could not add days to date.", e);
    }
  }

  public static void createExportTableFromTemplateWithPartitioning(
      Connection connection, OracleTable newTable, String tableStorageClause,
      OracleTable templateTable, boolean noLogging, String partitionName,
      Object jobDateTime, int numberOfMappers, String[] subPartitionNames)
      throws SQLException {

    String dateFormat = "yyyy-mm-dd hh24:mi:ss";

    Object partitionBound =
        OraOopOracleQueries.oraDATEAddJulianDays(jobDateTime, 0, 1);

    String partitionBoundStr =
        OraOopOracleQueries.oraDATEToString(partitionBound, dateFormat);

    StringBuilder subPartitions = new StringBuilder();
    for (int idx = 0; idx < numberOfMappers; idx++) {
      if (idx > 0) {
        subPartitions.append(",");
      }

      subPartitions.append(String.format(" SUBPARTITION %s VALUES (%d)",
          subPartitionNames[idx], idx));
    }

    String sql =
        String.format(
            "CREATE TABLE %s \n" + "%s %s \n" + "PARTITION BY RANGE (%s) \n"
                + "SUBPARTITION BY LIST (%s) \n" + "(PARTITION %s \n"
                + "VALUES LESS THAN (to_date('%s', '%s')) \n" + "( %s ) \n"
                + ") \n" + "AS \n"
                + "(SELECT t.*, sysdate %s, 0 %s, 0 %s FROM %s t \n"
                + "WHERE 0=1)", newTable.toString(), noLogging ? "NOLOGGING"
                : "", tableStorageClause,
            OraOopConstants.COLUMN_NAME_EXPORT_PARTITION,
            OraOopConstants.COLUMN_NAME_EXPORT_SUBPARTITION, partitionName,
            partitionBoundStr, dateFormat, subPartitions.toString(),
            OraOopConstants.COLUMN_NAME_EXPORT_PARTITION,
            OraOopConstants.COLUMN_NAME_EXPORT_SUBPARTITION,
            OraOopConstants.COLUMN_NAME_EXPORT_MAPPER_ROW, templateTable
                .toString());

    LOG.debug(String.format("SQL generated by %s:\n%s", OraOopUtilities
        .getCurrentMethodName(), sql));

    try {

      // Create the main export table...
      PreparedStatement preparedStatement = connection.prepareStatement(sql);
      preparedStatement.execute(sql);
      preparedStatement.close();
    } catch (SQLException ex) {
      LOG.error(String
          .format(
              "The error \"%s\" was encountered when executing the following "
              + "SQL statement:\n%s",
              ex.getMessage(), sql));
      throw ex;
    }
  }

  public static void createExportTableForMapper(Connection connection,
      OracleTable table, String tableStorageClause, OracleTable templateTable,
      boolean addOraOopPartitionColumns) throws SQLException {

    String sql = "";
    try {

      // Create the N tables to be used by the mappers...
      Statement statement = connection.createStatement();
      if (addOraOopPartitionColumns) {
        sql =
            String.format("CREATE TABLE %s \n" + "NOLOGGING %s \n" + "AS \n"
                + "(SELECT t.*, SYSDATE %s, 0 %s, 0 %s FROM %s t WHERE 0=1)",
                table.toString(), tableStorageClause,
                OraOopConstants.COLUMN_NAME_EXPORT_PARTITION,
                OraOopConstants.COLUMN_NAME_EXPORT_SUBPARTITION,
                OraOopConstants.COLUMN_NAME_EXPORT_MAPPER_ROW, templateTable
                    .toString());
      } else {
        sql =
            String.format("CREATE TABLE %s \n" + "NOLOGGING %s \n" + "AS \n"
                + "(SELECT * FROM %s WHERE 0=1)", table.toString(),
                tableStorageClause, templateTable.toString());
      }

      LOG.info(String.format("SQL generated by %s:\n%s", OraOopUtilities
          .getCurrentMethodName(), sql));

      statement.execute(sql);
      statement.close();
    } catch (SQLException ex) {
      LOG.error(String
          .format(
              "The error \"%s\" was encountered when executing the following "
              + "SQL statement:\n%s",
              ex.getMessage(), sql));
      throw ex;
    }
  }

  // public static void createExportTablesForMappers(Connection connection
  // ,String[] mapperTableNames
  // ,OracleTable templateTable
  // ,boolean addOraOopPartitionColumns)
  // throws SQLException {
  //
  // for(String tableName : mapperTableNames)
  // createExportTableForMapper(connection, tableName, templateTable,
  // addOraOopPartitionColumns);
  // }

  public static void createMoreExportTablePartitions(Connection connection,
      OracleTable table, String partitionName, Object jobDateTime,
      String[] subPartitionNames) throws SQLException {

    String dateFormat = "yyyy-mm-dd hh24:mi:ss";

    Object partitionBound =
        OraOopOracleQueries.oraDATEAddJulianDays(jobDateTime, 0, 1);
    String partitionBoundStr =
        OraOopOracleQueries.oraDATEToString(partitionBound, dateFormat);

    StringBuilder subPartitions = new StringBuilder();
    for (int idx = 0; idx < subPartitionNames.length; idx++) {
      if (idx > 0) {
        subPartitions.append(",");
      }

      subPartitions.append(String.format(" SUBPARTITION %s VALUES (%d)",
          subPartitionNames[idx], idx));
    }

    String sql =
        String.format("ALTER TABLE %s " + "ADD PARTITION %s "
            + "VALUES LESS THAN (to_date('%s', '%s'))" + "( %s ) ", table
            .toString(), partitionName, partitionBoundStr, dateFormat,
            subPartitions.toString());

    LOG.debug(String.format("SQL generated by %s:\n%s", OraOopUtilities
        .getCurrentMethodName(), sql));

    try {
      PreparedStatement preparedStatement = connection.prepareStatement(sql);
      preparedStatement.execute(sql);
      preparedStatement.close();

    } catch (SQLException ex) {
      LOG.error(String
          .format(
              "The error \"%s\" was encountered when executing the following "
              + "SQL statement:\n%s",
              ex.getMessage(), sql));
      throw ex;
    }
  }

  public static void mergeTable(Connection connection, OracleTable targetTable,
      OracleTable sourceTable, String[] mergeColumnNames,
      OracleTableColumns oracleTableColumns, Object oraOopSysDate,
      int oraOopMapperId, boolean parallelizationEnabled) throws SQLException {

    StringBuilder updateClause = new StringBuilder();
    StringBuilder insertClause = new StringBuilder();
    StringBuilder valuesClause = new StringBuilder();
    for (int idx = 0; idx < oracleTableColumns.size(); idx++) {
      OracleTableColumn oracleTableColumn = oracleTableColumns.get(idx);
      String columnName = oracleTableColumn.getName();

      if (insertClause.length() > 0) {
        insertClause.append(",");
      }
      insertClause.append(String.format("target.\"%s\"", columnName));

      if (valuesClause.length() > 0) {
        valuesClause.append(",");
      }
      valuesClause.append(String.format("source.\"%s\"", columnName));

      if (!OraOopUtilities.stringArrayContains(mergeColumnNames, columnName,
          true)) {

        // If we're performing a merge, then the table is not partitioned. (If
        // the table
        // was partitioned, we'd be deleting and then inserting rows.)
        if (!columnName
            .equalsIgnoreCase(OraOopConstants.COLUMN_NAME_EXPORT_PARTITION)
            && !columnName
            .equalsIgnoreCase(OraOopConstants.COLUMN_NAME_EXPORT_SUBPARTITION)
            && !columnName
            .equalsIgnoreCase(OraOopConstants.COLUMN_NAME_EXPORT_MAPPER_ROW)) {

          if (updateClause.length() > 0) {
            updateClause.append(",");
          }
          updateClause.append(String.format("target.\"%1$s\" = source.\"%1$s\"",
              columnName));

        }
      }
    }

    String sourceClause = valuesClause.toString();

    String sql =
        String.format("MERGE %7$s INTO %1$s target \n"
            + "USING (SELECT %8$s * FROM %2$s) source \n" + "  ON (%3$s) \n"
            + "WHEN MATCHED THEN \n" + "  UPDATE SET %4$s \n"
            + "WHEN NOT MATCHED THEN \n" + "  INSERT (%5$s) \n"
            + "  VALUES (%6$s)", targetTable.toString(),
            sourceTable.toString(),
            generateUpdateKeyColumnsWhereClauseFragment(mergeColumnNames,
                "target", "source"), updateClause.toString(), insertClause
                .toString(), sourceClause,
            parallelizationEnabled ? "/*+ append parallel(target) */" : "",
            parallelizationEnabled ? "/*+parallel*/" : "");

    LOG.info(String.format("Merge SQL statement:\n" + sql));

    Statement statement = connection.createStatement();
    ResultSet resultSet = statement.executeQuery(sql);
    resultSet.close();
    statement.close();
  }

  public static void updateTable(Connection connection,
      OracleTable targetTable, OracleTable sourceTable,
      String[] mergeColumnNames, OracleTableColumns oracleTableColumns,
      Object oraOopSysDate, int oraOopMapperId, boolean parallelizationEnabled)
      throws SQLException {

    StringBuilder targetColumnsClause = new StringBuilder();
    StringBuilder sourceColumnsClause = new StringBuilder();
    for (int idx = 0; idx < oracleTableColumns.size(); idx++) {
      OracleTableColumn oracleTableColumn = oracleTableColumns.get(idx);
      String columnName = oracleTableColumn.getName();

      if (targetColumnsClause.length() > 0) {
        targetColumnsClause.append(",");
      }
      targetColumnsClause.append(String.format("a.%s", OracleUtils.escapeIdentifier(columnName)));

      if (sourceColumnsClause.length() > 0) {
        sourceColumnsClause.append(",");
      }
      sourceColumnsClause.append(String.format("b.%s", OracleUtils.escapeIdentifier(columnName)));
    }

    String sourceClause = sourceColumnsClause.toString();

    sourceClause =
        sourceClause.replaceAll(OraOopConstants.COLUMN_NAME_EXPORT_PARTITION,
            String.format("to_date('%s', 'yyyy/mm/dd hh24:mi:ss')",
                OraOopOracleQueries.oraDATEToString(oraOopSysDate,
                    "yyyy/mm/dd hh24:mi:ss")));

    sourceClause =
        sourceClause.replaceAll(
            OraOopConstants.COLUMN_NAME_EXPORT_SUBPARTITION, Integer
                .toString(oraOopMapperId));

    String sql =
        String.format("UPDATE %5$s %1$s a \n" + "SET \n" + "(%2$s) \n"
            + "= (SELECT \n" + "%3$s \n" + "FROM %4$s b \n" + "WHERE %6$s) \n"
            + "WHERE EXISTS (SELECT null FROM %4$s c " + "WHERE %7$s)",
            targetTable.toString(), targetColumnsClause.toString(),
            sourceClause, sourceTable.toString(),
            parallelizationEnabled ? "/*+ parallel */" : "",
            generateUpdateKeyColumnsWhereClauseFragment(mergeColumnNames, "b",
                "a"), generateUpdateKeyColumnsWhereClauseFragment(
                mergeColumnNames, "c", "a"));

    LOG.info(String.format("Update SQL statement:\n" + sql));

    Statement statement = connection.createStatement();
    int rowsAffected = statement.executeUpdate(sql);

    LOG.info(String.format(
        "The number of rows affected by the update SQL was: %d", rowsAffected));

    statement.close();
  }

  /**
   * Whether new rows should be included in changes table or not.
   */
  public enum CreateExportChangesTableOptions {
    OnlyRowsThatDiffer, RowsThatDifferPlusNewRows
  }

  public static int createExportChangesTable(Connection connection,
      OracleTable tableToCreate, String tableToCreateStorageClause,
      OracleTable tableContainingUpdates, OracleTable tableToBeUpdated,
      String[] joinColumnNames, CreateExportChangesTableOptions options,
      boolean parallelizationEnabled) throws SQLException {

    List<String> columnNames =
        getTableColumnNames(connection, tableToBeUpdated
            , false // <- omitLobAndLongColumnsDuringImport
            , OraOopConstants.Sqoop.Tool.EXPORT
            , true // <- onlyOraOopSupportedTypes
            , false // <- omitOraOopPseudoColumns
        );

    StringBuilder columnClause = new StringBuilder(2 * columnNames.size());
    for (int idx = 0; idx < columnNames.size(); idx++) {
      if (idx > 0) {
        columnClause.append(",");
      }
      columnClause.append(String.format("a.%s", OracleUtils.escapeIdentifier(columnNames.get(idx))));
    }

    StringBuilder rowEqualityClause = new StringBuilder();
    for (int idx = 0; idx < columnNames.size(); idx++) {
      String columnName = columnNames.get(idx);

      // We need to omit the OraOop pseudo columns from the SQL statement that
      // compares the data in
      // the two tables we're interested in. Otherwise, EVERY row will be
      // considered to be changed,
      // since the values in the pseudo columns will differ. (i.e.
      // ORAOOP_EXPORT_SYSDATE will differ.)
      if (columnName
          .equalsIgnoreCase(OraOopConstants.COLUMN_NAME_EXPORT_PARTITION)
          || columnName
             .equalsIgnoreCase(OraOopConstants.COLUMN_NAME_EXPORT_SUBPARTITION)
          || columnName
             .equalsIgnoreCase(OraOopConstants.COLUMN_NAME_EXPORT_MAPPER_ROW)) {
        continue;
      }

      if (idx > 0) {
        rowEqualityClause.append("OR");
      }

      rowEqualityClause.append(String.format("(a.\"%1$s\" <> b.\"%1$s\" "
          + "OR (a.\"%1$s\" IS NULL AND b.\"%1$s\" IS NOT NULL) "
          + "OR (a.\"%1$s\" IS NOT NULL AND b.\"%1$s\" IS NULL))", columnName));
    }

    String sqlJoin = null;
    switch (options) {

      case OnlyRowsThatDiffer:
        sqlJoin = "";
        break;

      case RowsThatDifferPlusNewRows:
        sqlJoin = "(+)"; // <- An outer-join will cause the "new" rows to be
                         // included
        break;

      default:
        throw new RuntimeException(String.format(
            "Update %s to cater for the option \"%s\".", OraOopUtilities
                .getCurrentMethodName(), options.toString()));
    }

    String sql =
        String.format("CREATE TABLE %1$s \n" + "NOLOGGING %8$s \n" + "%7$s \n"
            + "AS \n " + "SELECT \n" + "%5$s \n" + "FROM %2$s a, %3$s b \n"
            + "WHERE (%4$s) \n" + "AND ( \n" + "%6$s \n" + ")", tableToCreate
            .toString(), tableContainingUpdates.toString(), tableToBeUpdated
            .toString(), generateUpdateKeyColumnsWhereClauseFragment(
            joinColumnNames, "a", "b", sqlJoin), columnClause.toString(),
            rowEqualityClause.toString(), parallelizationEnabled ? "PARALLEL"
                : "", tableToCreateStorageClause);

    LOG.info(String.format("The SQL to create the changes-table is:\n%s", sql));

    Statement statement = connection.createStatement();

    long start = System.nanoTime();
    statement.executeUpdate(sql);
    double timeInSec = (System.nanoTime() - start) / Math.pow(10, 9);
    LOG.info(String.format("Time spent creating change-table: %f sec.",
        timeInSec));

    String indexName = tableToCreate.toString().replaceAll("CHG", "IDX");
    start = System.nanoTime();
    statement.execute(String.format("CREATE INDEX %s ON %s (%s)", indexName,
        tableToCreate.toString(), OraOopUtilities
            .stringArrayToCSV(joinColumnNames)));
    timeInSec = (System.nanoTime() - start) / Math.pow(10, 9);
    LOG.info(String.format("Time spent creating change-table index: %f sec.",
        timeInSec));

    int changeTableRowCount = 0;

    ResultSet resultSet =
        statement.executeQuery(String.format("select count(*) from %s",
            tableToCreate.toString()));
    resultSet.next();
    changeTableRowCount = resultSet.getInt(1);
    LOG.info(String.format("The number of rows in the change-table is: %d",
        changeTableRowCount));

    statement.close();
    return changeTableRowCount;
  }

  public static void deleteRowsFromTable(Connection connection,
      OracleTable tableToDeleteRowsFrom,
      OracleTable tableContainingRowsToDelete, String[] joinColumnNames,
      boolean parallelizationEnabled) throws SQLException {

    String sql =
        String.format("DELETE %4$s FROM %1$s a \n" + "WHERE EXISTS ( \n"
            + "SELECT null FROM %3$s b WHERE \n" + "%2$s)",
            tableToDeleteRowsFrom.toString(),
            generateUpdateKeyColumnsWhereClauseFragment(joinColumnNames, "a",
                "b"), tableContainingRowsToDelete.toString(),
            parallelizationEnabled ? "/*+ parallel */" : "");

    LOG.info(String.format("The SQL to delete rows from a table:\n%s", sql));

    Statement statement = connection.createStatement();
    int rowsAffected = statement.executeUpdate(sql);

    LOG.info(String.format(
        "The number of rows affected by the delete SQL was: %d", rowsAffected));

    statement.close();
  }

  public static void insertRowsIntoExportTable(Connection connection,
      OracleTable tableToInsertRowsInto,
      OracleTable tableContainingRowsToInsert, Object oraOopSysDate,
      int oraOopMapperId, boolean parallelizationEnabled) throws SQLException {

    List<String> columnNames =
        getTableColumnNames(connection, tableToInsertRowsInto);

    StringBuilder columnClause =
        new StringBuilder(2 + (2 * columnNames.size()));
    for (int idx = 0; idx < columnNames.size(); idx++) {
      if (idx > 0) {
        columnClause.append(",");
      }
        columnClause.append(columnNames.get(idx));
    }

    String columnsClause = columnClause.toString();

    String sql =
        String.format("insert %4$s \n" + "into %1$s \n" + "select \n"
            + "%2$s \n" + "from %3$s", tableToInsertRowsInto.toString(),
            columnsClause, tableContainingRowsToInsert.toString(),
            parallelizationEnabled ? "/*+ append parallel */" : "");

    LOG.info(String.format(
        "The SQL to insert rows from one table into another:\n%s", sql));

    Statement statement = connection.createStatement();
    ResultSet resultSet = statement.executeQuery(sql);
    resultSet.close();
    statement.close();
  }

  public static boolean doesIndexOnColumnsExist(Connection connection,
      OracleTable oracleTable, String[] columnNames) throws SQLException {

    // Attempts to find an index on the table that *starts* with the N column
    // names passed.
    // These columns can be in any order.

    String columnNamesInClause =
        OraOopUtilities.stringArrayToCSV(columnNames, "'");

    String sql =
        String.format("SELECT b.index_name, \n"
            + "  sum(case when b.column_name in (%1$s) then 1 end) num_cols \n"
            + "FROM dba_indexes a, dba_ind_columns b \n" + "WHERE \n"
            + "a.owner = b.index_owner \n"
            + "AND a.index_name = b.index_name \n" + "AND b.table_owner = ? \n"
            + "AND b.table_name = ? \n" + "AND a.status = 'VALID' \n"
            + "AND b.column_position <= ? \n" + "GROUP BY b.index_name \n"
            + "HAVING sum(case when b.column_name in (%1$s) then 1 end) = ?",
            columnNamesInClause);

    PreparedStatement statement = connection.prepareStatement(sql);
    statement.setString(1, oracleTable.getSchema());
    statement.setString(2, oracleTable.getName());
    statement.setInt(3, columnNames.length);
    statement.setInt(4, columnNames.length);

    LOG.debug(String.format("SQL to find an index on the columns %s:\n%s",
        columnNamesInClause, sql));

    ResultSet resultSet = statement.executeQuery();

    boolean result = false;
    if (resultSet.next()) {
      LOG.debug(String
          .format(
              "The table %s has an index named %s starting with the column(s) "
              + "%s (in any order).",
              oracleTable.toString(), resultSet.getString("index_name"),
              columnNamesInClause));
      result = true;
    }

    resultSet.close();
    statement.close();

    return result;
  }

  private static String generateUpdateKeyColumnsWhereClauseFragment(
      String[] joinColumnNames, String prefix1, String prefix2) {

    return generateUpdateKeyColumnsWhereClauseFragment(joinColumnNames,
        prefix1, prefix2, "");
  }

  private static String generateUpdateKeyColumnsWhereClauseFragment(
      String[] joinColumnNames, String prefix1, String prefix2,
      String sqlJoinOperator) {

    StringBuilder result = new StringBuilder();
    for (int idx = 0; idx < joinColumnNames.length; idx++) {
      String joinColumnName = joinColumnNames[idx];
      if (idx > 0) {
        result.append(" AND ");
      }
      result.append(String.format("%1$s.%3$s = %2$s.%3$s %4$s", prefix1,
          prefix2, joinColumnName, sqlJoinOperator));
    }
    return result.toString();
  }

  public static String getCurrentSchema(Connection connection)
      throws SQLException {
    String sql = "SELECT SYS_CONTEXT('USERENV','CURRENT_SCHEMA') FROM DUAL";

    PreparedStatement statement = connection.prepareStatement(sql);

    ResultSet resultSet = statement.executeQuery();

    resultSet.next();
    String result = resultSet.getString(1);

    resultSet.close();
    statement.close();

    LOG.info("Current schema is: " + result);

    return result;
  }

  public static String getTableSchema(Connection connection, OracleTable table)
      throws SQLException {
    if (table.getSchema() == null || table.getSchema().isEmpty()) {
      return getCurrentSchema(connection);
    } else {
      return table.getSchema();
    }
  }

  public static long getCurrentScn(Connection connection) throws SQLException {
    String sql = "SELECT current_scn FROM v$database";
    PreparedStatement statement = connection.prepareStatement(sql);
    ResultSet resultSet = statement.executeQuery();

    resultSet.next();
    long result = resultSet.getLong(1);
    resultSet.close();
    statement.close();

    return result;
  }

  public static void setLongAtName(PreparedStatement statement,
      String bindName, long bindValue) throws SQLException {
    try {
      methSetLongAtName.invoke(statement, bindName, bindValue);
    } catch (Exception e) {
      if (e.getCause() instanceof SQLException) {
        throw (SQLException) e.getCause();
      } else {
        throw new RuntimeException("Could not set bind variable", e);
      }
    }
  }

  public static void setBigDecimalAtName(PreparedStatement statement,
      String bindName, BigDecimal bindValue) throws SQLException {
    try {
      methSetBigDecimalAtName.invoke(statement, bindName, bindValue);
    } catch (Exception e) {
      if (e.getCause() instanceof SQLException) {
        throw (SQLException) e.getCause();
      } else {
        throw new RuntimeException("Could not set bind variable", e);
      }
    }
  }

  public static void setStringAtName(PreparedStatement statement,
      String bindName, String bindValue) throws SQLException {
    try {
      methSetStringAtName.invoke(statement, bindName, bindValue);
    } catch (Exception e) {
      if (e.getCause() instanceof SQLException) {
        throw (SQLException) e.getCause();
      } else {
        throw new RuntimeException("Could not set bind variable", e);
      }
    }
  }

  public static void setTimestampAtName(PreparedStatement statement,
      String bindName, Timestamp bindValue) throws SQLException {
    try {
      methSetTimestampAtName.invoke(statement, bindName, bindValue);
    } catch (Exception e) {
      if (e.getCause() instanceof SQLException) {
        throw (SQLException) e.getCause();
      } else {
        throw new RuntimeException("Could not set bind variable", e);
      }
    }
  }

  public static void setBinaryDoubleAtName(PreparedStatement statement,
      String bindName, double bindValue) throws SQLException {
    try {
      methSetBinaryDoubleAtName.invoke(statement, bindName, bindValue);
    } catch (Exception e) {
      if (e.getCause() instanceof SQLException) {
        throw (SQLException) e.getCause();
      } else {
        throw new RuntimeException("Could not set bind variable", e);
      }
    }
  }

  public static void setObjectAtName(PreparedStatement statement,
      String bindName, Object bindValue) throws SQLException {
    try {
      methSetObjectAtName.invoke(statement, bindName, bindValue);
    } catch (Exception e) {
      if (e.getCause() instanceof SQLException) {
        throw (SQLException) e.getCause();
      } else {
        throw new RuntimeException("Could not set bind variable", e);
      }
    }
  }

  public static void setBinaryFloatAtName(PreparedStatement statement,
      String bindName, float bindValue) throws SQLException {
    try {
      methSetBinaryFloatAtName.invoke(statement, bindName, bindValue);
    } catch (Exception e) {
      if (e.getCause() instanceof SQLException) {
        throw (SQLException) e.getCause();
      } else {
        throw new RuntimeException("Could not set bind variable", e);
      }
    }
  }

  public static void setIntAtName(PreparedStatement statement, String bindName,
      int bindValue) throws SQLException {
    try {
      methSetIntAtName.invoke(statement, bindName, bindValue);
    } catch (Exception e) {
      if (e.getCause() instanceof SQLException) {
        throw (SQLException) e.getCause();
      } else {
        throw new RuntimeException("Could not set bind variable", e);
      }
    }
  }

  public static int getOracleType(String name) {
    Integer result = ORACLE_TYPES.get(name);
    if (result == null) {
      synchronized (ORACLE_TYPES) {
        try {
          result = oracleTypesClass.getField(name).getInt(null);
          ORACLE_TYPES.put(name, result);
        } catch (Exception e) {
          throw new RuntimeException("Invalid oracle type specified", e);
        }
      }
    }
    return result;
  }
}