初步重构 sql语句解析逻辑；

字段替换和重命名相关的逻辑在解析阶段完成；
目标：经过解析sql之后的所有sql是确定的。

Author: xuchao

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

@@ -20,13 +20,9 @@
 
 package com.dtstack.flink.sql.side;
 
-import com.dtstack.flink.sql.util.ReflectionUtils;
 import com.google.common.collect.HashBasedTable;
 import org.apache.commons.lang3.StringUtils;
 
-import java.lang.reflect.Field;
-import java.util.HashMap;
-import java.util.Map;
 
 /**
  * 用于记录转换之后的表和原来表直接字段的关联关系
@@ -89,26 +85,6 @@ public String getTargetFieldName(String tableName, String fieldName) {
             return targetFieldName;
         }
 
-
-       /* Field field  = ReflectionUtils.getDeclaredField(mappingTable, "backingMap");
-        field.setAccessible(true);
-        HashMap<String, Map<String, String>> map = null;
-        try {
-            map = (HashMap) field.get(mappingTable);
-        } catch (IllegalAccessException e) {
-            e.printStackTrace();
-        }
-
-        if(map.size() == 1 && tableName.equalsIgnoreCase(targetTableAlias)){
-            for(Map<String, String> tmp : map.values()){
-                targetFieldName = tmp.get(fieldName);
-                if(targetFieldName != null){
-                    return targetFieldName;
-                }
-            }
-        }*/
-
-
         if(preNode == null){
             return null;
         }

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

@@ -277,7 +277,7 @@ private JoinInfo dealNestJoin(SqlJoin joinNode, Set<String> sideTableSet,
             for(SqlNode sqlNode : parentSelectList.getList()){
                 for(String tbTmp : fromTableNameSet) {
                     //TODO
-                    TableUtils.replaceSelectFieldTable(sqlNode, tbTmp, newLeftTableName, fieldReplaceRef, null);
+                    TableUtils.replaceSelectFieldTable(sqlNode, tbTmp, newLeftTableName, fieldReplaceRef);
                 }
             }
 
@@ -330,7 +330,7 @@ public void addSideInfoToExeQueue(Queue<Object> queueInfo,
         //TODO
         for(SqlNode sqlNode : parentSelectList.getList()){
             for(String tbTmp : fromTableNameSet) {
-                TableUtils.replaceSelectFieldTable(sqlNode, tbTmp, newLeftTableName, fieldReplaceRef, null);
+                TableUtils.replaceSelectFieldTable(sqlNode, tbTmp, newLeftTableName, fieldReplaceRef);
             }
         }
 
@@ -393,7 +393,7 @@ private void extractTemporaryQuery(SqlNode node, String tableAlias, SqlBasicCall
             HashBiMap<String, String> fieldReplaceRef = HashBiMap.create();
             for(SqlNode tmpSelect : parentSelectList.getList()){
                 for(String tbTmp : fromTableNameSet) {
-                    TableUtils.replaceSelectFieldTable(tmpSelect, tbTmp, tableAlias, fieldReplaceRef, null);
+                    TableUtils.replaceSelectFieldTable(tmpSelect, tbTmp, tableAlias, fieldReplaceRef);
                 }
             }
 

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

@@ -108,25 +108,11 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
         Queue<Object> exeQueue = sideSQLParser.getExeQueue(sql, sideTableMap.keySet());
         Object pollObj = null;
 
-        //need clean
-        boolean preIsSideJoin = false;
-        List<FieldReplaceInfo> replaceInfoList = Lists.newArrayList();
-
         while((pollObj = exeQueue.poll()) != null){
 
             if(pollObj instanceof SqlNode){
                 SqlNode pollSqlNode = (SqlNode) pollObj;
 
-                if(preIsSideJoin){
-                    preIsSideJoin = false;
-                    List<String> fieldNames = null;
-                    for(FieldReplaceInfo replaceInfo : replaceInfoList){
-                        fieldNames = Lists.newArrayList();
-                        //replaceFieldName(pollSqlNode, replaceInfo);
-                        //TODO
-                        //addAliasForFieldNode(pollSqlNode, fieldNames, replaceInfo.getMappingTable());
-                    }
-                }
 
                 if(pollSqlNode.getKind() == INSERT){
                     System.out.println("----------real exec sql-----------" );
@@ -137,7 +123,7 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
                     }
 
                 }else if(pollSqlNode.getKind() == AS){
-                    dealAsSourceTable(tableEnv, pollSqlNode, tableCache, replaceInfoList);
+                    dealAsSourceTable(tableEnv, pollSqlNode, tableCache);
 
                 } else if (pollSqlNode.getKind() == WITH_ITEM) {
                     SqlWithItem sqlWithItem = (SqlWithItem) pollSqlNode;
@@ -166,8 +152,7 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
             }else if (pollObj instanceof JoinInfo){
                 System.out.println("----------exec join info----------");
                 System.out.println(pollObj.toString());
-                preIsSideJoin = true;
-                joinFun(pollObj, localTableCache, sideTableMap, tableEnv, replaceInfoList);
+                joinFun(pollObj, localTableCache, sideTableMap, tableEnv);
             }
         }
 
@@ -401,8 +386,7 @@ public List<String> getConditionFields(SqlNode conditionNode, String specifyTabl
 
     protected void dealAsSourceTable(StreamTableEnvironment tableEnv,
                                      SqlNode pollSqlNode,
-                                     Map<String, Table> tableCache,
-                                     List<FieldReplaceInfo> replaceInfoList) throws SqlParseException {
+                                     Map<String, Table> tableCache) throws SqlParseException {
 
         AliasInfo aliasInfo = parseASNode(pollSqlNode);
         if (localTableCache.containsKey(aliasInfo.getName())) {
@@ -424,19 +408,13 @@ protected void dealAsSourceTable(StreamTableEnvironment tableEnv,
         Set<String> fromTableNameSet = Sets.newHashSet();
         SqlNode fromNode = ((SqlBasicCall)pollSqlNode).getOperands()[0];
         TableUtils.getFromTableInfo(fromNode, fromTableNameSet);
-        for(FieldReplaceInfo tmp : replaceInfoList){
-            if(fromTableNameSet.contains(tmp.getTargetTableName())
-                    || fromTableNameSet.contains(tmp.getTargetTableAlias())){
-                fieldReplaceInfo.setPreNode(tmp);
-                break;
-            }
-        }
-        replaceInfoList.add(fieldReplaceInfo);
+
     }
 
-    private void joinFun(Object pollObj, Map<String, Table> localTableCache,
-                         Map<String, SideTableInfo> sideTableMap, StreamTableEnvironment tableEnv,
-                         List<FieldReplaceInfo> replaceInfoList) throws Exception{
+    private void joinFun(Object pollObj,
+                         Map<String, Table> localTableCache,
+                         Map<String, SideTableInfo> sideTableMap,
+                         StreamTableEnvironment tableEnv) throws Exception{
         JoinInfo joinInfo = (JoinInfo) pollObj;
 
         JoinScope joinScope = new JoinScope();
@@ -518,17 +496,6 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
         replaceInfo.setTargetTableName(targetTableName);
         replaceInfo.setTargetTableAlias(targetTableAlias);
 
-        //判断之前是不是被替换过,被替换过则设置之前的替换信息作为上一个节点
-        for(FieldReplaceInfo tmp : replaceInfoList){
-            if(tmp.getTargetTableName().equalsIgnoreCase(joinInfo.getLeftTableName())
-            ||tmp.getTargetTableName().equalsIgnoreCase(joinInfo.getLeftTableAlias())){
-                replaceInfo.setPreNode(tmp);
-                break;
-            }
-        }
-
-        replaceInfoList.add(replaceInfo);
-
         if (!tableEnv.isRegistered(joinInfo.getNewTableName())){
             Table joinTable = tableEnv.fromDataStream(dsOut);
             tableEnv.registerTable(joinInfo.getNewTableName(), joinTable);

core/src/main/java/com/dtstack/flink/sql/util/TableUtils.java

@@ -256,13 +256,14 @@ public static void replaceFromNodeForJoin(JoinInfo joinInfo, SqlSelect sqlNode)
         Map<String, String> leftFieldMapping = fieldMapping.row(joinInfo.getLeftTableAlias());
         Map<String, String> rightFieldMapping = fieldMapping.row(joinInfo.getRightTableAlias());
 
-        for(SqlNode oneSelectNode : sqlNode.getSelectList()){
+       /* for(SqlNode oneSelectNode : sqlNode.getSelectList()){
             replaceSelectFieldTable(oneSelectNode, joinInfo.getLeftTableAlias(), newAliasName, null ,leftFieldMapping);
             replaceSelectFieldTable(oneSelectNode, joinInfo.getRightTableAlias(), newAliasName, null , rightFieldMapping);
-        }
+        }*/
 
         //where中的条件属性为新的表名称和字段
-        replaceWhereCondition();
+        FieldReplaceUtil.replaceFieldName(sqlNode, joinInfo.getLeftTableAlias(), newAliasName, leftFieldMapping);
+        FieldReplaceUtil.replaceFieldName(sqlNode, joinInfo.getRightTableAlias(), newAliasName, rightFieldMapping);
         sqlNode.setFrom(sqlBasicCall);
     }
 
@@ -296,22 +297,18 @@ public static void getFromTableInfo(SqlNode fromTable, Set<String> tableNameSet)
 
     /**
      * 替换select 中的字段信息
-     * 如果mappingTable 非空则从该参数获取字段的映射
-     * 如果mappingTable 为空则根据是否存在新生成字段
      * @param selectNode
      * @param oldTbName
      * @param newTbName
      * @param fieldReplaceRef
-     * @param mappingTable
      */
     public static void replaceSelectFieldTable(SqlNode selectNode,
                                                String oldTbName,
                                                String newTbName,
-                                               HashBiMap<String, String> fieldReplaceRef,
-                                               Map<String, String> mappingTable) {
+                                               HashBiMap<String, String> fieldReplaceRef) {
         if (selectNode.getKind() == AS) {
             SqlNode leftNode = ((SqlBasicCall) selectNode).getOperands()[0];
-            replaceSelectFieldTable(leftNode, oldTbName, newTbName, fieldReplaceRef, mappingTable);
+            replaceSelectFieldTable(leftNode, oldTbName, newTbName, fieldReplaceRef);
 
         }else if(selectNode.getKind() == IDENTIFIER){
             SqlIdentifier sqlIdentifier = (SqlIdentifier) selectNode;
@@ -322,7 +319,7 @@ public static void replaceSelectFieldTable(SqlNode selectNode,
 
             String fieldTableName = sqlIdentifier.names.get(0);
             if(oldTbName.equalsIgnoreCase(fieldTableName)){
-                replaceOneSelectField(sqlIdentifier, newTbName, oldTbName, fieldReplaceRef, mappingTable);
+                replaceOneSelectField(sqlIdentifier, newTbName, oldTbName, fieldReplaceRef);
             }
 
         }else if(selectNode.getKind() == LITERAL || selectNode.getKind() == LITERAL_CHAIN){//字面含义
@@ -369,7 +366,7 @@ public static void replaceSelectFieldTable(SqlNode selectNode,
                     continue;
                 }
 
-                replaceSelectFieldTable(sqlNode, oldTbName, newTbName, fieldReplaceRef, mappingTable);
+                replaceSelectFieldTable(sqlNode, oldTbName, newTbName, fieldReplaceRef);
             }
 
         }else if(selectNode.getKind() == CASE){
@@ -380,16 +377,16 @@ public static void replaceSelectFieldTable(SqlNode selectNode,
 
             for(int i=0; i<whenOperands.size(); i++){
                 SqlNode oneOperand = whenOperands.get(i);
-                replaceSelectFieldTable(oneOperand, oldTbName, newTbName, fieldReplaceRef, mappingTable);
+                replaceSelectFieldTable(oneOperand, oldTbName, newTbName, fieldReplaceRef);
             }
 
             for(int i=0; i<thenOperands.size(); i++){
                 SqlNode oneOperand = thenOperands.get(i);
-                replaceSelectFieldTable(oneOperand, oldTbName, newTbName, fieldReplaceRef, mappingTable);
+                replaceSelectFieldTable(oneOperand, oldTbName, newTbName, fieldReplaceRef);
 
             }
 
-            replaceSelectFieldTable(elseNode, oldTbName, newTbName, fieldReplaceRef, mappingTable);
+            replaceSelectFieldTable(elseNode, oldTbName, newTbName, fieldReplaceRef);
         }else if(selectNode.getKind() == OTHER){
             //不处理
             return;
@@ -401,20 +398,11 @@ public static void replaceSelectFieldTable(SqlNode selectNode,
     private static void replaceOneSelectField(SqlIdentifier sqlIdentifier,
                                               String newTbName,
                                               String oldTbName,
-                                              HashBiMap<String, String> fieldReplaceRef,
-                                              Map<String, String> mappingTable){
+                                              HashBiMap<String, String> fieldReplaceRef){
         SqlIdentifier newField = sqlIdentifier.setName(0, newTbName);
         String fieldName = sqlIdentifier.names.get(1);
         String fieldKey = oldTbName + "_" + fieldName;
 
-        if(mappingTable != null){
-            String mappingFieldName = mappingTable.get(fieldName);
-            Preconditions.checkNotNull(mappingFieldName, "can't get any field from mappingTable with oldFieldName " + fieldName);
-            newField = newField.setName(1, mappingFieldName);
-            sqlIdentifier.assignNamesFrom(newField);
-            return;
-        }
-
         if(!fieldReplaceRef.containsKey(fieldKey)){
             if(fieldReplaceRef.inverse().get(fieldName) != null){
                 //换一个名字