[Fix-7291][datasource] Compatible with different versions of MySQL data source loading (#7291) (#7305)

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-api/src/main/java/org/apache/dolphinscheduler/plugin/datasource/api/client/CommonDataSourceClient.java

@@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.plugin.datasource.api.client;
 import org.apache.dolphinscheduler.plugin.datasource.api.provider.JdbcDataSourceProvider;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceClient;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 import org.apache.dolphinscheduler.spi.utils.StringUtils;
 
 import java.sql.Connection;
@@ -46,11 +47,11 @@ public class CommonDataSourceClient implements DataSourceClient {
     protected DataSource dataSource;
     protected JdbcTemplate jdbcTemplate;
 
-    public CommonDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public CommonDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
         this.baseConnectionParam = baseConnectionParam;
         preInit();
         checkEnv(baseConnectionParam);
-        initClient(baseConnectionParam);
+        initClient(baseConnectionParam, dbType);
         checkClient();
     }
 
@@ -63,8 +64,8 @@ public class CommonDataSourceClient implements DataSourceClient {
         checkUser(baseConnectionParam);
     }
 
-    protected void initClient(BaseConnectionParam baseConnectionParam) {
+    protected void initClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        this.dataSource = JdbcDataSourceProvider.createJdbcDataSource(baseConnectionParam);
+        this.dataSource = JdbcDataSourceProvider.createJdbcDataSource(baseConnectionParam, dbType);
         this.jdbcTemplate = new JdbcTemplate(dataSource);
     }
 

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-api/src/main/java/org/apache/dolphinscheduler/plugin/datasource/api/plugin/DataSourceClientProvider.java

@@ -61,7 +61,7 @@ public class DataSourceClientProvider {
             if (null == dataSourceChannel) {
                 throw new RuntimeException(String.format("datasource plugin '%s' is not found", dbType.getDescp()));
             }
-            return dataSourceChannel.createDataSourceClient(baseConnectionParam);
+            return dataSourceChannel.createDataSourceClient(baseConnectionParam, dbType);
         });
         return dataSourceClient.getConnection();
     }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-api/src/main/java/org/apache/dolphinscheduler/plugin/datasource/api/provider/JdbcDataSourceProvider.java

@@ -17,10 +17,15 @@
 
 package org.apache.dolphinscheduler.plugin.datasource.api.provider;
 
+import org.apache.dolphinscheduler.plugin.datasource.api.utils.DataSourceUtils;
 import org.apache.dolphinscheduler.plugin.datasource.api.utils.PasswordUtils;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 import org.apache.dolphinscheduler.spi.utils.Constants;
 import org.apache.dolphinscheduler.spi.utils.PropertyUtils;
+import org.apache.dolphinscheduler.spi.utils.StringUtils;
+
+import java.sql.Driver;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -34,10 +39,14 @@ public class JdbcDataSourceProvider {
 
     private static final Logger logger = LoggerFactory.getLogger(JdbcDataSourceProvider.class);
 
-    public static HikariDataSource createJdbcDataSource(BaseConnectionParam properties) {
+    public static HikariDataSource createJdbcDataSource(BaseConnectionParam properties, DbType dbType) {
         logger.info("Creating HikariDataSource pool for maxActive:{}", PropertyUtils.getInt(Constants.SPRING_DATASOURCE_MAX_ACTIVE, 50));
         HikariDataSource dataSource = new HikariDataSource();
 
+        //TODO Support multiple versions of data sources
+        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+        loaderJdbcDriver(classLoader, properties, dbType);
+
         dataSource.setDriverClassName(properties.getDriverClassName());
         dataSource.setJdbcUrl(properties.getJdbcUrl());
         dataSource.setUsername(properties.getUser());
@@ -80,4 +89,25 @@ public class JdbcDataSourceProvider {
         return dataSource;
     }
 
+    protected static void loaderJdbcDriver(ClassLoader classLoader, BaseConnectionParam properties, DbType dbType) {
+        String drv = StringUtils.isBlank(properties.getDriverClassName()) ? DataSourceUtils.getDatasourceProcessor(dbType).getDatasourceDriver() : properties.getDriverClassName();
+        try {
+            final Class<?> clazz = Class.forName(drv, true, classLoader);
+            final Driver driver = (Driver) clazz.newInstance();
+            if (!driver.acceptsURL(properties.getJdbcUrl())) {
+                logger.warn("Jdbc driver loading error. Driver {} cannot accept url.", drv);
+                throw new RuntimeException("Jdbc driver loading error.");
+            }
+            if (dbType.equals(DbType.MYSQL)) {
+                if (driver.getMajorVersion() >= 8) {
+                    properties.setDriverClassName(drv);
+                } else {
+                    properties.setDriverClassName(Constants.COM_MYSQL_JDBC_DRIVER);
+                }
+            }
+        } catch (final Exception e) {
+            logger.warn("The specified driver not suitable.");
+        }
+    }
+
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-api/src/test/java/org/apache/dolphinscheduler/plugin/datasource/api/client/CommonDataSourceClientTest.java

@@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.plugin.datasource.api.client;
 import org.apache.dolphinscheduler.plugin.datasource.api.datasource.mysql.MysqlConnectionParam;
 import org.apache.dolphinscheduler.plugin.datasource.api.provider.JdbcDataSourceProvider;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 import java.sql.Connection;
 
@@ -79,9 +80,9 @@ public class CommonDataSourceClientTest {
     @Test
     public void testInitClient() {
         BaseConnectionParam baseConnectionParam = new MysqlConnectionParam();
-        PowerMockito.doNothing().when(commonDataSourceClient).initClient(Mockito.any(BaseConnectionParam.class));
+        PowerMockito.doNothing().when(commonDataSourceClient).initClient(Mockito.any(BaseConnectionParam.class), Mockito.any());
-        commonDataSourceClient.initClient(baseConnectionParam);
+        commonDataSourceClient.initClient(baseConnectionParam, DbType.MYSQL);
-        Mockito.verify(commonDataSourceClient).initClient(Mockito.any(BaseConnectionParam.class));
+        Mockito.verify(commonDataSourceClient).initClient(Mockito.any(BaseConnectionParam.class), Mockito.any());
     }
 
     @Test

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-api/src/test/java/org/apache/dolphinscheduler/plugin/datasource/api/provider/JdbcDataSourceProviderTest.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.plugin.datasource.api.provider;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.datasource.mysql.MysqlConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -37,8 +38,8 @@ public class JdbcDataSourceProviderTest {
     public void testCreateJdbcDataSource() {
         PowerMockito.mockStatic(JdbcDataSourceProvider.class);
         HikariDataSource dataSource = PowerMockito.mock(HikariDataSource.class);
-        PowerMockito.when(JdbcDataSourceProvider.createJdbcDataSource(Mockito.any())).thenReturn(dataSource);
+        PowerMockito.when(JdbcDataSourceProvider.createJdbcDataSource(Mockito.any(), Mockito.any())).thenReturn(dataSource);
-        Assert.assertNotNull(JdbcDataSourceProvider.createJdbcDataSource(new MysqlConnectionParam()));
+        Assert.assertNotNull(JdbcDataSourceProvider.createJdbcDataSource(new MysqlConnectionParam(), DbType.MYSQL));
     }
 
     @Test

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-clickhouse/src/main/java/org/apache/dolphinscheduler/plugin/datasource/clickhouse/ClickhouseDataSourceChannel.java

@@ -20,11 +20,12 @@ package org.apache.dolphinscheduler.plugin.datasource.clickhouse;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceClient;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 public class ClickhouseDataSourceChannel implements DataSourceChannel {
 
     @Override
-    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        return new ClickhouseDataSourceClient(baseConnectionParam);
+        return new ClickhouseDataSourceClient(baseConnectionParam, dbType);
     }
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-clickhouse/src/main/java/org/apache/dolphinscheduler/plugin/datasource/clickhouse/ClickhouseDataSourceClient.java

@@ -19,11 +19,12 @@ package org.apache.dolphinscheduler.plugin.datasource.clickhouse;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.client.CommonDataSourceClient;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 public class ClickhouseDataSourceClient extends CommonDataSourceClient {
 
-    public ClickhouseDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public ClickhouseDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        super(baseConnectionParam);
+        super(baseConnectionParam, dbType);
     }
 
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-clickhouse/src/test/java/org/apache/dolphinscheduler/plugin/datasource/clickhouse/ClickhouseDataSourceChannelTest.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.plugin.datasource.clickhouse;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.datasource.clickhouse.ClickhouseConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -37,7 +38,7 @@ public class ClickhouseDataSourceChannelTest {
     public void testCreateDataSourceClient() {
         ClickhouseDataSourceChannel sourceChannel = PowerMockito.mock(ClickhouseDataSourceChannel.class);
         ClickhouseDataSourceClient dataSourceClient = PowerMockito.mock(ClickhouseDataSourceClient.class);
-        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any())).thenReturn(dataSourceClient);
+        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any(), Mockito.any())).thenReturn(dataSourceClient);
-        Assert.assertNotNull(sourceChannel.createDataSourceClient(new ClickhouseConnectionParam()));
+        Assert.assertNotNull(sourceChannel.createDataSourceClient(new ClickhouseConnectionParam(), DbType.CLICKHOUSE));
     }
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-db2/src/main/java/org/apache/dolphinscheduler/plugin/datasource/db2/DB2DataSourceChannel.java

@@ -20,11 +20,12 @@ package org.apache.dolphinscheduler.plugin.datasource.db2;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceClient;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 public class DB2DataSourceChannel implements DataSourceChannel {
 
     @Override
-    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        return new DB2DataSourceClient(baseConnectionParam);
+        return new DB2DataSourceClient(baseConnectionParam, dbType);
     }
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-db2/src/main/java/org/apache/dolphinscheduler/plugin/datasource/db2/DB2DataSourceClient.java

@@ -19,11 +19,12 @@ package org.apache.dolphinscheduler.plugin.datasource.db2;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.client.CommonDataSourceClient;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 public class DB2DataSourceClient extends CommonDataSourceClient {
 
-    public DB2DataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public DB2DataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        super(baseConnectionParam);
+        super(baseConnectionParam, dbType);
     }
 
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-db2/src/test/java/org/apache/dolphinscheduler/plugin/datasource/db2/DB2DataSourceChannelTest.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.plugin.datasource.db2;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.datasource.db2.Db2ConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -37,7 +38,7 @@ public class DB2DataSourceChannelTest {
     public void testCreateDataSourceClient() {
         DB2DataSourceChannel sourceChannel = PowerMockito.mock(DB2DataSourceChannel.class);
         DB2DataSourceClient dataSourceClient = PowerMockito.mock(DB2DataSourceClient.class);
-        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any())).thenReturn(dataSourceClient);
+        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any(), Mockito.any())).thenReturn(dataSourceClient);
-        Assert.assertNotNull(sourceChannel.createDataSourceClient(new Db2ConnectionParam()));
+        Assert.assertNotNull(sourceChannel.createDataSourceClient(new Db2ConnectionParam(), DbType.DB2));
     }
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-hive/src/main/java/org/apache/dolphinscheduler/plugin/datasource/hive/HiveDataSourceChannel.java

@@ -20,11 +20,12 @@ package org.apache.dolphinscheduler.plugin.datasource.hive;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceClient;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 public class HiveDataSourceChannel implements DataSourceChannel {
 
     @Override
-    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        return new HiveDataSourceClient(baseConnectionParam);
+        return new HiveDataSourceClient(baseConnectionParam, dbType);
     }
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-hive/src/main/java/org/apache/dolphinscheduler/plugin/datasource/hive/HiveDataSourceClient.java

@@ -24,6 +24,7 @@ import org.apache.dolphinscheduler.plugin.datasource.api.client.CommonDataSource
 import org.apache.dolphinscheduler.plugin.datasource.api.provider.JdbcDataSourceProvider;
 import org.apache.dolphinscheduler.plugin.datasource.utils.CommonUtil;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 import org.apache.dolphinscheduler.spi.utils.Constants;
 import org.apache.dolphinscheduler.spi.utils.PropertyUtils;
 import org.apache.dolphinscheduler.spi.utils.StringUtils;
@@ -56,8 +57,8 @@ public class HiveDataSourceClient extends CommonDataSourceClient {
     protected HikariDataSource oneSessionDataSource;
     private UserGroupInformation ugi;
 
-    public HiveDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public HiveDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        super(baseConnectionParam);
+        super(baseConnectionParam, dbType);
     }
 
     @Override
@@ -67,7 +68,7 @@ public class HiveDataSourceClient extends CommonDataSourceClient {
     }
 
     @Override
-    protected void initClient(BaseConnectionParam baseConnectionParam) {
+    protected void initClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
         logger.info("Create Configuration for hive configuration.");
         this.hadoopConf = createHadoopConf();
         logger.info("Create Configuration success.");
@@ -76,7 +77,7 @@ public class HiveDataSourceClient extends CommonDataSourceClient {
         this.ugi = createUserGroupInformation(baseConnectionParam.getUser());
         logger.info("Create ugi success.");
 
-        super.initClient(baseConnectionParam);
+        super.initClient(baseConnectionParam, dbType);
         this.oneSessionDataSource = JdbcDataSourceProvider.createOneSessionJdbcDataSource(baseConnectionParam);
         logger.info("Init {} success.", getClass().getName());
     }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-hive/src/test/java/org/apache/dolphinscheduler/plugin/datasource/hive/HiveDataSourceChannelTest.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.plugin.datasource.hive;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.datasource.hive.HiveConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -37,7 +38,7 @@ public class HiveDataSourceChannelTest {
     public void testCreateDataSourceClient() {
         HiveDataSourceChannel sourceChannel = PowerMockito.mock(HiveDataSourceChannel.class);
         HiveDataSourceClient dataSourceClient = PowerMockito.mock(HiveDataSourceClient.class);
-        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any())).thenReturn(dataSourceClient);
+        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any(), Mockito.any())).thenReturn(dataSourceClient);
-        Assert.assertNotNull(sourceChannel.createDataSourceClient(new HiveConnectionParam()));
+        Assert.assertNotNull(sourceChannel.createDataSourceClient(new HiveConnectionParam(), DbType.HIVE));
     }
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-mysql/src/main/java/org/apache/dolphinscheduler/plugin/datasource/mysql/MysqlDataSourceChannel.java

@@ -20,11 +20,12 @@ package org.apache.dolphinscheduler.plugin.datasource.mysql;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceClient;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 public class MysqlDataSourceChannel implements DataSourceChannel {
 
     @Override
-    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        return new MysqlDataSourceClient(baseConnectionParam);
+        return new MysqlDataSourceClient(baseConnectionParam, dbType);
     }
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-mysql/src/main/java/org/apache/dolphinscheduler/plugin/datasource/mysql/MysqlDataSourceClient.java

@@ -19,11 +19,12 @@ package org.apache.dolphinscheduler.plugin.datasource.mysql;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.client.CommonDataSourceClient;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 public class MysqlDataSourceClient extends CommonDataSourceClient {
 
-    public MysqlDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public MysqlDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        super(baseConnectionParam);
+        super(baseConnectionParam, dbType);
     }
 
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-mysql/src/test/java/org/apache/dolphinscheduler/plugin/datasource/mysql/MysqlDataSourceChannelTest.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.plugin.datasource.mysql;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.datasource.mysql.MysqlConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -37,7 +38,7 @@ public class MysqlDataSourceChannelTest {
     public void testCreateDataSourceClient() {
         MysqlDataSourceChannel sourceChannel = PowerMockito.mock(MysqlDataSourceChannel.class);
         MysqlDataSourceClient dataSourceClient = PowerMockito.mock(MysqlDataSourceClient.class);
-        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any())).thenReturn(dataSourceClient);
+        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any(), Mockito.any())).thenReturn(dataSourceClient);
-        Assert.assertNotNull(sourceChannel.createDataSourceClient(new MysqlConnectionParam()));
+        Assert.assertNotNull(sourceChannel.createDataSourceClient(new MysqlConnectionParam(), DbType.MYSQL));
     }
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-oracle/src/main/java/org/apache/dolphinscheduler/plugin/datasource/oracle/OracleDataSourceChannel.java

@@ -20,11 +20,12 @@ package org.apache.dolphinscheduler.plugin.datasource.oracle;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceClient;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 public class OracleDataSourceChannel implements DataSourceChannel {
 
     @Override
-    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        return new OracleDataSourceClient(baseConnectionParam);
+        return new OracleDataSourceClient(baseConnectionParam, dbType);
     }
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-oracle/src/main/java/org/apache/dolphinscheduler/plugin/datasource/oracle/OracleDataSourceClient.java

@@ -19,11 +19,12 @@ package org.apache.dolphinscheduler.plugin.datasource.oracle;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.client.CommonDataSourceClient;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 public class OracleDataSourceClient extends CommonDataSourceClient {
 
-    public OracleDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public OracleDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        super(baseConnectionParam);
+        super(baseConnectionParam, dbType);
     }
 
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-oracle/src/test/java/org/apache/dolphinscheduler/plugin/datasource/oracle/OracleDataSourceChannelTest.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.plugin.datasource.oracle;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.datasource.oracle.OracleConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -37,7 +38,7 @@ public class OracleDataSourceChannelTest {
     public void testCreateDataSourceClient() {
         OracleDataSourceChannel sourceChannel = PowerMockito.mock(OracleDataSourceChannel.class);
         OracleDataSourceClient dataSourceClient = PowerMockito.mock(OracleDataSourceClient.class);
-        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any())).thenReturn(dataSourceClient);
+        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any(), Mockito.any())).thenReturn(dataSourceClient);
-        Assert.assertNotNull(sourceChannel.createDataSourceClient(new OracleConnectionParam()));
+        Assert.assertNotNull(sourceChannel.createDataSourceClient(new OracleConnectionParam(), DbType.ORACLE));
     }
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-postgresql/src/main/java/org/apache/dolphinscheduler/plugin/datasource/postgresql/PostgresqlDataSourceChannel.java

@@ -20,11 +20,12 @@ package org.apache.dolphinscheduler.plugin.datasource.postgresql;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceClient;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 public class PostgresqlDataSourceChannel implements DataSourceChannel {
 
     @Override
-    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        return new PostgresqlDataSourceClient(baseConnectionParam);
+        return new PostgresqlDataSourceClient(baseConnectionParam, dbType);
     }
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-postgresql/src/main/java/org/apache/dolphinscheduler/plugin/datasource/postgresql/PostgresqlDataSourceClient.java

@@ -19,11 +19,12 @@ package org.apache.dolphinscheduler.plugin.datasource.postgresql;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.client.CommonDataSourceClient;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 public class PostgresqlDataSourceClient extends CommonDataSourceClient {
 
-    public PostgresqlDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public PostgresqlDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        super(baseConnectionParam);
+        super(baseConnectionParam, dbType);
     }
 
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-postgresql/src/test/java/org/apache/dolphinscheduler/plugin/datasource/postgresql/PostgresqlDataSourceChannelTest.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.plugin.datasource.postgresql;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.datasource.postgresql.PostgreSqlConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -37,7 +38,7 @@ public class PostgresqlDataSourceChannelTest {
     public void testCreateDataSourceClient() {
         PostgresqlDataSourceChannel sourceChannel = PowerMockito.mock(PostgresqlDataSourceChannel.class);
         PostgresqlDataSourceClient dataSourceClient = PowerMockito.mock(PostgresqlDataSourceClient.class);
-        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any())).thenReturn(dataSourceClient);
+        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any(), Mockito.any())).thenReturn(dataSourceClient);
-        Assert.assertNotNull(sourceChannel.createDataSourceClient(new PostgreSqlConnectionParam()));
+        Assert.assertNotNull(sourceChannel.createDataSourceClient(new PostgreSqlConnectionParam(), DbType.POSTGRESQL));
     }
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-sqlserver/src/main/java/org/apache/dolphinscheduler/plugin/datasource/sqlserver/SqlServerDataSourceChannel.java

@@ -20,11 +20,12 @@ package org.apache.dolphinscheduler.plugin.datasource.sqlserver;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceClient;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 public class SqlServerDataSourceChannel implements DataSourceChannel {
 
     @Override
-    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        return new SqlserverDataSourceClient(baseConnectionParam);
+        return new SqlserverDataSourceClient(baseConnectionParam, dbType);
     }
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-sqlserver/src/main/java/org/apache/dolphinscheduler/plugin/datasource/sqlserver/SqlserverDataSourceClient.java

@@ -19,11 +19,12 @@ package org.apache.dolphinscheduler.plugin.datasource.sqlserver;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.client.CommonDataSourceClient;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 public class SqlserverDataSourceClient extends CommonDataSourceClient {
 
-    public SqlserverDataSourceClient(BaseConnectionParam baseConnectionParam) {
+    public SqlserverDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
-        super(baseConnectionParam);
+        super(baseConnectionParam, dbType);
     }
 
 }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-sqlserver/src/test/java/org/apache/dolphinscheduler/plugin/datasource/sqlserver/SqlserverDataSourceChannelTest.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.plugin.datasource.sqlserver;
 
 import org.apache.dolphinscheduler.plugin.datasource.api.datasource.sqlserver.SqlServerConnectionParam;
+import org.apache.dolphinscheduler.spi.enums.DbType;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -37,7 +38,7 @@ public class SqlserverDataSourceChannelTest {
     public void testCreateDataSourceClient() {
         SqlServerDataSourceChannel sourceChannel = PowerMockito.mock(SqlServerDataSourceChannel.class);
         SqlserverDataSourceClient dataSourceClient = PowerMockito.mock(SqlserverDataSourceClient.class);
-        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any())).thenReturn(dataSourceClient);
+        PowerMockito.when(sourceChannel.createDataSourceClient(Mockito.any(), Mockito.any())).thenReturn(dataSourceClient);
-        Assert.assertNotNull(sourceChannel.createDataSourceClient(new SqlServerConnectionParam()));
+        Assert.assertNotNull(sourceChannel.createDataSourceClient(new SqlServerConnectionParam(), DbType.SQLSERVER));
     }
 }

dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/datasource/DataSourceChannel.java

@@ -17,7 +17,9 @@
 
 package org.apache.dolphinscheduler.spi.datasource;
 
+import org.apache.dolphinscheduler.spi.enums.DbType;
+
 public interface DataSourceChannel {
 
-    DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam);
+    DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType);
 }

dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/utils/Constants.java

@@ -123,6 +123,7 @@ public class Constants {
      */
     public static final String ORG_POSTGRESQL_DRIVER = "org.postgresql.Driver";
     public static final String COM_MYSQL_CJ_JDBC_DRIVER = "com.mysql.cj.jdbc.Driver";
+    public static final String COM_MYSQL_JDBC_DRIVER = "com.mysql.jdbc.Driver";
     public static final String ORG_APACHE_HIVE_JDBC_HIVE_DRIVER = "org.apache.hive.jdbc.HiveDriver";
     public static final String COM_CLICKHOUSE_JDBC_DRIVER = "ru.yandex.clickhouse.ClickHouseDriver";
     public static final String COM_ORACLE_JDBC_DRIVER = "oracle.jdbc.OracleDriver";