Merge branch 'master' of github.com:DTStack/flinkStreamSQL into v1.8.0_dev

Author: todd5167

README.md

@@ -58,7 +58,7 @@ mvn clean package -Dmaven.test.skip
 #### 1.4.1 启动命令
 
 ```
-sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack/150_flinkplugin/sqlplugin   -localSqlPluginPath D:\gitspace\flinkStreamSQL\plugins   -addjar \["udf.jar\"\] -mode yarn -flinkconf D:\flink_home\kudu150etc  -yarnconf D:\hadoop\etc\hadoopkudu -confProp \{\"time.characteristic\":\"EventTime\",\"sql.checkpoint.interval\":10000\}
+sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack/150_flinkplugin/sqlplugin   -localSqlPluginPath D:\gitspace\flinkStreamSQL\plugins   -addjar \["udf.jar\"\] -mode yarn -flinkconf D:\flink_home\kudu150etc  -yarnconf D:\hadoop\etc\hadoopkudu -confProp \{\"time.characteristic\":\"EventTime\",\"sql.checkpoint.interval\":10000\} -yarnSessionConf \{\"yid\":\"application_1564971615273_38182\"}
 ```
 
 #### 1.4.2 命令行参数选项
@@ -149,6 +149,11 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
 	* 描述：per_job 模式下指定的yarn queue
 	* 必选：否
 	* 默认值：false	
+	
+* **yarnSessionConf**
+	* 描述：yarn session 模式下指定的运行的一些参数，[可参考](https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/cli.html),目前只支持指定yid
+	* 必选：否
+	* 默认值：false	
 
 ## 2 结构
 ### 2.1 源表插件

core/src/main/java/com/dtstack/flink/sql/option/Options.java

@@ -68,6 +68,9 @@ public class Options {
     @OptionRequired(description = "yarn queue")
     private String queue = "default";
 
+    @OptionRequired(description = "yarn session configuration,such as yid")
+    private String yarnSessionConf = "{}";
+
     public String getMode() {
         return mode;
     }
@@ -171,4 +174,12 @@ public String getQueue() {
     public void setQueue(String queue) {
         this.queue = queue;
     }
+
+    public String getYarnSessionConf() {
+        return yarnSessionConf;
+    }
+
+    public void setYarnSessionConf(String yarnSessionConf) {
+        this.yarnSessionConf = yarnSessionConf;
+    }
 }

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

@@ -29,9 +29,12 @@
 import org.apache.calcite.sql.SqlJoin;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlOrderBy;
 import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.SqlWith;
+import org.apache.calcite.sql.SqlWithItem;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
 import org.apache.calcite.sql.parser.SqlParserPos;
@@ -70,6 +73,17 @@ public Queue<Object> getExeQueue(String exeSql, Set<String> sideTableSet) throws
     private Object parseSql(SqlNode sqlNode, Set<String> sideTableSet, Queue<Object> queueInfo){
         SqlKind sqlKind = sqlNode.getKind();
         switch (sqlKind){
+            case WITH: {
+                SqlWith sqlWith = (SqlWith) sqlNode;
+                SqlNodeList sqlNodeList = sqlWith.withList;
+                for (SqlNode withAsTable : sqlNodeList) {
+                    SqlWithItem sqlWithItem = (SqlWithItem) withAsTable;
+                    parseSql(sqlWithItem.query, sideTableSet, queueInfo);
+                    queueInfo.add(sqlWithItem);
+                }
+                parseSql(sqlWith.body, sideTableSet, queueInfo);
+                break;
+            }
             case INSERT:
                 SqlNode sqlSource = ((SqlInsert)sqlNode).getSource();
                 return parseSql(sqlSource, sideTableSet, queueInfo);

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

@@ -39,6 +39,7 @@
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlOrderBy;
 import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.SqlWithItem;
 import org.apache.calcite.sql.fun.SqlCase;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParserPos;
@@ -120,6 +121,11 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
                     Table table = tableEnv.sqlQuery(aliasInfo.getName());
                     tableEnv.registerTable(aliasInfo.getAlias(), table);
                     localTableCache.put(aliasInfo.getAlias(), table);
+                } else if (pollSqlNode.getKind() == WITH_ITEM) {
+                    SqlWithItem sqlWithItem = (SqlWithItem) pollSqlNode;
+                    String TableAlias = sqlWithItem.name.toString();
+                    Table table = tableEnv.sqlQuery(sqlWithItem.query.toString());
+                    tableEnv.registerTable(TableAlias, table);
                 }
 
             }else if (pollObj instanceof JoinInfo){

launcher/src/main/java/com/dtstack/flink/sql/launcher/ClusterClientFactory.java

@@ -51,6 +51,8 @@
 
 import com.dtstack.flink.sql.enums.ClusterMode;
 import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.util.StringHelper;
+
 import java.io.IOException;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -113,7 +115,15 @@ public static ClusterClient createYarnClient(Options launcherOptions, String mod
                     ApplicationId applicationId = null;
                     ClusterClient clusterClient = null;
                     if(mode.equals(ClusterMode.yarn.name())) {//on yarn cluster mode
-                        applicationId = getYarnClusterApplicationId(yarnClient);
+                        String yarnSessionConf = launcherOptions.getYarnSessionConf();
+                        yarnSessionConf = URLDecoder.decode(yarnSessionConf, Charsets.UTF_8.toString());
+                        Properties yarnSessionConfProperties = PluginUtil.jsonStrToObject(yarnSessionConf, Properties.class);
+                        String yid = yarnSessionConfProperties.get("yid").toString();
+                        if(StringUtils.isNotBlank(yid)){
+                            applicationId = toApplicationId(yid);
+                        }else{
+                            applicationId = getYarnClusterApplicationId(yarnClient);
+                        }
                         System.out.println("applicationId="+applicationId.toString());
 
                         AbstractYarnClusterDescriptor clusterDescriptor = new YarnClusterDescriptor(
@@ -249,4 +259,21 @@ private static org.apache.hadoop.conf.Configuration haYarnConf(org.apache.hadoop
         return yarnConf;
     }
 
+    private static ApplicationId toApplicationId(String appIdStr) {
+        Iterator<String> it = StringHelper._split(appIdStr).iterator();
+        if (!(it.next()).equals("application")) {
+            throw new IllegalArgumentException("Invalid ApplicationId prefix: " + appIdStr + ". The valid ApplicationId should start with prefix " + "application");
+        } else {
+            try {
+                return toApplicationId(it);
+            } catch (NumberFormatException e) {
+                throw new IllegalArgumentException("Invalid AppAttemptId: " + appIdStr, e);
+            }
+        }
+    }
+
+    private static ApplicationId toApplicationId(Iterator<String> it) throws NumberFormatException {
+        return ApplicationId.newInstance(Long.parseLong(it.next()), Integer.parseInt(it.next()));
+    }
+
 }