Upgrade mysql and snowflake drivers.

Improve connection management, support mysql on windows.

Author: piotrczarnas

dqops/src/main/java/com/dqops/connectors/jdbc/AbstractJdbcSourceConnection.java

@@ -25,6 +25,7 @@
 import com.dqops.utils.exceptions.RunSilently;
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
+import lombok.extern.slf4j.Slf4j;
 import tech.tablesaw.api.Table;
 import tech.tablesaw.columns.Column;
 
@@ -37,6 +38,7 @@
 /**
  * Base abstract class for JDBC based source connections.
  */
+@Slf4j
 public abstract class AbstractJdbcSourceConnection extends AbstractSqlSourceConnection {
     private final JdbcConnectionPool jdbcConnectionPool;
     private Connection jdbcConnection;
@@ -85,6 +87,17 @@ public void open(SecretValueLookupContext secretValueLookupContext) {
             this.jdbcConnection.setAutoCommit(true);
         }
         catch (Exception ex) {
+            if (this.jdbcConnection != null) {
+                try {
+                    this.jdbcConnection.close();
+                }
+                catch (SQLException sqlException) {
+                    log.error("Cannot close a connection during a failure, error: " + sqlException.getMessage(), sqlException);
+                } finally {
+                    this.jdbcConnection = null;
+                }
+            }
+
             if (this.getConnectionSpec().getHierarchyId() != null) {
                 String connectionName = this.getConnectionSpec().getConnectionName();
                 throw new JdbcConnectionFailedException("Connection failed for source " + connectionName + ", error: " + ex.getMessage(), ex);

dqops/src/main/java/com/dqops/connectors/jdbc/JdbcConnectionPoolImpl.java

@@ -19,8 +19,10 @@
 import com.dqops.metadata.sources.ConnectionSpec;
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.RemovalNotification;
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.context.annotation.Scope;
@@ -31,10 +33,11 @@
 import java.util.concurrent.TimeUnit;
 
 /**
- * JDDB connection pool that supports multiple connections.
+ * JDBC connection pool that supports multiple connections.
  */
 @Component
 @Scope(ConfigurableBeanFactory.SCOPE_SINGLETON)
+@Slf4j
 public class JdbcConnectionPoolImpl implements JdbcConnectionPool {
     /**
      * Data sources cache.
@@ -51,9 +54,24 @@ public JdbcConnectionPoolImpl(DqoJdbcConnectionsConfigurationProperties jdbcConn
                 CacheBuilder.newBuilder()
                         .maximumSize(jdbcConnectionsConfigurationProperties.getMaxConnectionInPool())
                         .expireAfterAccess(jdbcConnectionsConfigurationProperties.getExpireAfterAccessSeconds(), TimeUnit.SECONDS)
+                        .removalListener(notification -> onRemoveDataSource(notification))
                         .build();
     }
 
+    /**
+     * Notification called when the cache decided to remove a data source.
+     * @param notification Notification that a data source is removed.
+     */
+    private void onRemoveDataSource(RemovalNotification<Object, Object> notification) {
+        try {
+            HikariDataSource dataSource = (HikariDataSource)notification.getValue();
+            dataSource.close();
+        }
+        catch (Exception ex) {
+            log.error("Cannot close a data source for the connection: " + notification.getKey().toString() + ", error: " + ex.getMessage(), ex);
+        }
+    }
+
     /**
      * Returns or creates a data source for the given connection specification.
      * @param connectionSpec Connection specification (should be not mutable).

dqops/src/main/java/com/dqops/connectors/mysql/MysqlSourceConnection.java

@@ -223,7 +223,41 @@ public String buildListColumnsSql(String schemaName, List<String> tableNames) {
         if (mysqlParametersSpec.getMysqlEngineType() == MysqlEngineType.singlestoredb) {
             return SingleStoreDbSourceConnection.buildListColumnsSql(getConnectionSpec(), schemaName, tableNames, this.getInformationSchemaName());
         } else {
-            return super.buildListColumnsSql(schemaName, tableNames);
+            return this.buildListColumnsSqlForMySql(schemaName, tableNames);
         }
     }
+
+    /**
+     * Creates an SQL for listing columns in the given tables.
+     * @param schemaName Schema name (bigquery dataset name).
+     * @param tableNames Table names to list.
+     * @return SQL of the INFORMATION_SCHEMA query.
+     */
+    public String buildListColumnsSqlForMySql(String schemaName, List<String> tableNames) {
+        StringBuilder sqlBuilder = new StringBuilder();
+        sqlBuilder.append("SELECT * FROM ");
+
+        sqlBuilder.append(getInformationSchemaName());
+        sqlBuilder.append(".COLUMNS ");
+        sqlBuilder.append("WHERE TABLE_SCHEMA='");
+        sqlBuilder.append(schemaName.replace("'", "''"));
+        sqlBuilder.append("'");
+
+        if (tableNames != null && tableNames.size() > 0) {
+            sqlBuilder.append(" AND TABLE_NAME IN (");
+            for (int ti = 0; ti < tableNames.size(); ti++) {
+                String tableName = tableNames.get(ti);
+                if (ti > 0) {
+                    sqlBuilder.append(",");
+                }
+                sqlBuilder.append('\'');
+                sqlBuilder.append(tableName.replace("'", "''"));
+                sqlBuilder.append('\'');
+            }
+            sqlBuilder.append(") ");
+        }
+        sqlBuilder.append("ORDER BY TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME, ORDINAL_POSITION");
+        String sql = sqlBuilder.toString();
+        return sql;
+    }
 }

pom.xml

@@ -46,7 +46,7 @@
             <dependency>
                 <groupId>mysql</groupId>
                 <artifactId>mysql-connector-java</artifactId>
-                <version>8.0.30</version>
+                <version>8.0.33</version>
             </dependency>
             <dependency>
                 <groupId>org.mariadb.jdbc</groupId>
@@ -71,7 +71,7 @@
             <dependency>
                 <groupId>net.snowflake</groupId>
                 <artifactId>snowflake-jdbc</artifactId>
-                <version>3.13.32</version>
+                <version>3.15.1</version>
             </dependency>
         </dependencies>
     </dependencyManagement>