Merge pull request #15 from harbby/master

Add `with` support

Author: yangsishu

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){