Add with support

Author: harbby

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

@@ -29,8 +29,11 @@
 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.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;
@@ -66,6 +69,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

@@ -33,6 +33,7 @@
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlNodeList;
 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;
@@ -101,6 +102,11 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
                     Table table = tableEnv.sql(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){