Merge remote-tracking branch 'origin/v1.5.0_dev' into v1.8.0_dev

# Conflicts:
#	core/src/main/java/com/dtstack/flink/sql/parser/InsertSqlParser.java

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/parser/InsertSqlParser.java

@@ -21,7 +21,13 @@
 package com.dtstack.flink.sql.parser;
 
 import org.apache.calcite.config.Lex;
-import org.apache.calcite.sql.*;
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlInsert;
+import org.apache.calcite.sql.SqlJoin;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOrderBy;
+import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
 import org.apache.commons.lang3.StringUtils;
@@ -32,7 +38,8 @@
 import static org.apache.calcite.sql.SqlKind.IDENTIFIER;
 
 /**
- * parser flink sql
+ * 解析flink sql
+ * sql 只支持 insert 开头的
  * Date: 2018/6/22
  * Company: www.dtstack.com
  * @author xuchao
@@ -112,10 +119,6 @@ private static void parseNode(SqlNode sqlNode, SqlParseResult sqlParseResult){
                     sqlParseResult.addSourceTable(identifierNode.toString());
                 }
                 break;
-//            case MATCH_RECOGNIZE:
-//                SqlMatchRecognize node = (SqlMatchRecognize) sqlNode;
-//                sqlParseResult.addSourceTable(node.getTableRef().toString());
-//                break;
             case UNION:
                 SqlNode unionLeft = ((SqlBasicCall)sqlNode).getOperands()[0];
                 SqlNode unionRight = ((SqlBasicCall)sqlNode).getOperands()[1];
@@ -130,6 +133,10 @@ private static void parseNode(SqlNode sqlNode, SqlParseResult sqlParseResult){
                     parseNode(unionRight, sqlParseResult);
                 }
                 break;
+            case ORDER_BY:
+                SqlOrderBy sqlOrderBy  = (SqlOrderBy) sqlNode;
+                parseNode(sqlOrderBy.query, sqlParseResult);
+                break;
             default:
                 //do nothing
                 break;

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

@@ -30,6 +30,7 @@
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlOrderBy;
 import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
@@ -119,6 +120,10 @@ private Object parseSql(SqlNode sqlNode, Set<String> sideTableSet, Queue<Object>
                 parseSql(unionRight, sideTableSet, queueInfo);
 
                 break;
+
+            case ORDER_BY:
+                SqlOrderBy sqlOrderBy  = (SqlOrderBy) sqlNode;
+                parseSql(sqlOrderBy.query, sideTableSet, queueInfo);
         }
 
         return "";

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

@@ -37,6 +37,7 @@
 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.fun.SqlCase;
 import org.apache.calcite.sql.parser.SqlParseException;
@@ -317,11 +318,39 @@ private void replaceFieldName(SqlNode sqlNode, HashBasedTable<String, String, St
                 replaceFieldName(unionRight, mappingTable, targetTableName, tableAlias);
 
                 break;
+            case ORDER_BY:
+                SqlOrderBy sqlOrderBy  = (SqlOrderBy) sqlNode;
+                replaceFieldName(sqlOrderBy.query, mappingTable, targetTableName, tableAlias);
+                SqlNodeList orderFiledList = sqlOrderBy.orderList;
+                for (int i=0 ;i<orderFiledList.size();i++) {
+                    SqlNode replaceNode = replaceOrderByTableName(orderFiledList.get(i), tableAlias);
+                    orderFiledList.set(i, replaceNode);
+                }
+
             default:
                 break;
         }
     }
 
+    private SqlNode replaceOrderByTableName(SqlNode orderNode, String tableAlias) {
+        if(orderNode.getKind() == IDENTIFIER){
+            SqlIdentifier sqlIdentifier = (SqlIdentifier) orderNode;
+            if (sqlIdentifier.names.size() == 1) {
+                return orderNode;
+            }
+            return sqlIdentifier.setName(0, tableAlias);
+        } else if (orderNode instanceof  SqlBasicCall) {
+            SqlBasicCall sqlBasicCall = (SqlBasicCall) orderNode;
+            for(int i=0; i<sqlBasicCall.getOperandList().size(); i++){
+                SqlNode sqlNode = sqlBasicCall.getOperandList().get(i);
+                sqlBasicCall.getOperands()[i] = replaceOrderByTableName(sqlNode , tableAlias);
+            }
+            return sqlBasicCall;
+        } else {
+            return orderNode;
+        }
+    }
+
     private SqlNode replaceNodeInfo(SqlNode groupNode, HashBasedTable<String, String, String> mappingTable, String tableAlias){
         if(groupNode.getKind() == IDENTIFIER){
             SqlIdentifier sqlIdentifier = (SqlIdentifier) groupNode;