/*
 * 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.ambari.server.orm;

import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.math.BigDecimal;
import java.nio.charset.Charset;
import java.sql.Blob;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.configuration.Configuration.DatabaseType;
import org.apache.ambari.server.orm.helpers.ScriptRunner;
import org.apache.ambari.server.orm.helpers.dbms.DbmsHelper;
import org.apache.ambari.server.orm.helpers.dbms.DerbyHelper;
import org.apache.ambari.server.orm.helpers.dbms.GenericDbmsHelper;
import org.apache.ambari.server.orm.helpers.dbms.H2Helper;
import org.apache.ambari.server.orm.helpers.dbms.MySqlHelper;
import org.apache.ambari.server.orm.helpers.dbms.OracleHelper;
import org.apache.ambari.server.orm.helpers.dbms.PostgresHelper;
import org.apache.ambari.server.utils.CustomStringUtils;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
import org.eclipse.persistence.internal.helper.DBPlatformHelper;
import org.eclipse.persistence.internal.sessions.DatabaseSessionImpl;
import org.eclipse.persistence.logging.AbstractSessionLog;
import org.eclipse.persistence.logging.SessionLogEntry;
import org.eclipse.persistence.platform.database.DatabasePlatform;
import org.eclipse.persistence.platform.database.DerbyPlatform;
import org.eclipse.persistence.platform.database.H2Platform;
import org.eclipse.persistence.platform.database.MySQLPlatform;
import org.eclipse.persistence.platform.database.OraclePlatform;
import org.eclipse.persistence.platform.database.PostgreSQLPlatform;
import org.eclipse.persistence.sessions.DatabaseLogin;
import org.eclipse.persistence.sessions.DatabaseSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.support.JdbcUtils;

import com.google.inject.Inject;
import com.google.inject.Singleton;

@Singleton
public class DBAccessorImpl implements DBAccessor {
  private static final Logger LOG = LoggerFactory.getLogger(DBAccessorImpl.class);
  public static final String USER = "user";
  public static final String PASSWORD = "password";
  public static final String NULL_CATALOG_MEANS_CURRENT = "nullCatalogMeansCurrent";
  public static final String TRUE = "true";
  public static final int SUPPORT_CONNECTOR_VERSION = 5;
  private final DatabasePlatform databasePlatform;
  private final Connection connection;
  private final DbmsHelper dbmsHelper;
  private Configuration configuration;
  private DatabaseMetaData databaseMetaData;
  private static final String dbURLPatternString = "jdbc:(.*?):.*";
  private DbType dbType;
  private final String dbSchema;

  @Inject
  public DBAccessorImpl(Configuration configuration) {
    this.configuration = configuration;

    try {
      Class.forName(configuration.getDatabaseDriver());

      connection = getNewConnection();

      connection.setAutoCommit(true); //enable autocommit

      //TODO create own mapping and platform classes for supported databases
      String vendorName = connection.getMetaData().getDatabaseProductName();
      String majorVersion = Integer.toString(connection.getMetaData().getDatabaseMajorVersion());
      String minorVersion = Integer.toString(connection.getMetaData().getDatabaseMinorVersion());

      String dbPlatform = DBPlatformHelper.getDBPlatform(vendorName, majorVersion, minorVersion, new AbstractSessionLog() {
        @Override
        public void log(SessionLogEntry sessionLogEntry) {
          LOG.debug(sessionLogEntry.getMessage());
        }
      });
      databasePlatform = (DatabasePlatform) Class.forName(dbPlatform).newInstance();
      dbmsHelper = loadHelper(databasePlatform);
      dbSchema = convertObjectName(configuration.getDatabaseSchema());
    } catch (Exception e) {
      String message = "";
      if (e instanceof ClassNotFoundException) {
        message = "If you are using a non-default database for Ambari and a custom JDBC driver jar, you need to set property \"server.jdbc.driver.path={path/to/custom_jdbc_driver}\" " +
                "in ambari.properties config file, to include it in ambari-server classpath.";
      } else {
        message = "Error while creating database accessor ";
      }
      LOG.error(message, e);
      throw new RuntimeException(message,e);
    }
  }

  protected DbmsHelper loadHelper(DatabasePlatform databasePlatform) {
    if (databasePlatform instanceof OraclePlatform) {
      dbType = DbType.ORACLE;
      return new OracleHelper(databasePlatform);
    } else if (databasePlatform instanceof MySQLPlatform) {
      dbType = DbType.MYSQL;
      return new MySqlHelper(databasePlatform);
    } else if (databasePlatform instanceof PostgreSQLPlatform) {
      dbType = DbType.POSTGRES;
      return new PostgresHelper(databasePlatform);
    } else if (databasePlatform instanceof DerbyPlatform) {
      dbType = DbType.DERBY;
      return new DerbyHelper(databasePlatform);
    } else if (databasePlatform instanceof H2Platform) {
      dbType = DbType.H2;
      return new H2Helper(databasePlatform);
    } else {
      dbType = DbType.UNKNOWN;
      return new GenericDbmsHelper(databasePlatform);
    }
  }

  /**
   * Map SQL datatype to Java Class type
   *
   * @param type  SQL datatype
   * @return Java class or null if no mapping found
   */
  private static Class<?> fromSqlTypeToClass(int type) {
    switch (type) {
      case Types.VARCHAR:
      case Types.CHAR:
      case Types.LONGVARCHAR:
        return String.class;
      case Types.NUMERIC:
      case Types.DECIMAL:
        return BigDecimal.class;
      case Types.BIT:
        return Boolean.class;
      case Types.TINYINT:
        return Byte.class;
      case Types.SMALLINT:
        return Short.class;
      case Types.INTEGER:
        return Integer.class;
      case Types.BIGINT:
        return Long.class;
      case Types.FLOAT:
      case Types.REAL:
        return Float.class;
      case Types.DOUBLE:
        return Double.class;
      case Types.BINARY:
      case Types.VARBINARY:
      case Types.LONGVARBINARY:
        return Byte[].class;
      case Types.DATE:
        return java.sql.Date.class;
      case Types.TIME:
        return java.sql.Timestamp.class;
      default:
        return null;
    }
  }

  @Override
  public Connection getConnection() {
    return connection;
  }

  @Override
  public Connection getNewConnection() {
    try {
      Properties properties = new Properties();
      properties.setProperty(USER, configuration.getDatabaseUser());
      properties.setProperty(PASSWORD, configuration.getDatabasePassword());
      if (configuration.getDatabaseUrl().contains("mysql")
          && DriverManager.getDriver(configuration.getDatabaseUrl()).getMajorVersion() > SUPPORT_CONNECTOR_VERSION) {
        properties.setProperty(NULL_CATALOG_MEANS_CURRENT, TRUE);// jdbc mysql connector 8.x turn on legacy behaviour
      }
      return DriverManager.getConnection(configuration.getDatabaseUrl(),properties);
    } catch (SQLException e) {
      throw new RuntimeException("Unable to connect to database", e);
    }
  }

  @Override
  public String quoteObjectName(String name) {
    return dbmsHelper.quoteObjectName(name);
  }

  @Override
  public void createTable(String tableName, List<DBColumnInfo> columnInfo,
      String... primaryKeyColumns) throws SQLException {
    // do nothing if the table already exists
    if (tableExists(tableName)) {
      return;
    }

    // guard against null PKs
    primaryKeyColumns = ArrayUtils.nullToEmpty(primaryKeyColumns);

    String query = dbmsHelper.getCreateTableStatement(tableName, columnInfo,
        Arrays.asList(primaryKeyColumns));

    executeQuery(query);
  }

  protected DatabaseMetaData getDatabaseMetaData() throws SQLException {
    if (databaseMetaData == null) {
      databaseMetaData = connection.getMetaData();
    }

    return databaseMetaData;
  }

  private String convertObjectName(String objectName) throws SQLException {
    //tolerate null names for proper usage in filters
    if (objectName == null) {
      return null;
    }
    DatabaseMetaData metaData = getDatabaseMetaData();
    if (metaData.storesLowerCaseIdentifiers()) {
      return objectName.toLowerCase();
    } else if (metaData.storesUpperCaseIdentifiers()) {
      return objectName.toUpperCase();
    }

    return objectName;
  }

  /**
   * Setting arguments for prepared statement
   *
   * @param preparedStatement {@link PreparedStatement} object
   * @param arguments array of arguments
   *
   * @throws SQLException
   */
  private void setArgumentsForPreparedStatement(PreparedStatement preparedStatement, Object[] arguments) throws SQLException{
    for (int i = 0; i < arguments.length; i++) {
      if (arguments[i] instanceof byte[]) {
        byte[] binaryData = (byte[]) arguments[i];

        // JDBC drivers supports only this function signature
        preparedStatement.setBinaryStream(i+1, new ByteArrayInputStream(binaryData), binaryData.length);
      } else {
        preparedStatement.setObject(i+1, arguments[i]);
      }
    }
  }

  @Override
  public boolean tableExists(String tableName) throws SQLException {
    boolean result = false;
    DatabaseMetaData metaData = getDatabaseMetaData();

    ResultSet res = metaData.getTables(null, dbSchema, convertObjectName(tableName), new String[]{"TABLE"});

    if (res != null) {
      try {
        if (res.next()) {
          result = res.getString("TABLE_NAME") != null && res.getString("TABLE_NAME").equalsIgnoreCase(tableName);
        }
        if (res.next()) {
          throw new IllegalStateException(
                  String.format("Request for table [%s] existing returned more than one results",
                          tableName));
        }
      } finally {
        res.close();
      }
    }

    return result;
  }

  @Override
  public DbType getDbType() {
    return dbType;
  }

  @Override
  public String getDbSchema() {
    return dbSchema;
  }

  @Override
  public boolean tableHasData(String tableName) throws SQLException {
    String query = "SELECT count(*) from " + tableName;
    Statement statement = getConnection().createStatement();
    boolean retVal = false;
    ResultSet rs = null;
    try {
      rs = statement.executeQuery(query);
      if (rs != null) {
        if (rs.next()) {
          return rs.getInt(1) > 0;
        }
      }
    } catch (Exception e) {
      LOG.error("Unable to check if table " + tableName + " has any data. Exception: " + e.getMessage());
    } finally {
      if (statement != null) {
        statement.close();
      }
      if (rs != null) {
        rs.close();
      }
    }
    return retVal;
  }

  @Override
  public boolean tableHasColumn(String tableName, String columnName) throws SQLException {
    boolean result = false;
    DatabaseMetaData metaData = getDatabaseMetaData();

    ResultSet rs = metaData.getColumns(null, dbSchema, convertObjectName(tableName), convertObjectName(columnName));

    if (rs != null) {
      try {
        if (rs.next()) {
          result = rs.getString("COLUMN_NAME") != null && rs.getString("COLUMN_NAME").equalsIgnoreCase(columnName);
        }
        if (rs.next()) {
          throw new IllegalStateException(
                  String.format("Request for column [%s] existing in table [%s] returned more than one results",
                          columnName, tableName));
        }
      } finally {
        rs.close();
      }
    }

    return result;
  }

  @Override
  public boolean tableHasColumn(String tableName, String... columnName) throws SQLException {
    List<String> columnsList = new ArrayList<>(Arrays.asList(columnName));
    DatabaseMetaData metaData = getDatabaseMetaData();

    CustomStringUtils.toUpperCase(columnsList);
    Set<String> columnsListToCheckCopies = new HashSet<>(columnsList);
    List<String> duplicatedColumns = new ArrayList<>();
    ResultSet rs = metaData.getColumns(null, dbSchema, convertObjectName(tableName), null);

    if (rs != null) {
      try {
        while (rs.next()) {
          String actualColumnName = rs.getString("COLUMN_NAME");
          if (actualColumnName != null) {
            boolean removingResult = columnsList.remove(actualColumnName.toUpperCase());
            if (!removingResult && columnsListToCheckCopies.contains(actualColumnName.toUpperCase())) {
              duplicatedColumns.add(actualColumnName.toUpperCase());
            }
          }
        }
      } finally {
        rs.close();
      }
    }
    if (!duplicatedColumns.isEmpty()) {
      throw new IllegalStateException(
              String.format("Request for columns [%s] existing in table [%s] returned too many results [%s] for columns [%s]",
                      Arrays.toString(columnName), tableName, duplicatedColumns.size(), duplicatedColumns.toString()));
    }

    return columnsList.size() == 0;
  }

  @Override
  public boolean tableHasForeignKey(String tableName, String fkName) throws SQLException {
    return getCheckedForeignKey(tableName, fkName) != null;
  }

  public String getCheckedForeignKey(String rawTableName, String rawForeignKeyName) throws SQLException {
    DatabaseMetaData metaData = getDatabaseMetaData();

    String tableName = convertObjectName(rawTableName);
    String foreignKeyName = convertObjectName(rawForeignKeyName);
    try (ResultSet rs = metaData.getImportedKeys(null, dbSchema, tableName)) {
      while (rs.next()) {
        String foundName = rs.getString("FK_NAME");
        if (StringUtils.equals(foreignKeyName, foundName)) {
          return foundName;
        }
      }
    }

    DatabaseType databaseType = configuration.getDatabaseType();
    if (databaseType == DatabaseType.ORACLE) {
      try (PreparedStatement ps = getConnection().prepareStatement("SELECT constraint_name FROM user_constraints WHERE table_name = ? AND constraint_type = 'R' AND constraint_name = ?")) {
        ps.setString(1, tableName);
        ps.setString(2, foreignKeyName);
        try (ResultSet rs = ps.executeQuery()) {
          if (rs.next()) {
            return foreignKeyName;
          }
        }
      }
    }

    LOG.warn("FK {} not found for table {}", foreignKeyName, tableName);

    return null;
  }

  @Override
  public boolean tableHasForeignKey(String tableName, String refTableName,
          String columnName, String refColumnName) throws SQLException {
    return tableHasForeignKey(tableName, refTableName, new String[]{columnName}, new String[]{refColumnName});
  }

  @Override
  public boolean tableHasForeignKey(String tableName, String referenceTableName, String[] keyColumns,
          String[] referenceColumns) throws SQLException {
    DatabaseMetaData metaData = getDatabaseMetaData();

    //NB: reference table contains pk columns while key table contains fk columns
    ResultSet rs = metaData.getCrossReference(null, dbSchema, convertObjectName(referenceTableName),
            null, dbSchema, convertObjectName(tableName));

    List<String> pkColumns = new ArrayList<>(referenceColumns.length);
    for (String referenceColumn : referenceColumns) {
      pkColumns.add(convertObjectName(referenceColumn));
    }
    List<String> fkColumns = new ArrayList<>(keyColumns.length);
    for (String keyColumn : keyColumns) {
      fkColumns.add(convertObjectName(keyColumn));
    }

    if (rs != null) {
      try {
        while (rs.next()) {

          String pkColumn = rs.getString("PKCOLUMN_NAME");
          String fkColumn = rs.getString("FKCOLUMN_NAME");

          int pkIndex = pkColumns.indexOf(pkColumn);
          int fkIndex = fkColumns.indexOf(fkColumn);
          if (pkIndex != -1 && fkIndex != -1) {
            if (pkIndex != fkIndex) {
              LOG.warn("Columns for FK constraint should be provided in exact order");
            } else {
              pkColumns.remove(pkIndex);
              fkColumns.remove(fkIndex);
            }

          } else {
            LOG.debug("pkCol={}, fkCol={} not found in provided column names, skipping", pkColumn, fkColumn); //TODO debug
          }

        }
        if (pkColumns.isEmpty() && fkColumns.isEmpty()) {
          return true;
        }

      } finally {
        rs.close();
      }
    }

    return false;

  }

  @Override
  public boolean tableHasIndex(String tableName, boolean unique, String indexName) throws SQLException{
    if (tableExists(tableName)){
      List<String> indexList = getIndexesList(tableName, false);
      return (CustomStringUtils.containsCaseInsensitive(indexName, indexList));
    }
    return false;
  }

  @Override
  public void createIndex(String indexName, String tableName,
          String... columnNames) throws SQLException {
    createIndex(indexName, tableName, false, columnNames);
  }

  @Override
  public void createIndex(String indexName, String tableName, boolean isUnique,
                          String... columnNames) throws SQLException {
    if (!tableHasIndex(tableName, false, indexName)) {
      String query = dbmsHelper.getCreateIndexStatement(indexName, tableName, isUnique, columnNames);
      executeQuery(query);
    } else {
      LOG.info("Index {} already exist, skipping creation, table = {}", indexName, tableName);
    }
  }

  @Override
  public void addFKConstraint(String tableName, String constraintName,
          String keyColumn, String referenceTableName,
          String referenceColumn, boolean ignoreFailure) throws SQLException {

    addFKConstraint(tableName, constraintName, new String[]{keyColumn}, referenceTableName,
            new String[]{referenceColumn}, false, ignoreFailure);
  }

  @Override
  public void addFKConstraint(String tableName, String constraintName,
          String keyColumn, String referenceTableName,
          String referenceColumn, boolean shouldCascadeOnDelete,
          boolean ignoreFailure) throws SQLException {

    addFKConstraint(tableName, constraintName, new String[]{keyColumn}, referenceTableName,
            new String[]{referenceColumn}, shouldCascadeOnDelete, ignoreFailure);
  }

  @Override
  public void addFKConstraint(String tableName, String constraintName,
          String[] keyColumns, String referenceTableName,
          String[] referenceColumns,
          boolean ignoreFailure) throws SQLException {
    addFKConstraint(tableName, constraintName, keyColumns, referenceTableName, referenceColumns, false, ignoreFailure);
  }

  @Override
  public void addFKConstraint(String tableName, String constraintName,
          String[] keyColumns, String referenceTableName,
          String[] referenceColumns, boolean shouldCascadeOnDelete,
          boolean ignoreFailure) throws SQLException {
    if (!tableHasForeignKey(tableName, referenceTableName, keyColumns, referenceColumns)) {
      String query = dbmsHelper.getAddForeignKeyStatement(tableName, constraintName,
              Arrays.asList(keyColumns),
              referenceTableName,
              Arrays.asList(referenceColumns),
              shouldCascadeOnDelete);

      try {
        executeQuery(query, ignoreFailure);
      } catch (SQLException e) {
        LOG.warn("Add FK constraint failed"
                + ", constraintName = " + constraintName
                + ", tableName = " + tableName, e.getMessage());
        if (!ignoreFailure) {
          throw e;
        }
      }
    } else {
      LOG.info("Foreign Key constraint {} already exists, skipping", constraintName);
    }
  }

  public boolean tableHasConstraint(String tableName, String constraintName) throws SQLException {
    // this kind of request is well lower level as we querying system tables, due that we need for some the name of catalog.
    String query = dbmsHelper.getTableConstraintsStatement(connection.getCatalog(), tableName);
    Statement statement = null;
    ResultSet rs = null;
    try {
      statement = getConnection().createStatement();
      rs = statement.executeQuery(query);
      if (rs != null) {
        while (rs.next()) {
          if (rs.getString("CONSTRAINT_NAME").equalsIgnoreCase(constraintName)) {
            return true;
          }
        }
      }
    } finally {
      if (statement != null) {
        statement.close();
      }
      if (rs != null) {
        rs.close();
      }
    }
    return false;
  }

  @Override
  public void addUniqueConstraint(String tableName, String constraintName, String... columnNames)
          throws SQLException {
    if (!tableHasConstraint(tableName, constraintName) && tableHasColumn(tableName, columnNames)) {
      String query = dbmsHelper.getAddUniqueConstraintStatement(tableName, constraintName, columnNames);
      try {
        executeQuery(query);
      } catch (SQLException e) {
        LOG.warn("Add unique constraint failed, constraintName={},tableName={}", constraintName, tableName);
        throw e;
      }
    } else {
      LOG.info("Unique constraint {} already exists or columns {} not found, skipping", constraintName, StringUtils.join(columnNames, ", "));
    }
  }

  @Override
  public void updateUniqueConstraint(String tableName, String constraintName, String... columnNames)
      throws SQLException {
    dropUniqueConstraint(tableName, constraintName);
    addUniqueConstraint(tableName, constraintName, columnNames);
  }

  @Override
  public void addPKConstraint(String tableName, String constraintName, boolean ignoreErrors, String... columnName) throws SQLException {
    if (!tableHasPrimaryKey(tableName, null) && tableHasColumn(tableName, columnName)) {
      String query = dbmsHelper.getAddPrimaryKeyConstraintStatement(tableName, constraintName, columnName);
      executeQuery(query, ignoreErrors);
    } else {
      LOG.warn("Primary constraint {} not altered to table {} as column {} not present or constraint already exists",
              constraintName, tableName, columnName);
    }
  }

  @Override
  public void addPKConstraint(String tableName, String constraintName, String... columnName) throws SQLException {
    addPKConstraint(tableName, constraintName, false, columnName);
  }

  @Override
  public void renameColumn(String tableName, String oldColumnName,
          DBColumnInfo columnInfo) throws SQLException {
    //it is mandatory to specify type in column change clause for mysql
    String renameColumnStatement = dbmsHelper.getRenameColumnStatement(tableName, oldColumnName, columnInfo);
    executeQuery(renameColumnStatement);

  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void addColumn(String tableName, DBColumnInfo columnInfo) throws SQLException {
    if (tableHasColumn(tableName, columnInfo.getName())) {
      return;
    }

    DatabaseType databaseType = configuration.getDatabaseType();
    switch (databaseType) {
      case ORACLE: {
        // capture the original null value and set the column to nullable if
        // there is a default value
        boolean originalNullable = columnInfo.isNullable();
        if (columnInfo.getDefaultValue() != null) {
          columnInfo.setNullable(true);
        }

        String query = dbmsHelper.getAddColumnStatement(tableName, columnInfo);
        executeQuery(query);

        // update the column after it's been created with the default value and
        // then set the nullable field back to the specified value
        if (columnInfo.getDefaultValue() != null) {
          updateTable(tableName, columnInfo.getName(), columnInfo.getDefaultValue(), "");

          // if the column wasn't originally nullable, then set that here
          if (!originalNullable) {
            setColumnNullable(tableName, columnInfo, originalNullable);
          }

          // finally, add the DEFAULT constraint to the table
          addDefaultConstraint(tableName, columnInfo);
        }
        break;
      }
      case DERBY:
      case MYSQL:
      case POSTGRES:
      case SQL_ANYWHERE:
      case SQL_SERVER:
      default: {  // ToDo: getAddColumnStatement not supporting default clause for binary fields
        String query = dbmsHelper.getAddColumnStatement(tableName, columnInfo);
        executeQuery(query);
        break;
      }
    }
  }

  @Override
  public void alterColumn(String tableName, DBColumnInfo columnInfo) throws SQLException {
    //varchar extension only (derby limitation, but not too much for others),
    if (dbmsHelper.supportsColumnTypeChange()) {
      String statement = dbmsHelper.getAlterColumnStatement(tableName, columnInfo);
      executeQuery(statement);
    } else {
      //use addColumn: add_tmp-update-drop-rename for Derby
      DBColumnInfo columnInfoTmp = new DBColumnInfo(
              columnInfo.getName() + "_TMP",
              columnInfo.getType(),
              columnInfo.getLength());
      String statement = dbmsHelper.getAddColumnStatement(tableName, columnInfoTmp);
      executeQuery(statement);
      updateTable(tableName, columnInfo, columnInfoTmp);
      dropColumn(tableName, columnInfo.getName());
      renameColumn(tableName, columnInfoTmp.getName(), columnInfo);
    }
    if (isColumnNullable(tableName, columnInfo.getName()) != columnInfo.isNullable()) {
      setColumnNullable(tableName, columnInfo, columnInfo.isNullable());
    }
  }

  @Override
  public void updateTable(String tableName, DBColumnInfo columnNameFrom,
          DBColumnInfo columnNameTo) throws SQLException {
    LOG.info("Executing query: UPDATE TABLE " + tableName + " SET "
            + columnNameTo.getName() + "=" + columnNameFrom.getName());

    String statement = "SELECT * FROM " + tableName;
    int typeFrom = getColumnType(tableName, columnNameFrom.getName());
    int typeTo = getColumnType(tableName, columnNameTo.getName());
    Statement dbStatement = null;
    ResultSet rs = null;
    try {
    dbStatement = getConnection().createStatement(ResultSet.TYPE_SCROLL_SENSITIVE,
            ResultSet.CONCUR_UPDATABLE);
    rs = dbStatement.executeQuery(statement);

    while (rs.next()) {
      convertUpdateData(rs, columnNameFrom, typeFrom, columnNameTo, typeTo);
      rs.updateRow();
    }
    } finally {
      if (rs != null) {
        rs.close();
      }
      if (dbStatement != null) {
        dbStatement.close();
      }
    }
  }

  private void convertUpdateData(ResultSet rs, DBColumnInfo columnNameFrom,
          int typeFrom,
          DBColumnInfo columnNameTo, int typeTo) throws SQLException {
    if (typeFrom == Types.BLOB && typeTo == Types.CLOB) {
      //BLOB-->CLOB
      Blob data = rs.getBlob(columnNameFrom.getName());
      if (data != null) {
        rs.updateClob(columnNameTo.getName(),
                new BufferedReader(new InputStreamReader(data.getBinaryStream(), Charset.defaultCharset())));
      }
    } else {
      Object data = rs.getObject(columnNameFrom.getName());
      rs.updateObject(columnNameTo.getName(), data);
    }

  }

  @Override
  public boolean insertRow(String tableName, String[] columnNames, String[] values, boolean ignoreFailure) throws SQLException {
    StringBuilder builder = new StringBuilder();
    builder.append("INSERT INTO ").append(tableName).append("(");
    if (columnNames.length != values.length) {
      throw new IllegalArgumentException("number of columns should be equal to number of values");
    }

    for (int i = 0; i < columnNames.length; i++) {
      builder.append(columnNames[i]);
      if (i != columnNames.length - 1) {
        builder.append(",");
      }
    }

    builder.append(") VALUES(");

    for (int i = 0; i < values.length; i++) {
      builder.append(values[i]);
      if (i != values.length - 1) {
        builder.append(",");
      }
    }

    builder.append(")");

    Statement statement = getConnection().createStatement();
    int rowsUpdated = 0;
    String query = builder.toString();
    try {
      rowsUpdated = statement.executeUpdate(query);
    } catch (SQLException e) {
      LOG.warn("Unable to execute query: " + query, e);
      if (!ignoreFailure) {
        throw e;
      }
    } finally {
      if (statement != null) {
        statement.close();
      }
    }

    return rowsUpdated != 0;
  }

  @Override
  public boolean insertRowIfMissing(String tableName, String[] columnNames, String[] values, boolean ignoreFailure) throws SQLException {
    if (columnNames.length == 0) {
      return false;
    }

    if (columnNames.length != values.length) {
      throw new IllegalArgumentException("number of columns should be equal to number of values");
    }

    StringBuilder builder = new StringBuilder();
    builder.append("SELECT COUNT(*) FROM ").append(tableName);

    builder.append(" WHERE ").append(columnNames[0]).append("=").append(values[0]);
    for (int i = 1; i < columnNames.length; i++) {
      builder.append(" AND ").append(columnNames[i]).append("=").append(values[i]);
    }

    Statement statement = getConnection().createStatement();
    ResultSet resultSet = null;
    int count = -1;
    String query = builder.toString();
    try {
      resultSet = statement.executeQuery(query);

      if ((resultSet != null) && (resultSet.next())) {
        count = resultSet.getInt(1);
      }
    } catch (SQLException e) {
      LOG.warn("Unable to execute query: " + query, e);
      if (!ignoreFailure) {
        throw e;
      }
    } finally {
      if (resultSet != null) {
        resultSet.close();
      }
      if (statement != null) {
        statement.close();
      }
    }

    return (count == 0) && insertRow(tableName, columnNames, values, ignoreFailure);
  }

  @Override
  public int updateTable(String tableName, String columnName, Object value,
          String whereClause) throws SQLException {

    StringBuilder query = new StringBuilder(String.format("UPDATE %s SET %s = ", tableName, columnName));
    query.append(escapeParameter(value));
    query.append(" ");
    query.append(whereClause);

    Statement statement = getConnection().createStatement();
    int res = -1;
    try {
      res = statement.executeUpdate(query.toString());
    } finally {
      if (statement != null) {
        statement.close();
      }
    }
    return res;
  }

  @Override
  public int executeUpdate(String query) throws SQLException {
    return executeUpdate(query, false);
  }

  @Override
  public int executeUpdate(String query, boolean ignoreErrors) throws SQLException {
    Statement statement = getConnection().createStatement();
    try {
      return statement.executeUpdate(query);
    } catch (SQLException e) {
      LOG.warn("Error executing query: " + query + ", "
              + "errorCode = " + e.getErrorCode() + ", message = " + e.getMessage());
      if (!ignoreErrors) {
        throw e;
      }
    } finally {
      if (statement != null) {
        statement.close();
      }
    }
    return 0;  // If error appears and ignoreError is set, return 0 (no changes was made)
  }

  @Override
  public void executeQuery(String query, String tableName, String hasColumnName) throws SQLException {
    if (tableHasColumn(tableName, hasColumnName)) {
      executeQuery(query);
    }
  }

  @Override
  public void executeQuery(String query) throws SQLException {
    executeQuery(query, false);
  }

  @Override
  public void executeQuery(String query, boolean ignoreFailure) throws SQLException {
    LOG.info("Executing query: {}", query);
    Statement statement = getConnection().createStatement();
    try {
      statement.execute(query);
    } catch (SQLException e) {
