merge conflicts.

Author: HiLany

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -55,6 +55,7 @@
 import org.apache.calcite.sql.SqlInsert;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.commons.io.Charsets;
+import org.apache.commons.lang3.SerializationUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
@@ -79,6 +80,7 @@
 import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -244,7 +246,11 @@ private static void sqlTranslation(String localSqlPluginPath,
             scope++;
         }
 
+        final Map<String, AbstractSideTableInfo> tmpTableMap = new HashMap<>();
         for (InsertSqlParser.SqlParseResult result : sqlTree.getExecSqlList()) {
+            // prevent current sql use last sql's sideTableInfo
+            sideTableMap.forEach((s, abstractSideTableInfo) -> tmpTableMap.put(s, SerializationUtils.clone(abstractSideTableInfo)));
+
             if (LOG.isInfoEnabled()) {
                 LOG.info("exe-sql:\n" + result.getExecSql());
             }
@@ -257,17 +263,17 @@ private static void sqlTranslation(String localSqlPluginPath,
                     SqlNode sqlNode = flinkPlanner.getParser().parse(realSql);
                     String tmpSql = ((SqlInsert) sqlNode).getSource().toString();
                     tmp.setExecSql(tmpSql);
-                    sideSqlExec.exec(tmp.getExecSql(), sideTableMap, tableEnv, registerTableCache, tmp, scope + "");
+                    sideSqlExec.exec(tmp.getExecSql(), tmpTableMap, tableEnv, registerTableCache, tmp, scope + "");
                 } else {
                     for (String sourceTable : result.getSourceTableList()) {
-                        if (sideTableMap.containsKey(sourceTable)) {
+                        if (tmpTableMap.containsKey(sourceTable)) {
                             isSide = true;
                             break;
                         }
                     }
                     if (isSide) {
                         //sql-dimensional table contains the dimension table of execution
-                        sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache, null, String.valueOf(scope));
+                        sideSqlExec.exec(result.getExecSql(), tmpTableMap, tableEnv, registerTableCache, null, String.valueOf(scope));
                     } else {
                         LOG.info("----------exec sql without dimension join-----------");
                         LOG.info("----------real sql exec is--------------------------\n{}", result.getExecSql());
@@ -280,6 +286,7 @@ private static void sqlTranslation(String localSqlPluginPath,
 
                 scope++;
             }
+            tmpTableMap.clear();
         }
     }
 

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java

@@ -84,7 +84,7 @@ public class RdbAsyncReqRow extends BaseAsyncReqRow {
 
     public final static String PREFERRED_TEST_QUERY_SQL = "SELECT 1 FROM DUAL";
 
-    private static SQLClient rdbSqlClient;
+    private transient SQLClient rdbSqlClient;
 
     private AtomicBoolean connectionStatus = new AtomicBoolean(true);
 
@@ -126,7 +126,7 @@ public void handleAsyncInvoke(Map<String, Object> inputParams, BaseRow input, Re
             Thread.sleep(100);
         }
         Map<String, Object> params = formatInputParam(inputParams);
-        executor.execute(() -> connectWithRetry(params, input, resultFuture, RdbAsyncReqRow.rdbSqlClient));
+        executor.execute(() -> connectWithRetry(params, input, resultFuture, rdbSqlClient));
     }
 
     protected void asyncQueryData(Map<String, Object> inputParams,
@@ -280,8 +280,8 @@ public BaseRow fillData(BaseRow input, Object line) {
     @Override
     public void close() throws Exception {
         super.close();
-        if (RdbAsyncReqRow.rdbSqlClient != null) {
-            RdbAsyncReqRow.rdbSqlClient.close();
+        if (rdbSqlClient != null) {
+            rdbSqlClient.close();
         }
 
         if (executor != null) {
@@ -291,7 +291,7 @@ public void close() throws Exception {
     }
 
     public void setRdbSqlClient(SQLClient rdbSqlClient) {
-        RdbAsyncReqRow.rdbSqlClient = rdbSqlClient;
+        this.rdbSqlClient = rdbSqlClient;
     }
 
     private void handleQuery(SQLConnection connection, Map<String, Object> inputParams, BaseRow input, ResultFuture<BaseRow> resultFuture) {