/**
 * 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.sql.Connection;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;

import org.apache.sqoop.SqoopOptions;
import org.apache.sqoop.SqoopOptions.IncrementalMode;
import org.apache.sqoop.manager.ConnManager;
import org.apache.sqoop.manager.ManagerFactory;
import org.apache.sqoop.metastore.JobData;

import org.apache.sqoop.manager.OracleManager;
import org.apache.sqoop.manager.oracle.OraOopOutputFormatUpdate.UpdateMode;
import org.apache.sqoop.manager.oracle.OraOopUtilities.
           JdbcOracleThinConnectionParsingError;

import static org.apache.sqoop.mapreduce.JobBase.HADOOP_MAP_TASK_MAX_ATTEMTPS;

/**
 * OraOop manager - if OraOop cannot be used it should fall back to the default
 * OracleManager.
 *
 * To increase the amount of heap memory available to the mappers:
 *                                     -Dmapred.child.java.opts=-Xmx4000M
 * To prevent failed mapper tasks from being reattempted:
 *                                     -Dmapred.map.max.attempts=1
 */
public class OraOopManagerFactory extends ManagerFactory {

  @SuppressWarnings("unused")
  private static final OraOopLog ORAOOP_LOG = OraOopLogFactory
      .getLog("org.apache.sqoop.manager.oracle");
  private static final OraOopLog LOG = OraOopLogFactory
      .getLog(OraOopManagerFactory.class.getName());

  static {
    Configuration
        .addDefaultResource(OraOopConstants.ORAOOP_SITE_TEMPLATE_FILENAME);
    Configuration.addDefaultResource(OraOopConstants.ORAOOP_SITE_FILENAME);
  }

  @Override
  public ConnManager accept(JobData jobData) {

    OraOopUtilities.enableDebugLoggingIfRequired(jobData.getSqoopOptions()
        .getConf());

    LOG.debug(String.format("%s can be called by Sqoop!",
        OraOopConstants.ORAOOP_PRODUCT_NAME));

    ConnManager result = null;

    if (jobData != null) {

      SqoopOptions sqoopOptions = jobData.getSqoopOptions();

      String connectString = sqoopOptions.getConnectString();
      if (connectString != null
          && connectString.toLowerCase().trim().startsWith("jdbc:oracle")) {

        if (!isOraOopEnabled(sqoopOptions)) {
          return result;
        }

        OraOopConnManager oraOopConnManager = null;

        OraOopConstants.Sqoop.Tool jobType = getSqoopJobType(jobData);
        OraOopUtilities.rememberSqoopJobType(jobType, jobData.getSqoopOptions()
            .getConf());

        List<OraOopLogMessage> messagesToDisplayAfterWelcome =
            new ArrayList<OraOopLogMessage>();

        switch (jobType) {

          case IMPORT:
            if (isNumberOfImportMappersOkay(sqoopOptions)
                && !isSqoopImportIncremental(jobData)
                && isSqoopImportJobTableBased(sqoopOptions)) {

              // At this stage, the Sqoop import job appears to be one we're
              // interested in accepting. We now need to connect to
              // the Oracle database to perform more tests...

              oraOopConnManager = new OraOopConnManager(sqoopOptions);

              try {
                Connection connection = oraOopConnManager.getConnection();

                if (isSqoopTableAnOracleTable(connection, sqoopOptions
                    .getUsername(),
                    oraOopConnManager.getOracleTableContext())) {

                  // OraOop will not accept responsibility for an Index
                  // Organized Table (IOT)...
                  if (!isSqoopTableAnIndexOrganizedTable(connection,
                      oraOopConnManager.getOracleTableContext())) {
                    result = oraOopConnManager; // <- OraOop accepts
                                                // responsibility for this Sqoop
                                                // job!
                  } else {
                    OraOopConstants.OraOopOracleDataChunkMethod method =
                        OraOopUtilities.getOraOopOracleDataChunkMethod(
                            sqoopOptions.getConf());
                    if (method == OraOopConstants.
                                      OraOopOracleDataChunkMethod.PARTITION) {
                      result = oraOopConnManager;
                    } else {
                      LOG.info(String.format("%s will not process this Sqoop"
                         + " connection, as the Oracle table %s is an"
                         + " index-organized table. If the table is"
                         + " partitioned, set "
                         + OraOopConstants.ORAOOP_ORACLE_DATA_CHUNK_METHOD
                         + " to "
                         + OraOopConstants.OraOopOracleDataChunkMethod.PARTITION
                         + ".",
                         OraOopConstants.ORAOOP_PRODUCT_NAME,
                         oraOopConnManager.getOracleTableContext().toString()));
                    }
                  }
                }
              } catch (SQLException ex) {
                throw new RuntimeException(String.format(
                    "Unable to connect to the Oracle database at %s\n"
                        + "Error:%s", sqoopOptions.getConnectString(), ex
                        .getMessage()), ex);
              }
            }
            break;

          case EXPORT:
            if (isNumberOfExportMappersOkay(sqoopOptions)) {

              // At this stage, the Sqoop export job appears to be one we're
              // interested in accepting. We now need to connect to
              // the Oracle database to perform more tests...

              oraOopConnManager = new OraOopConnManager(sqoopOptions);

              Connection connection = null;
              try {
                connection = oraOopConnManager.getConnection();
              } catch (SQLException ex) {
                throw new RuntimeException(String.format(
                    "Unable to connect to the Oracle database at %s\n"
                        + "Error:%s", sqoopOptions.getConnectString(), ex
                        .getMessage()), ex);
              }

              try {

                createAnyRequiredOracleObjects(sqoopOptions, connection,
                    oraOopConnManager, messagesToDisplayAfterWelcome);

                if (isSqoopTableAnOracleTable(connection, sqoopOptions
                    .getUsername(),
                    oraOopConnManager.getOracleTableContext())) {

                  result = oraOopConnManager; // <- OraOop accepts
                                              // responsibility for this Sqoop
                                              // job!
                }

              } catch (SQLException ex) {
                LOG.error(OraOopUtilities.getFullExceptionMessage(ex));
              }
            }

            break;
          default:
            // OraOop doesn't know how to handle other types of jobs - so won't
            // accept them.
            break;
        }

        // If OraOop has accepted this Sqoop job...
        if (result != null) {

          showUserTheOraOopWelcomeMessage();

          for (OraOopLogMessage message : messagesToDisplayAfterWelcome) {
            message.log(LOG);
          }

          // By the time we get into getSplits(), the number of mappers
          // stored in the config can be either 4 or 1 - so it seems
          // a bit unreliable. We'll use our own property name to ensure
          // getSplits() gets the correct value...
          sqoopOptions.getConf().setInt(
              OraOopConstants.ORAOOP_DESIRED_NUMBER_OF_MAPPERS,
              sqoopOptions.getNumMappers());

          // Generate the "action" name that we'll assign to our Oracle sessions
          // so that the user knows which Oracle sessions belong to OraOop...
          sqoopOptions.getConf().set(
              OraOopConstants.ORACLE_SESSION_ACTION_NAME,
              getOracleSessionActionName(jobData));

          OraOopUtilities.appendJavaSecurityEgd(sqoopOptions.getConf());

          // Get the Oracle database version...
          try {
            OracleVersion oracleVersion =
                OraOopOracleQueries.getOracleVersion(result.getConnection());
            LOG.info(String.format("Oracle Database version: %s",
                oracleVersion.getBanner()));
            sqoopOptions.getConf().setInt(
                OraOopConstants.ORAOOP_ORACLE_DATABASE_VERSION_MAJOR,
                oracleVersion.getMajor());
            sqoopOptions.getConf().setInt(
                OraOopConstants.ORAOOP_ORACLE_DATABASE_VERSION_MINOR,
                oracleVersion.getMinor());
          } catch (SQLException ex) {
            LOG.error("Unable to obtain the Oracle database version.", ex);
          }

          try {
            if (sqoopOptions.getConf().getBoolean(
                OraOopConstants.ORAOOP_IMPORT_CONSISTENT_READ, false)) {
              long scn =
                  sqoopOptions.getConf().getLong(
                      OraOopConstants.ORAOOP_IMPORT_CONSISTENT_READ_SCN, 0);
              if (scn == 0) {
                scn = OraOopOracleQueries.getCurrentScn(result.getConnection());
              }
              sqoopOptions.getConf().setLong(
                  OraOopConstants.ORAOOP_IMPORT_CONSISTENT_READ_SCN, scn);
              LOG.info("Performing a consistent read using SCN: " + scn);
            }
          } catch (SQLException ex) {
            throw new RuntimeException("Unable to determine SCN of database.",
                ex);
          }

          // Generate the JDBC URLs to be used by each mapper...
          setMapperConnectionDetails(oraOopConnManager, jobData);

          // Show the user the Oracle command that can be used to kill this
          // OraOop
          // job via Oracle...
          showUserTheOracleCommandToKillOraOop(sqoopOptions);
        }

      }
    }

    return result;
  }

  private void setMapperConnectionDetails(OraOopConnManager oraOopConnManager,
      JobData jobData) {

    // Ensure we have a connection to the database...
    Connection connection = null;
    try {
      connection = oraOopConnManager.getConnection();
    } catch (SQLException ex) {
      throw new RuntimeException(String.format(
          "Unable to connect to the Oracle database at %s\n" + "Error:%s",
          jobData.getSqoopOptions().getConnectString(), ex.getMessage()));
    }

    // Query v$active_instances to get a list of all instances in the Oracle RAC
    // (assuming this *could* be a RAC)...
    List<OracleActiveInstance> activeInstances = null;
    try {
      activeInstances =
          OraOopOracleQueries.getOracleActiveInstances(connection);
    } catch (SQLException ex) {
      throw new RuntimeException(
          "An error was encountered when attempting to determine the "
        + "configuration of the Oracle RAC.",
        ex);
    }

    if (activeInstances == null) {
      LOG.info("This Oracle database is not a RAC.");
    } else {
      LOG.info("This Oracle database is a RAC.");
    }

    // Is dynamic JDBC URL generation disabled?...
    if (OraOopUtilities.oracleJdbcUrlGenerationDisabled(jobData
        .getSqoopOptions().getConf())) {
      LOG.info(String
          .format(
              "%s will not use dynamically generated JDBC URLs - this feature "
            + "has been disabled.",
            OraOopConstants.ORAOOP_PRODUCT_NAME));
      return;
    }

    boolean generateRacBasedJdbcUrls = false;

    // Decide whether this is a multi-instance RAC, and whether we need to do
    // anything more...
    if (activeInstances != null) {
      generateRacBasedJdbcUrls = true;

      if (activeInstances.size() < OraOopUtilities
          .getMinNumberOfOracleRacActiveInstancesForDynamicJdbcUrlUse(jobData
              .getSqoopOptions().getConf())) {
        LOG.info(String.format(
            "There are only %d active instances in the Oracle RAC. "
          + "%s will not bother utilizing dynamically generated JDBC URLs.",
          activeInstances.size(), OraOopConstants.ORAOOP_PRODUCT_NAME));
        generateRacBasedJdbcUrls = false;
      }
    }

    // E.g. jdbc:oracle:thin:@localhost.localdomain:1521:orcl
    String jdbcConnectStr = jobData.getSqoopOptions().getConnectString();

    // Parse the JDBC URL to obtain the port number for the TNS listener...
    String jdbcHost = "";
    int jdbcPort = 0;
    String jdbcSid = "";
    String jdbcService = "";
    String jdbcTnsName = "";
    try {

      OraOopJdbcUrl oraOopJdbcUrl = new OraOopJdbcUrl(jdbcConnectStr);
      OraOopUtilities.JdbcOracleThinConnection jdbcConnection =
          oraOopJdbcUrl.parseJdbcOracleThinConnectionString();
      jdbcHost = jdbcConnection.getHost();
      jdbcPort = jdbcConnection.getPort();
      jdbcSid = jdbcConnection.getSid();
      jdbcService = jdbcConnection.getService();
      jdbcTnsName = jdbcConnection.getTnsName();
    } catch (JdbcOracleThinConnectionParsingError ex) {
      LOG.info(String.format(
          "Unable to parse the JDBC connection URL \"%s\" as a connection "
        + "that uses the Oracle 'thin' JDBC driver.\n"
        + "This problem prevents %s from being able to dynamically generate "
        + "JDBC URLs that specify 'dedicated server connections' or spread "
        + "mapper sessions across multiple Oracle instances.\n"
        + "If the JDBC driver-type is 'OCI' (instead of 'thin'), then "
        + "load-balancing should be appropriately managed automatically.",
              jdbcConnectStr, OraOopConstants.ORAOOP_PRODUCT_NAME, ex));
      return;
    }

    if (generateRacBasedJdbcUrls) {

      // Retrieve the Oracle service name to use when connecting to the RAC...
      String oracleServiceName =
          OraOopUtilities.getOracleServiceName(jobData.getSqoopOptions()
              .getConf());

      // Generate JDBC URLs for each of the mappers...
      if (!oracleServiceName.isEmpty()) {
        if (!generateRacJdbcConnectionUrlsByServiceName(jdbcHost, jdbcPort,
            oracleServiceName, jobData)) {
          throw new RuntimeException(String.format(
              "Unable to connect to the Oracle database at %s "
                  + "via the service name \"%s\".", jobData.getSqoopOptions()
                  .getConnectString(), oracleServiceName));
        }
      } else {
        generateJdbcConnectionUrlsByActiveInstance(activeInstances, jdbcPort,
            jobData);
      }
    } else {
      generateJdbcConnectionUrlsByTnsnameSidOrService(jdbcHost, jdbcPort,
        jdbcSid, jdbcService, jdbcTnsName, jobData);
    }

  }

  private void generateJdbcConnectionUrlsByTnsnameSidOrService(String hostName,
    int port, String sid, String serviceName, String tnsName, JobData jobData) {

    String jdbcUrl = null;
    if (tnsName != null && !tnsName.isEmpty()) {
      jdbcUrl = OraOopUtilities.generateOracleTnsNameJdbcUrl(tnsName);
    } else if (sid != null && !sid.isEmpty()) {
      jdbcUrl = OraOopUtilities.generateOracleSidJdbcUrl(hostName, port, sid);
    } else {
      jdbcUrl =
          OraOopUtilities.generateOracleServiceNameJdbcUrl(hostName, port,
              serviceName);
    }

    // Now store these connection strings in such a way that each mapper knows
    // which one to use...
    for (int idxMapper = 0; idxMapper < jobData.getSqoopOptions()
        .getNumMappers(); idxMapper++) {
      storeJdbcUrlForMapper(idxMapper, jdbcUrl, jobData);
    }
  }

  private boolean generateRacJdbcConnectionUrlsByServiceName(String hostName,
      int port, String serviceName, JobData jobData) {

    boolean result = false;
    String jdbcUrl =
        OraOopUtilities.generateOracleServiceNameJdbcUrl(hostName, port,
            serviceName);

    if (testDynamicallyGeneratedOracleRacInstanceConnection(jdbcUrl, jobData
        .getSqoopOptions().getUsername(), jobData.getSqoopOptions()
        .getPassword(), jobData.getSqoopOptions().getConnectionParams()
        , false // <- ShowInstanceSysTimestamp
        , "" // <- instanceDescription
    )) {

      LOG.info(String.format(
          "%s will load-balance sessions across the Oracle RAC instances "
              + "by connecting each mapper to the Oracle Service \"%s\".",
          OraOopConstants.ORAOOP_PRODUCT_NAME, serviceName));

      // Now store these connection strings in such a way that each mapper knows
      // which one to use...
      for (int idxMapper = 0; idxMapper < jobData.getSqoopOptions()
          .getNumMappers(); idxMapper++) {
        storeJdbcUrlForMapper(idxMapper, jdbcUrl, jobData);
      }
      result = true;
    }
    return result;
  }

  private void
      generateJdbcConnectionUrlsByActiveInstance(
          List<OracleActiveInstance> activeInstances, int jdbcPort,
          JobData jobData) {

    // Generate JDBC URLs for each of the instances in the RAC...
    ArrayList<OraOopUtilities.JdbcOracleThinConnection>
        jdbcOracleActiveThinConnections =
            new ArrayList<OraOopUtilities.JdbcOracleThinConnection>(
                activeInstances.size());

    for (OracleActiveInstance activeInstance : activeInstances) {

      OraOopUtilities.JdbcOracleThinConnection
          jdbcActiveInstanceThinConnection =
              new OraOopUtilities.JdbcOracleThinConnection(
                  activeInstance.getHostName(),
                  jdbcPort, activeInstance.getInstanceName(), "", "");

      if (testDynamicallyGeneratedOracleRacInstanceConnection(
          jdbcActiveInstanceThinConnection.toString(), jobData
              .getSqoopOptions().getUsername(), jobData.getSqoopOptions()
              .getPassword(), jobData.getSqoopOptions().getConnectionParams(),
          true, activeInstance.getInstanceName())) {
        jdbcOracleActiveThinConnections.add(jdbcActiveInstanceThinConnection);
      }
    }

    // If there are multiple JDBC URLs that work okay for the RAC, then we'll
    // make use of them...
    if (jdbcOracleActiveThinConnections.size() < OraOopUtilities
        .getMinNumberOfOracleRacActiveInstancesForDynamicJdbcUrlUse(jobData
            .getSqoopOptions().getConf())) {
      LOG.info(String
          .format(
              "%s will not attempt to load-balance sessions across instances "
            + "of an Oracle RAC - as multiple JDBC URLs to the "
            + "Oracle RAC could not be dynamically generated.",
              OraOopConstants.ORAOOP_PRODUCT_NAME));
      return;
    } else {
      StringBuilder msg = new StringBuilder();
      msg.append(String
          .format(
              "%s will load-balance sessions across the following instances of"
            + "the Oracle RAC:\n",
              OraOopConstants.ORAOOP_PRODUCT_NAME));

      for (OraOopUtilities.JdbcOracleThinConnection thinConnection
               : jdbcOracleActiveThinConnections) {
        msg.append(String.format("\tInstance: %s \t URL: %s\n",
            thinConnection.getSid(), thinConnection.toString()));
      }
      LOG.info(msg.toString());
    }

    // Now store these connection strings in such a way that each mapper knows
    // which one to use...
    int racInstanceIdx = 0;
    OraOopUtilities.JdbcOracleThinConnection thinUrl;
    for (int idxMapper = 0; idxMapper < jobData.getSqoopOptions()
        .getNumMappers(); idxMapper++) {
      if (racInstanceIdx > jdbcOracleActiveThinConnections.size() - 1) {
        racInstanceIdx = 0;
      }
      thinUrl = jdbcOracleActiveThinConnections.get(racInstanceIdx);
      racInstanceIdx++;
      storeJdbcUrlForMapper(idxMapper, thinUrl.toString(), jobData);
    }
  }

  private boolean testDynamicallyGeneratedOracleRacInstanceConnection(
      String url, String userName, String password, Properties additionalProps,
      boolean showInstanceSysTimestamp, String instanceDescription) {

    boolean result = false;

    // Test the connection...
    try {
      Connection testConnection =
          OracleConnectionFactory.createOracleJdbcConnection(
              OraOopConstants.ORACLE_JDBC_DRIVER_CLASS, url, userName,
              password, additionalProps);

      // Show the system time on each instance...
      if (showInstanceSysTimestamp) {
        LOG.info(String.format("\tDatabase time on %s is %s",
            instanceDescription, OraOopOracleQueries
                .getSysTimeStamp(testConnection)));
      }

      testConnection.close();
      result = true;
    } catch (SQLException ex) {
      LOG.warn(
          String
              .format(
                  "The dynamically generated JDBC URL \"%s\" was unable to "
                  + "connect to an instance in the Oracle RAC.",
                  url), ex);
    }

    return result;
  }

  private void storeJdbcUrlForMapper(int mapperIdx, String jdbcUrl,
      JobData jobData) {

    // Now store these connection strings in such a way that each mapper knows
    // which one to use...
    Configuration conf = jobData.getSqoopOptions().getConf();
    String mapperJdbcUrlPropertyName =
        OraOopUtilities.getMapperJdbcUrlPropertyName(mapperIdx, conf);
    LOG.debug("Setting mapper url " + mapperJdbcUrlPropertyName + " = "
      + jdbcUrl);
    conf.set(mapperJdbcUrlPropertyName, jdbcUrl);
  }

  private boolean isOraOopEnabled(SqoopOptions sqoopOptions) {

    String oraOopDisabled =
        sqoopOptions.getConf().get(OraOopConstants.ORAOOP_DISABLED, "false")
            .toLowerCase();
    boolean oraOopIsDisabled =
        oraOopDisabled.equalsIgnoreCase("true")
            || oraOopDisabled.equalsIgnoreCase("yes")
            || oraOopDisabled.equalsIgnoreCase("y")
            || oraOopDisabled.equalsIgnoreCase("1");

    oraOopIsDisabled = oraOopIsDisabled || !sqoopOptions.isDirect();

    if (oraOopIsDisabled) {
      LOG.info(String.format("%s is disabled.",
          OraOopConstants.ORAOOP_PRODUCT_NAME));
    }

    return !oraOopIsDisabled;
  }

  private OraOopConstants.Sqoop.Tool getSqoopJobType(JobData jobData) {

    OraOopConstants.Sqoop.Tool result = OraOopConstants.Sqoop.Tool.UNKNOWN;

    String sqoopToolName = getSqoopToolName(jobData).toUpperCase().trim();
    try {
      result = OraOopConstants.Sqoop.Tool.valueOf(sqoopToolName);
    } catch (IllegalArgumentException ex) {
      LOG.debug(String.format(
          "The Sqoop tool name \"%s\" is not supported by OraOop",
          sqoopToolName), ex);
    }
    return result;
  }

  private boolean isNumberOfImportMappersOkay(SqoopOptions sqoopOptions) {

    // Check whether there are enough mappers for OraOop to be of benefit...
    boolean result =
      (sqoopOptions.getNumMappers() >= OraOopUtilities
          .getMinNumberOfImportMappersAcceptedByOraOop(sqoopOptions.getConf()));

    if (!result) {
      LOG.info(String.format(
        "%s will not process this sqoop connection, as an insufficient number "
            + "of mappers are being used.",
        OraOopConstants.ORAOOP_PRODUCT_NAME));
    }

    return result;
  }

  private boolean isNumberOfExportMappersOkay(SqoopOptions sqoopOptions) {

    // Check whether there are enough mappers for OraOop to be of benefit...
    boolean result =
        (sqoopOptions.getNumMappers() >= OraOopUtilities
          .getMinNumberOfExportMappersAcceptedByOraOop(sqoopOptions.getConf()));

    if (!result) {
      LOG.info(String.format(
        "%s will not process this sqoop connection, as an insufficient number "
            + "of mappers are being used.",
        OraOopConstants.ORAOOP_PRODUCT_NAME));
    }

    return result;
  }

  private boolean isSqoopImportJobTableBased(SqoopOptions sqoopOptions) {

    String tableName = sqoopOptions.getTableName();
    return (tableName != null && !tableName.isEmpty());
  }

  private boolean isSqoopTableAnOracleTable(Connection connection,
      String connectionUserName, OracleTable tableContext) {

    String oracleObjectType;

    try {

      // Find the table via dba_tables...
      OracleTable oracleTable =
          OraOopOracleQueries.getTable(connection, tableContext.getSchema(),
              tableContext.getName());
      if (oracleTable != null) {
        return true;
      }

      // If we could not find the table via dba_tables, then try and determine
      // what type of database object the
      // user was referring to. Perhaps they've specified the name of a view?...
      oracleObjectType =
          OraOopOracleQueries.getOracleObjectType(connection, tableContext);

      if (oracleObjectType == null) {
        LOG.info(String.format(
            "%1$s will not process this Sqoop connection, "
          + "as the Oracle user %2$s does not own a table named %3$s.\n"
          + "\tPlease prefix the table name with the owner.\n "
          + "\tNote: You may need to double-quote the owner and/or table name."
          + "\n\tE.g. sqoop ... --username %4$s --table %2$s.%3$s\n",
          OraOopConstants.ORAOOP_PRODUCT_NAME, tableContext.getSchema(),
          tableContext.getName(), connectionUserName));
        return false;
      }

    } catch (SQLException ex) {
      LOG.warn(String.format(
        "Unable to determine the Oracle-type of the object named %s owned by "
            + "%s.\nError:\n" + "%s", tableContext.getName(), tableContext
            .getSchema(), ex.getMessage()));

      // In the absence of conflicting information, let's assume the object is
      // actually a table...
      return true;
    }

    boolean result =
        oracleObjectType
            .equalsIgnoreCase(OraOopConstants.Oracle.OBJECT_TYPE_TABLE);

    if (!result) {
      LOG.info(String.format("%s will not process this sqoop connection, "
          + "as %s is not an Oracle table, it's a %s.",
          OraOopConstants.ORAOOP_PRODUCT_NAME, tableContext.toString(),
          oracleObjectType));
    }

    return result;
  }

  private boolean isSqoopTableAnIndexOrganizedTable(Connection connection,
      OracleTable tableContext) {

    boolean result = false;
    try {
      result =
          OraOopOracleQueries.isTableAnIndexOrganizedTable(connection,
              tableContext);
      return result;
    } catch (SQLException ex) {
      LOG.warn(String.format(
         "Unable to determine whether the Oracle table %s is an index-organized"
       + " table.\nError:\n" + "%s", tableContext.toString(), ex.getMessage()));
    }
    return result;
  }

  private String getSqoopToolName(JobData jobData) {

    return jobData.getSqoopTool().getToolName();
  }

  private String getOracleSessionActionName(JobData jobData) {

    // This method has been written assuming that:
    // (1) OraOop only processes Sqoop "import" and "export" jobs; and
    // (2) a table will be used during the import/export (not a query).

    if (getSqoopJobType(jobData) != OraOopConstants.Sqoop.Tool.IMPORT
        && getSqoopJobType(jobData) != OraOopConstants.Sqoop.Tool.EXPORT) {
      throw new UnsupportedOperationException(String.format(
          "%s needs to be updated to cope " + "with Sqoop jobs of type %s.",
          OraOopUtilities.getCurrentMethodName(), getSqoopToolName(jobData)));
    }

    String timeStr =
        (new SimpleDateFormat("yyyyMMddHHmmsszzz")).format(new Date());

    String result = String.format("%s %s", getSqoopToolName(jobData), timeStr);

    // NOTE: The "action" column of v$session is only a 32 character column.
    // Therefore we need to ensure that the string returned by this
    // method does not exceed 32 characters...
    if (result.length() > 32) {
      result = result.substring(0, 32).trim();
    }

    return result;
  }

  private boolean isSqoopImportIncremental(JobData jobData) {

    boolean result =
        jobData.getSqoopOptions().getIncrementalMode() != IncrementalMode.None;

    if (result) {
      LOG.info(String.format("%1$s will not process this sqoop connection, "
          + "as incremental mode is not supported by %1$s.",
          OraOopConstants.ORAOOP_PRODUCT_NAME));
    }

    return result;
  }

  private void showUserTheOraOopWelcomeMessage() {

    String msg1 =
        String.format("Using %s", OraOopConstants.ORAOOP_PRODUCT_NAME);

    int longestMessage = msg1.length();

    msg1 = OraOopUtilities.padRight(msg1, longestMessage);

    char[] asterisks = new char[longestMessage + 8];
    Arrays.fill(asterisks, '*');

    String msg =
        String.format("\n" + "%1$s\n" + "*** %2$s ***\n" + "%1$s", new String(
            asterisks), msg1);
    LOG.info(msg);
  }

  private void showUserTheOracleCommandToKillOraOop(SqoopOptions sqoopOptions) {

    int taskAttempts =
        sqoopOptions.getConf().getInt(HADOOP_MAP_TASK_MAX_ATTEMTPS, -1);

    // If killing the Oracle session if futile - because the job will be
    // reattempted, then don't
    // bother telling the user about this feature...
    if (taskAttempts != 1) {
      return;
    }

    String moduleName = OraOopConstants.ORACLE_SESSION_MODULE_NAME;
    String actionName =
        sqoopOptions.getConf().get(OraOopConstants.ORACLE_SESSION_ACTION_NAME);

    String msg = String.format(
        "\nNote: This %s job can be killed via Oracle by executing the "
      + "following statement:\n\tbegin\n"
      + "\t\tfor row in (select sid,serial# from v$session where module='%s' "
      + "and action='%s') loop\n"
      + "\t\t\texecute immediate 'alter system kill session ''' || row.sid || "
      + "',' || row.serial# || '''';\n"
      + "\t\tend loop;\n" + "\tend;",
      OraOopConstants.ORAOOP_PRODUCT_NAME, moduleName, actionName);
    LOG.info(msg);
  }

  private void createAnyRequiredOracleObjects(SqoopOptions sqoopOptions,
    Connection connection, OraOopConnManager oraOopConnManager,
    List<OraOopLogMessage> messagesToDisplayAfterWelcome) throws SQLException {

    Configuration conf = sqoopOptions.getConf();

    // The SYSDATE on the Oracle database will be used as the partition value
    // for this export job...
    Object sysDateTime = OraOopOracleQueries.getSysDate(connection);
    String sysDateStr =
      OraOopOracleQueries.oraDATEToString(sysDateTime, "yyyy-mm-dd hh24:mi:ss");
    OraOopUtilities.rememberOracleDateTime(conf,
        OraOopConstants.ORAOOP_JOB_SYSDATE, sysDateStr);

    checkForOldOraOopTemporaryOracleTables(connection, sysDateTime,
        OraOopOracleQueries.getCurrentSchema(connection),
        messagesToDisplayAfterWelcome);

    // Store the actual partition value, so the N mappers know what value to
    // insert...
    String partitionValue =
        OraOopOracleQueries.oraDATEToString(sysDateTime,
            OraOopConstants.ORAOOP_EXPORT_PARTITION_DATE_FORMAT);
    conf.set(OraOopConstants.ORAOOP_EXPORT_PARTITION_DATE_VALUE,
             partitionValue);

    // Generate the (22 character) partition name...
    String partitionName =
        OraOopUtilities
            .createExportTablePartitionNameFromOracleTimestamp(sysDateTime);

    int numMappers = sqoopOptions.getNumMappers();

    String exportTableTemplate =
        conf.get(OraOopConstants.ORAOOP_EXPORT_CREATE_TABLE_TEMPLATE, "");

    String user = sqoopOptions.getUsername();
    if (user == null) {
      user = OracleManager.getSessionUser(connection);
    }

    OracleTable templateTableContext =
        OraOopUtilities.decodeOracleTableName(user, exportTableTemplate);

    boolean noLoggingOnNewTable =
        conf.getBoolean(OraOopConstants.ORAOOP_EXPORT_CREATE_TABLE_NO_LOGGING,
            false);

    String updateKeyCol = sqoopOptions.getUpdateKeyCol();

    /* =========================== */
    /* VALIDATION OF INPUTS */
    /* =========================== */

    if (updateKeyCol == null || updateKeyCol.isEmpty()) {
      // We're performing an "insert" export, not an "update" export.

      // Check that the "oraoop.export.merge" property has not been specified,
      // as this would be
      // an invalid scenario...
      if (OraOopUtilities.getExportUpdateMode(conf) == UpdateMode.Merge) {
        throw new RuntimeException(String.format(
            "\n\nThe option \"%s\" can only be used if \"%s\" is "
                + "also used.\n", OraOopConstants.ORAOOP_EXPORT_MERGE,
            "--update-key"));
      }
    }

    if (OraOopUtilities
        .userWantsToCreatePartitionedExportTableFromTemplate(conf)
        || OraOopUtilities
            .userWantsToCreateNonPartitionedExportTableFromTemplate(conf)) {

      // OraOop will create the export table.

      if (oraOopConnManager.getOracleTableContext().getName().length()
              > OraOopConstants.Oracle.MAX_IDENTIFIER_LENGTH) {
        String msg =
            String.format(
                "The Oracle table name \"%s\" is longer than %d characters.\n"
              + "Oracle will not allow a table with this name to be created.",
            oraOopConnManager.getOracleTableContext().getName(),
            OraOopConstants.Oracle.MAX_IDENTIFIER_LENGTH);
        throw new RuntimeException(msg);
      }

      if (updateKeyCol != null && !updateKeyCol.isEmpty()) {

        // We're performing an "update" export, not an "insert" export.

        // Check whether the user is attempting an "update" (i.e. a non-merge).
        // If so, they're
        // asking to only UPDATE rows in a (about to be created) (empty) table
        // that contains no rows.
        // This will be a waste of time, as we'd be attempting to perform UPDATE
        // operations against a
        // table with no rows in it...
        UpdateMode updateMode = OraOopUtilities.getExportUpdateMode(conf);
        if (updateMode == UpdateMode.Update) {
          throw new RuntimeException(String.format(
              "\n\nCombining the option \"%s\" with the option \"%s=false\" is "
            + "nonsensical, as this would create an "
            + "empty table and then perform "
            + "a lot of work that results in a table containing no rows.\n",
              OraOopConstants.ORAOOP_EXPORT_CREATE_TABLE_TEMPLATE,
              OraOopConstants.ORAOOP_EXPORT_MERGE));
        }
      }

      // Check that the specified template table actually exists and is a
      // table...
      String templateTableObjectType =
          OraOopOracleQueries.getOracleObjectType(connection,
              templateTableContext);
      if (templateTableObjectType == null) {
        throw new RuntimeException(String.format(
            "The specified Oracle template table \"%s\" does not exist.",
            templateTableContext.toString()));
      }

      if (!templateTableObjectType
          .equalsIgnoreCase(OraOopConstants.Oracle.OBJECT_TYPE_TABLE)) {
        throw new RuntimeException(
          String.format(
              "The specified Oracle template table \"%s\" is not an "
            + "Oracle table, it's a %s.",
              templateTableContext.toString(), templateTableObjectType));
      }

      if (conf.getBoolean(OraOopConstants.ORAOOP_EXPORT_CREATE_TABLE_DROP,
          false)) {
        OraOopOracleQueries.dropTable(connection, oraOopConnManager
            .getOracleTableContext());
      }

      // Check that there is no existing database object with the same name of
      // the table to be created...
      String newTableObjectType =
          OraOopOracleQueries.getOracleObjectType(connection, oraOopConnManager
              .getOracleTableContext());
      if (newTableObjectType != null) {
        throw new RuntimeException(
          String.format(
              "%s cannot create a new Oracle table named %s as a \"%s\" "
            + "with this name already exists.",
            OraOopConstants.ORAOOP_PRODUCT_NAME, oraOopConnManager
                .getOracleTableContext().toString(), newTableObjectType));
      }
    } else {
      // The export table already exists.

      if (updateKeyCol != null && !updateKeyCol.isEmpty()) {

        // We're performing an "update" export, not an "insert" export.

        // Check that there exists an index on the export table on the
        // update-key column(s).
        // Without such an index, this export may perform like a real dog...
        String[] updateKeyColumns =
            OraOopUtilities.getExportUpdateKeyColumnNames(sqoopOptions);
        if (!OraOopOracleQueries.doesIndexOnColumnsExist(connection,
            oraOopConnManager.getOracleTableContext(), updateKeyColumns)) {
          String msg = String.format(
              "\n**************************************************************"
            + "***************************************************************"
            + "\n\tThe table %1$s does not have a valid index on "
            + "the column(s) %2$s.\n"
            + "\tAs a consequence, this export may take a long time to "
            + "complete.\n"
            + "\tIf performance is unacceptable, consider reattempting this "
            + "job after creating an index "
            + "on this table via the SQL...\n"
            + "\t\tcreate index <index_name> on %1$s(%2$s);\n"
            + "****************************************************************"
            + "*************************************************************",
                      oraOopConnManager.getOracleTableContext().toString(),
                      OraOopUtilities.stringArrayToCSV(updateKeyColumns));
          messagesToDisplayAfterWelcome.add(new OraOopLogMessage(
              OraOopConstants.Logging.Level.WARN, msg));
        }
      }
    }

    /* ================================= */
    /* CREATE A PARTITIONED TABLE */
    /* ================================= */
    if (OraOopUtilities
        .userWantsToCreatePartitionedExportTableFromTemplate(conf)) {

      // Create a new Oracle table using the specified template...

      String[] subPartitionNames =
          OraOopUtilities.generateExportTableSubPartitionNames(numMappers,
              sysDateTime, conf);
      // Create the export table from a template table...
      String tableStorageClause =
          OraOopUtilities.getExportTableStorageClause(conf);

      OraOopOracleQueries.createExportTableFromTemplateWithPartitioning(
          connection, oraOopConnManager.getOracleTableContext(),
          tableStorageClause, templateTableContext, noLoggingOnNewTable,
          partitionName, sysDateTime, sqoopOptions.getNumMappers(),
          subPartitionNames);
      return;
    }

    /* ===================================== */
    /* CREATE A NON-PARTITIONED TABLE */
    /* ===================================== */
    if (OraOopUtilities
        .userWantsToCreateNonPartitionedExportTableFromTemplate(conf)) {

      String tableStorageClause =
          OraOopUtilities.getExportTableStorageClause(conf);

      OraOopOracleQueries.createExportTableFromTemplate(connection,
          oraOopConnManager.getOracleTableContext(), tableStorageClause,
          templateTableContext, noLoggingOnNewTable);
      return;
    }

    /* ===================================================== */
    /* ADD ADDITIONAL PARTITIONS TO AN EXISTING TABLE */
    /* ===================================================== */

    // If the export table is partitioned, and the partitions were created by
    // OraOop, then we need
    // create additional partitions...

    OracleTablePartitions tablePartitions =
        OraOopOracleQueries.getPartitions(connection, oraOopConnManager
            .getOracleTableContext());
    // Find any partition name starting with "ORAOOP_"...
    OracleTablePartition oraOopPartition =
        tablePartitions.findPartitionByRegEx("^"
            + OraOopConstants.EXPORT_TABLE_PARTITION_NAME_PREFIX);

    if (tablePartitions.size() > 0 && oraOopPartition == null) {

      for (int idx = 0; idx < tablePartitions.size(); idx++) {
        messagesToDisplayAfterWelcome.add(new OraOopLogMessage(
            OraOopConstants.Logging.Level.INFO, String.format(
                "The Oracle table %s has a partition named \"%s\".",
                oraOopConnManager.getOracleTableContext().toString(),
                tablePartitions.get(idx).getName())));
      }

      messagesToDisplayAfterWelcome.add(new OraOopLogMessage(
          OraOopConstants.Logging.Level.WARN, String.format(
              "The Oracle table %s is partitioned.\n"
                  + "These partitions were not created by %s.",
              oraOopConnManager.getOracleTableContext().toString(),
              OraOopConstants.ORAOOP_PRODUCT_NAME)));
    }

    if (oraOopPartition != null) {

      // Indicate in the configuration what's happening...
      conf.setBoolean(OraOopConstants.EXPORT_TABLE_HAS_ORAOOP_PARTITIONS, true);

      messagesToDisplayAfterWelcome
          .add(new OraOopLogMessage(
              OraOopConstants.Logging.Level.INFO,
              String
                  .format(
                      "The Oracle table %s is partitioned.\n"
                          + "These partitions were created by %s, so "
                          + "additional partitions will now be created.\n"
                          + "The name of the new partition will be \"%s\".",
                      oraOopConnManager.getOracleTableContext().toString(),
                      OraOopConstants.ORAOOP_PRODUCT_NAME, partitionName)));

      String[] subPartitionNames =
          OraOopUtilities.generateExportTableSubPartitionNames(numMappers,
              sysDateTime, conf);

      // Add another partition (and N subpartitions) to this existing,
      // partitioned export table...
      OraOopOracleQueries.createMoreExportTablePartitions(connection,
          oraOopConnManager.getOracleTableContext(), partitionName,
          sysDateTime, subPartitionNames);

      return;
    }
  }

  private void checkForOldOraOopTemporaryOracleTables(Connection connection,
      Object sysDateTime, String schema,
      List<OraOopLogMessage> messagesToDisplayAfterWelcome) {

    try {

      StringBuilder message = new StringBuilder();
      message
        .append(String.format(
          "The following tables appear to be old temporary tables created by "
        + "%s that have not been deleted.\n"
        + "They are probably left over from jobs that encountered an error and "
        + "could not clean up after themselves.\n"
        + "You might want to drop these Oracle tables in order to reclaim "
        + "Oracle storage space:\n", OraOopConstants.ORAOOP_PRODUCT_NAME));
      boolean showMessage = false;

      String generatedTableName =
          OraOopUtilities.generateExportTableMapperTableName(0, sysDateTime,
              schema).getName();
      generatedTableName = generatedTableName.replaceAll("[0-9]", "%");
      generatedTableName =
          OraOopUtilities.replaceAll(generatedTableName, "%%", "%");
      Date sysDate = OraOopOracleQueries.oraDATEToDate(sysDateTime);

      List<OracleTable> tables =
          OraOopOracleQueries.getTablesWithTableNameLike(connection, schema,
              generatedTableName);

      for (OracleTable oracleTable : tables) {
        OraOopUtilities.DecodedExportMapperTableName tableName =
            OraOopUtilities.decodeExportTableMapperTableName(oracleTable);
        if (tableName != null) {
          Date tableDate =
              OraOopOracleQueries.oraDATEToDate(tableName.getTableDateTime());
          double daysApart =
              (sysDate.getTime() - tableDate.getTime()) / (1000 * 60 * 60 * 24);
          if (daysApart > 1.0) {
            showMessage = true;
            message.append(String.format("\t%s\n", oracleTable.toString()));
          }
        }
      }

      if (showMessage) {
        messagesToDisplayAfterWelcome.add(new OraOopLogMessage(
            OraOopConstants.Logging.Level.INFO, message.toString()));
      }
    } catch (Exception ex) {
      messagesToDisplayAfterWelcome.add(new OraOopLogMessage(
          OraOopConstants.Logging.Level.WARN, String.format(
              "%s was unable to check for the existance of old "
                  + "temporary Oracle tables.\n" + "Error:\n%s",
              OraOopConstants.ORAOOP_PRODUCT_NAME, ex.toString())));
    }
  }
}