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

xuchao · xuchao · commit 8e2e91d6307f · 2020-03-22T22:42:52.000+08:00
字段替换和重命名相关的逻辑在解析阶段完成；
目标：经过解析sql之后的所有sql是确定的。
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/JoinInfo.java b/core/src/main/java/com/dtstack/flink/sql/side/JoinInfo.java
@@ -21,7 +21,6 @@
 package com.dtstack.flink.sql.side;
 
 import com.google.common.collect.HashBasedTable;
-import com.google.common.collect.HashBiMap;
 import com.google.common.collect.Maps;
 import org.apache.calcite.sql.JoinType;
 import org.apache.calcite.sql.SqlNode;
@@ -34,7 +33,6 @@
  * Join信息
  * Date: 2018/7/24
  * Company: www.dtstack.com
- *
  * @author xuchao
  */
 
@@ -45,8 +43,6 @@ public class JoinInfo implements Serializable {
     //左表是否是维表
     private boolean leftIsSideTable;
 
-    private boolean leftIsTmpTable = false;
-
     //右表是否是维表
     private boolean rightIsSideTable;
 
@@ -70,8 +66,6 @@ public class JoinInfo implements Serializable {
 
     private JoinType joinType;
 
-    private HashBiMap<String, String> fieldRefInfo = HashBiMap.create();
-
     /**
      * 左表需要查询的字段信息和output的时候对应的列名称
      */
@@ -210,22 +204,6 @@ public void setJoinType(JoinType joinType) {
         this.joinType = joinType;
     }
 
-    public boolean isLeftIsTmpTable() {
-        return leftIsTmpTable;
-    }
-
-    public void setLeftIsTmpTable(boolean leftIsTmpTable) {
-        this.leftIsTmpTable = leftIsTmpTable;
-    }
-
-    public HashBiMap<String, String> getFieldRefInfo() {
-        return fieldRefInfo;
-    }
-
-    public void setFieldRefInfo(HashBiMap<String, String> fieldRefInfo) {
-        this.fieldRefInfo = fieldRefInfo;
-    }
-
     public Map<String, String> getLeftSelectFieldInfo() {
         return leftSelectFieldInfo;
     }
@@ -259,7 +237,6 @@ public HashBasedTable<String, String, String> getTableFieldRef(){
     public String toString() {
         return "JoinInfo{" +
                 "leftIsSideTable=" + leftIsSideTable +
-                ", leftIsTmpTable=" + leftIsTmpTable +
                 ", rightIsSideTable=" + rightIsSideTable +
                 ", leftTableName='" + leftTableName + '\'' +
                 ", leftTableAlias='" + leftTableAlias + '\'' +
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/JoinNodeDealer.java b/core/src/main/java/com/dtstack/flink/sql/side/JoinNodeDealer.java
@@ -81,12 +81,15 @@ public JoinNodeDealer(SideSQLParser sideSQLParser){
      * @param tableRef 存储构建临时表查询后源表和新表之间的关联关系
      * @return
      */
-    public JoinInfo dealJoinNode(SqlJoin joinNode, Set<String> sideTableSet,
-                                 Queue<Object> queueInfo, SqlNode parentWhere,
+    public JoinInfo dealJoinNode(SqlJoin joinNode,
+                                 Set<String> sideTableSet,
+                                 Queue<Object> queueInfo,
+                                 SqlNode parentWhere,
                                  SqlNodeList parentSelectList,
                                  Set<Tuple2<String, String>> joinFieldSet,
                                  Map<String, String> tableRef,
                                  Map<String, String> fieldRef) {
+
         SqlNode leftNode = joinNode.getLeft();
         SqlNode rightNode = joinNode.getRight();
         JoinType joinType = joinNode.getJoinType();
@@ -95,8 +98,8 @@ public JoinInfo dealJoinNode(SqlJoin joinNode, Set<String> sideTableSet,
         String leftTbAlias = "";
         String rightTableName = "";
         String rightTableAlias = "";
-        boolean leftTbisTmp = false;
-        //TODO 这个是啥东西
+
+        //TODO 含义需要更明确
         HashBiMap<String, String> fieldReplaceRef = HashBiMap.create();
 
         //如果是连续join 判断是否已经处理过添加到执行队列
@@ -109,14 +112,12 @@ public JoinInfo dealJoinNode(SqlJoin joinNode, Set<String> sideTableSet,
             dealNestJoin(joinNode, sideTableSet,
                     queueInfo, parentWhere, parentSelectList, joinFieldSet, tableRef, fieldRef, parentSelectList, fieldReplaceRef);
             leftNode = joinNode.getLeft();
-
         }
 
         if (leftNode.getKind() == AS) {
             AliasInfo aliasInfo = (AliasInfo) sideSQLParser.parseSql(leftNode, sideTableSet, queueInfo, parentWhere, parentSelectList);
             leftTbName = aliasInfo.getName();
             leftTbAlias = aliasInfo.getAlias();
-
         }
 
         boolean leftIsSide = checkIsSideTable(leftTbName, sideTableSet);
@@ -146,69 +147,20 @@ public JoinInfo dealJoinNode(SqlJoin joinNode, Set<String> sideTableSet,
             tableInfo.setRightTableAlias(rightTableAlias);
         }
 
-
-        tableInfo.setLeftIsTmpTable(leftTbisTmp);
         tableInfo.setLeftIsSideTable(leftIsSide);
         tableInfo.setRightIsSideTable(rightIsSide);
         tableInfo.setLeftNode(leftNode);
         tableInfo.setRightNode(rightNode);
         tableInfo.setJoinType(joinType);
-        tableInfo.setFieldRefInfo(fieldReplaceRef);
 
         tableInfo.setCondition(joinNode.getCondition());
         TableUtils.replaceJoinFieldRefTableName(joinNode.getCondition(), fieldRef);
 
         //extract 需要查询的字段信息
-        //TODO 抽取===========================
         if(rightIsSide){
-            Set<String> fromTableNameSet = Sets.newHashSet();
-            TableUtils.getFromTableInfo(leftNode, fromTableNameSet);
-            Set<String> extractCondition = Sets.newHashSet();
-            extractWhereCondition(fromTableNameSet, (SqlBasicCall) parentWhere, extractCondition);
-            Set<String> extractSelectField = extractSelectFields(parentSelectList, fromTableNameSet, tableRef);
-            Set<String> fieldFromJoinCondition = extractSelectFieldFromJoinCondition(joinFieldSet, fromTableNameSet, tableRef);
-
-            extractSelectField.addAll(extractCondition);
-            extractSelectField.addAll(fieldFromJoinCondition);
-
-            Set<String> rightFromTableNameSet = Sets.newHashSet();
-            TableUtils.getFromTableInfo(rightNode, rightFromTableNameSet);
-            Set<String> extractRightCondition = Sets.newHashSet();
-            extractWhereCondition(rightFromTableNameSet, (SqlBasicCall) parentWhere, extractRightCondition);
-            Set<String> rightExtractSelectField = extractSelectFields(parentSelectList, rightFromTableNameSet, tableRef);
-            Set<String> rightFieldFromJoinCondition = extractSelectFieldFromJoinCondition(joinFieldSet, rightFromTableNameSet, tableRef);
-            rightExtractSelectField.addAll(extractRightCondition);
-            rightExtractSelectField.addAll(rightFieldFromJoinCondition);
-
-            //重命名right 中和 left 重名的
-            Map<String, String> leftTbSelectField = Maps.newHashMap();
-            Map<String, String> rightTbSelectField = Maps.newHashMap();
-            String newTableName = tableInfo.getNewTableAlias();
-
-            for(String tmpField : extractSelectField){
-                String[] tmpFieldSplit = StringUtils.split(tmpField, '.');
-                leftTbSelectField.put(tmpFieldSplit[1], tmpFieldSplit[1]);
-                fieldRef.put(tmpField, TableUtils.buildTableField(newTableName, tmpFieldSplit[1]));
-            }
-
-            for(String tmpField : rightExtractSelectField){
-                String[] tmpFieldSplit = StringUtils.split(tmpField, '.');
-                String originalFieldName = tmpFieldSplit[1];
-                String targetFieldName = originalFieldName;
-                if(leftTbSelectField.containsKey(originalFieldName)){
-                    targetFieldName = ParseUtils.dealDuplicateFieldName(leftTbSelectField, originalFieldName);
-                }
-
-                rightTbSelectField.put(originalFieldName, targetFieldName);
-                fieldRef.put(tmpField, TableUtils.buildTableField(newTableName, targetFieldName));
-            }
-
-            tableInfo.setLeftSelectFieldInfo(leftTbSelectField);
-            tableInfo.setRightSelectFieldInfo(rightTbSelectField);
-
+            extractJoinNeedSelectField(leftNode, rightNode, parentWhere, parentSelectList, tableRef, joinFieldSet, fieldRef, tableInfo);
         }
 
-        //=========================================
         if(tableInfo.getLeftNode().getKind() != AS){
             return tableInfo;
         }
@@ -222,19 +174,80 @@ public JoinInfo dealJoinNode(SqlJoin joinNode, Set<String> sideTableSet,
         return tableInfo;
     }
 
+    /**
+     * 获取join 之后需要查询的字段信息
+     */
+    public void extractJoinNeedSelectField(SqlNode leftNode,
+                                           SqlNode rightNode,
+                                           SqlNode parentWhere,
+                                           SqlNodeList parentSelectList,
+                                           Map<String, String> tableRef,
+                                           Set<Tuple2<String, String>> joinFieldSet,
+                                           Map<String, String> fieldRef,
+                                           JoinInfo tableInfo){
+        Set<String> fromTableNameSet = Sets.newHashSet();
+        TableUtils.getFromTableInfo(leftNode, fromTableNameSet);
+        Set<String> extractCondition = Sets.newHashSet();
+        extractWhereCondition(fromTableNameSet, (SqlBasicCall) parentWhere, extractCondition);
+        Set<String> extractSelectField = extractSelectFields(parentSelectList, fromTableNameSet, tableRef);
+        Set<String> fieldFromJoinCondition = extractSelectFieldFromJoinCondition(joinFieldSet, fromTableNameSet, tableRef);
+
+        extractSelectField.addAll(extractCondition);
+        extractSelectField.addAll(fieldFromJoinCondition);
+
+        Set<String> rightFromTableNameSet = Sets.newHashSet();
+        TableUtils.getFromTableInfo(rightNode, rightFromTableNameSet);
+        Set<String> extractRightCondition = Sets.newHashSet();
+        extractWhereCondition(rightFromTableNameSet, (SqlBasicCall) parentWhere, extractRightCondition);
+        Set<String> rightExtractSelectField = extractSelectFields(parentSelectList, rightFromTableNameSet, tableRef);
+        Set<String> rightFieldFromJoinCondition = extractSelectFieldFromJoinCondition(joinFieldSet, rightFromTableNameSet, tableRef);
+        rightExtractSelectField.addAll(extractRightCondition);
+        rightExtractSelectField.addAll(rightFieldFromJoinCondition);
+
+        //重命名right 中和 left 重名的
+        Map<String, String> leftTbSelectField = Maps.newHashMap();
+        Map<String, String> rightTbSelectField = Maps.newHashMap();
+        String newTableName = tableInfo.getNewTableAlias();
+
+        for(String tmpField : extractSelectField){
+            String[] tmpFieldSplit = StringUtils.split(tmpField, '.');
+            leftTbSelectField.put(tmpFieldSplit[1], tmpFieldSplit[1]);
+            fieldRef.put(tmpField, TableUtils.buildTableField(newTableName, tmpFieldSplit[1]));
+        }
+
+        for(String tmpField : rightExtractSelectField){
+            String[] tmpFieldSplit = StringUtils.split(tmpField, '.');
+            String originalFieldName = tmpFieldSplit[1];
+            String targetFieldName = originalFieldName;
+            if(leftTbSelectField.containsKey(originalFieldName)){
+                targetFieldName = ParseUtils.dealDuplicateFieldName(leftTbSelectField, originalFieldName);
+            }
+
+            rightTbSelectField.put(originalFieldName, targetFieldName);
+            fieldRef.put(tmpField, TableUtils.buildTableField(newTableName, targetFieldName));
+        }
+
+        tableInfo.setLeftSelectFieldInfo(leftTbSelectField);
+        tableInfo.setRightSelectFieldInfo(rightTbSelectField);
+    }
+
 
     /**
      * 处理多层join
-     * 判断左节点是否需要创建零时查询
+     * 判断左节点是否需要创建临时查询
      * （1）右节点是维表
      * （2）左节点不是 as 节点
      */
-    private JoinInfo dealNestJoin(SqlJoin joinNode, Set<String> sideTableSet,
-                                                   Queue<Object> queueInfo, SqlNode parentWhere,
-                                                   SqlNodeList selectList, Set<Tuple2<String, String>> joinFieldSet,
-                                                   Map<String, String> tableRef, Map<String, String> fieldRef,
-                                                   SqlNodeList parentSelectList,
-                                                   HashBiMap<String, String> fieldReplaceRef){
+    private JoinInfo dealNestJoin(SqlJoin joinNode,
+                                  Set<String> sideTableSet,
+                                  Queue<Object> queueInfo,
+                                  SqlNode parentWhere,
+                                  SqlNodeList selectList,
+                                  Set<Tuple2<String, String>> joinFieldSet,
+                                  Map<String, String> tableRef,
+                                  Map<String, String> fieldRef,
+                                  SqlNodeList parentSelectList,
+                                  HashBiMap<String, String> fieldReplaceRef){
 
         SqlJoin leftJoinNode = (SqlJoin) joinNode.getLeft();
         SqlNode parentRightJoinNode = joinNode.getRight();
@@ -249,7 +262,6 @@ private JoinInfo dealNestJoin(SqlJoin joinNode, Set<String> sideTableSet,
         boolean rightIsSide = checkIsSideTable(rightTableName, sideTableSet);
         SqlBasicCall buildAs = TableUtils.buildAsNodeByJoinInfo(joinInfo, null, null);
 
-        //TODO 抽取相同代码
         if(rightIsSide){
             addSideInfoToExeQueue(queueInfo, joinInfo, joinNode, parentSelectList, parentWhere, fieldReplaceRef, tableRef);
         }
@@ -262,37 +274,17 @@ private JoinInfo dealNestJoin(SqlJoin joinNode, Set<String> sideTableSet,
             extractTemporaryQuery(newLeftNode, leftTbAlias, (SqlBasicCall) parentWhere,
                     parentSelectList, queueInfo, joinFieldSet, tableRef, fieldRef);
 
-            //记录表之间的关联关系
-            String newLeftTableName = buildAs.getOperands()[1].toString();
-            Set<String> fromTableNameSet = Sets.newHashSet();
-            TableUtils.getFromTableInfo(newLeftNode, fromTableNameSet);
-            for(String tbTmp : fromTableNameSet){
-                tableRef.put(tbTmp, newLeftTableName);
-            }
-
             //替换leftNode 为新的查询
             joinNode.setLeft(buildAs);
-
-            //替换select field 中的对应字段
-            for(SqlNode sqlNode : parentSelectList.getList()){
-                for(String tbTmp : fromTableNameSet) {
-                    //TODO
-                    TableUtils.replaceSelectFieldTable(sqlNode, tbTmp, newLeftTableName, fieldReplaceRef);
-                }
-            }
-
-            //替换where 中的条件相关
-            for(String tbTmp : fromTableNameSet){
-                TableUtils.replaceWhereCondition(parentWhere, tbTmp, newLeftTableName);
-            }
-
+            replaceSelectAndWhereField(buildAs, leftJoinNode, tableRef, parentSelectList, parentWhere, fieldReplaceRef);
         }
 
         return joinInfo;
     }
 
     /**
      * 右边表是维表需要重新构建左表的临时查询
+     * 并将joinInfo 添加到执行队列里面
      * @param queueInfo
      * @param joinInfo
      * @param joinNode
@@ -319,15 +311,34 @@ public void addSideInfoToExeQueue(Queue<Object> queueInfo,
         //替换左表为新的表名称
         joinNode.setLeft(buildAs);
 
+        replaceSelectAndWhereField(buildAs, leftJoinNode, tableRef, parentSelectList, parentWhere, fieldReplaceRef);
+    }
+
+    /**
+     * 替换指定的查询和条件节点中的字段为新的字段
+     * @param buildAs
+     * @param leftJoinNode
+     * @param tableRef
+     * @param parentSelectList
+     * @param parentWhere
+     * @param fieldReplaceRef
+     */
+    public void replaceSelectAndWhereField(SqlBasicCall buildAs,
+                   SqlNode leftJoinNode,
+                   Map<String, String> tableRef,
+                   SqlNodeList parentSelectList,
+                   SqlNode parentWhere,
+                   HashBiMap<String, String> fieldReplaceRef){
+
         String newLeftTableName = buildAs.getOperands()[1].toString();
         Set<String> fromTableNameSet = Sets.newHashSet();
         TableUtils.getFromTableInfo(leftJoinNode, fromTableNameSet);
+
         for(String tbTmp : fromTableNameSet){
             tableRef.put(tbTmp, newLeftTableName);
         }
 
         //替换select field 中的对应字段
-        //TODO
         for(SqlNode sqlNode : parentSelectList.getList()){
             for(String tbTmp : fromTableNameSet) {
                 TableUtils.replaceSelectFieldTable(sqlNode, tbTmp, newLeftTableName, fieldReplaceRef);
@@ -351,8 +362,10 @@ public void addSideInfoToExeQueue(Queue<Object> queueInfo,
      * @param tableRef
      * @return 源自段和新生成字段之间的映射关系
      */
-    private void extractTemporaryQuery(SqlNode node, String tableAlias, SqlBasicCall parentWhere,
-                                       SqlNodeList parentSelectList, Queue<Object> queueInfo,
+    private void extractTemporaryQuery(SqlNode node, String tableAlias,
+                                       SqlBasicCall parentWhere,
+                                       SqlNodeList parentSelectList,
+                                       Queue<Object> queueInfo,
                                        Set<Tuple2<String, String>> joinFieldSet,
                                        Map<String, String> tableRef,
                                        Map<String, String> fieldRef){
@@ -389,7 +402,6 @@ private void extractTemporaryQuery(SqlNode node, String tableAlias, SqlBasicCall
             queueInfo.offer(sqlBasicCall);
 
             //替换select中的表结构
-            //TODO
             HashBiMap<String, String> fieldReplaceRef = HashBiMap.create();
             for(SqlNode tmpSelect : parentSelectList.getList()){
                 for(String tbTmp : fromTableNameSet) {
diff --git a/core/src/main/java/com/dtstack/flink/sql/util/TableUtils.java b/core/src/main/java/com/dtstack/flink/sql/util/TableUtils.java
@@ -297,6 +297,8 @@ public static void getFromTableInfo(SqlNode fromTable, Set<String> tableNameSet)
 
     /**
      * 替换select 中的字段信息
+     * 如果mappingTable 非空则从该参数获取字段的映射
+     * 如果mappingTable 为空则根据是否存在新生成字段
      * @param selectNode
      * @param oldTbName
      * @param newTbName
