merge 修改

Author: zoudaokoulife

core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java

@@ -436,7 +436,7 @@ private SqlNode replaceSelectFieldName(SqlNode selectNode, HashBasedTable<String
             SqlIdentifier sqlIdentifier = (SqlIdentifier) selectNode;
 
             if(sqlIdentifier.names.size() == 1){
-                return null;
+                return selectNode;
             }
 
             String mappingFieldName = mappingTable.get(sqlIdentifier.getComponent(0).getSimple(), sqlIdentifier.getComponent(1).getSimple());

core/src/main/java/com/dtstack/flink/sql/util/JDBCUtils.java

@@ -0,0 +1,34 @@
+package com.dtstack.flink.sql.util;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.DriverManager;
+
+public class JDBCUtils {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ClassUtil.class);
+
+    public final static String lock_str = "jdbc_lock_str";
+
+    public static void forName(String clazz, ClassLoader classLoader)  {
+        synchronized (lock_str){
+            try {
+                Class.forName(clazz, true, classLoader);
+                DriverManager.setLoginTimeout(10);
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
+
+    public synchronized static void forName(String clazz) {
+        try {
+            Class<?> driverClass = Class.forName(clazz);
+            driverClass.newInstance();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/RetractJDBCOutputFormat.java

@@ -19,6 +19,7 @@
 package com.dtstack.flink.sql.sink.rdb.format;
 
 import com.dtstack.flink.sql.sink.rdb.RdbSink;
+import com.dtstack.flink.sql.util.JDBCUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
@@ -136,7 +137,7 @@ public void open(int taskNumber, int numTasks) throws IOException {
 
     private Connection establishConnection() throws SQLException, ClassNotFoundException {
         Connection connection ;
-        Class.forName(drivername);
+        JDBCUtils.forName(drivername, getClass().getClassLoader());
         if (username == null) {
             connection = DriverManager.getConnection(dbURL);
         } else {