Merge branch '1.8_release_3.10.x' into 1.8_test_3.10.x

Author: gituser

README.md

@@ -15,8 +15,8 @@ FlinkStreamSQL
     * 自定义create view 语法
     * 自定义create function 语法
     * 实现了流与维表的join
-    * 支持原生FLinkSQL所有的语法
-    * 扩展了输入和输出的性能指标到promethus
+    * 支持原生FlinkSQL所有的语法
+    * 扩展了输入和输出的性能指标到Task metrics
 
 ## 目录
 

core/src/main/java/com/dtstack/flink/sql/environment/MyLocalStreamEnvironment.java

@@ -22,6 +22,7 @@
 import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.minicluster.MiniCluster;
@@ -93,33 +94,48 @@ public JobExecutionResult execute(String jobName) throws Exception {
         // transform the streaming program into a JobGraph
         StreamGraph streamGraph = getStreamGraph();
         streamGraph.setJobName(jobName);
+        return execute(streamGraph);
+    }
+
+    public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
 
         JobGraph jobGraph = streamGraph.getJobGraph();
         jobGraph.setClasspaths(classpaths);
+        jobGraph.setAllowQueuedScheduling(true);
 
         Configuration configuration = new Configuration();
         configuration.addAll(jobGraph.getJobConfiguration());
-
-        configuration.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "512M");
-        configuration.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, jobGraph.getMaximumParallelism());
+        configuration.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "0");
 
         // add (and override) the settings with what the user defined
         configuration.addAll(this.conf);
 
-        MiniClusterConfiguration.Builder configBuilder = new MiniClusterConfiguration.Builder();
-        configBuilder.setConfiguration(configuration);
+        if (!configuration.contains(RestOptions.BIND_PORT)) {
+            configuration.setString(RestOptions.BIND_PORT, "0");
+        }
+
+        int numSlotsPerTaskManager = configuration.getInteger(TaskManagerOptions.NUM_TASK_SLOTS, jobGraph.getMaximumParallelism());
+
+        MiniClusterConfiguration cfg = new MiniClusterConfiguration.Builder()
+                .setConfiguration(configuration)
+                .setNumSlotsPerTaskManager(numSlotsPerTaskManager)
+                .build();
 
         if (LOG.isInfoEnabled()) {
             LOG.info("Running job on local embedded Flink mini cluster");
         }
 
-        try (MiniCluster exec = new MiniCluster(configBuilder.build());) {
-            exec.start();
-            JobExecutionResult jobExecutionResult = exec.executeJobBlocking(jobGraph);
+        MiniCluster miniCluster = new MiniCluster(cfg);
+
+        try {
+            miniCluster.start();
+            configuration.setInteger(RestOptions.PORT, miniCluster.getRestAddress().get().getPort());
+
+            return miniCluster.executeJobBlocking(jobGraph);
+        }
+        finally {
             transformations.clear();
-            return jobExecutionResult;
-        } catch (Exception e) {
-            throw new RuntimeException(e);
+            miniCluster.close();
         }
     }
 }

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

@@ -198,14 +198,18 @@ private static void sqlTranslation(String localSqlPluginPath,
 
         SideSqlExec sideSqlExec = new SideSqlExec();
         sideSqlExec.setLocalSqlPluginPath(localSqlPluginPath);
+
+        int scope = 0;
         for (CreateTmpTableParser.SqlParserResult result : sqlTree.getTmpSqlList()) {
-            sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig, result);
+            sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig, result, scope + "");
+            scope++;
         }
 
         for (InsertSqlParser.SqlParseResult result : sqlTree.getExecSqlList()) {
             if (LOG.isInfoEnabled()) {
                 LOG.info("exe-sql:\n" + result.getExecSql());
             }
+
             boolean isSide = false;
             for (String tableName : result.getTargetTableList()) {
                 if (sqlTree.getTmpTableMap().containsKey(tableName)) {
@@ -216,7 +220,7 @@ private static void sqlTranslation(String localSqlPluginPath,
                     SqlNode sqlNode = flinkPlanner.parse(realSql);
                     String tmpSql = ((SqlInsert) sqlNode).getSource().toString();
                     tmp.setExecSql(tmpSql);
-                    sideSqlExec.exec(tmp.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig, tmp);
+                    sideSqlExec.exec(tmp.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig, tmp, scope + "");
                 } else {
                     for (String sourceTable : result.getSourceTableList()) {
                         if (sideTableMap.containsKey(sourceTable)) {
@@ -226,7 +230,7 @@ private static void sqlTranslation(String localSqlPluginPath,
                     }
                     if (isSide) {
                         //sql-dimensional table contains the dimension table of execution
-                        sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig, null);
+                        sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig, null, null);
                     } else {
                         LOG.info("----------exec sql without dimension join-----------");
                         LOG.info("----------real sql exec is--------------------------\n{}", result.getExecSql());
@@ -236,6 +240,8 @@ private static void sqlTranslation(String localSqlPluginPath,
                         }
                     }
                 }
+
+                scope++;
             }
         }
     }

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

@@ -203,7 +203,7 @@ private void invokeWithCache(Map<String, Object> inputParams, CRow input, Result
                     return;
                 }else if(ECacheContentType.SingleLine == val.getType()){
                     try {
-                        Row row = fillData(input.row(), val);
+                        Row row = fillData(input.row(), val.getContent());
                         resultFuture.complete(Collections.singleton(new CRow(row, input.change())));
                     } catch (Exception e) {
                         dealFillDataError(input, resultFuture, e);

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

@@ -20,6 +20,7 @@
 
 package com.dtstack.flink.sql.side;
 
+import com.dtstack.flink.sql.util.TableUtils;
 import com.google.common.collect.HashBasedTable;
 import com.google.common.collect.Maps;
 import org.apache.calcite.sql.JoinType;
@@ -66,6 +67,8 @@ public class JoinInfo implements Serializable {
 
     private JoinType joinType;
 
+    private String scope = "";
+
     /**
      * 左表需要查询的字段信息和output的时候对应的列名称
      */
@@ -96,12 +99,14 @@ public String getNewTableName(){
         //兼容左边表是as 的情况
         String leftStr = leftTableName;
         leftStr = Strings.isNullOrEmpty(leftStr) ? leftTableAlias : leftStr;
-        return leftStr + "_" + rightTableName;
+        String newName =  leftStr + "_" + rightTableName;
+        return TableUtils.buildTableNameWithScope(newName, scope);
     }
 
 
     public String getNewTableAlias(){
-        return leftTableAlias + "_" + rightTableAlias;
+        String newName = leftTableAlias + "_" + rightTableAlias;
+        return TableUtils.buildTableNameWithScope(newName, scope);
     }
 
     public boolean isLeftIsSideTable() {
@@ -233,6 +238,14 @@ public HashBasedTable<String, String, String> getTableFieldRef(){
         return mappingTable;
     }
 
+    public String getScope() {
+        return scope;
+    }
+
+    public void setScope(String scope) {
+        this.scope = scope;
+    }
+
     @Override
     public String toString() {
         return "JoinInfo{" +