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

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

Author: xuchao

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

@@ -20,9 +20,14 @@
 
 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;
+
 /**
  * 用于记录转换之后的表和原来表直接字段的关联关系
  * Date: 2018/8/30
@@ -78,12 +83,32 @@ public void setTargetTableAlias(String targetTableAlias) {
      * @param fieldName
      * @return
      */
-    public String getTargetFieldName(String tableName, String fieldName){
+    public String getTargetFieldName(String tableName, String fieldName) {
         String targetFieldName = mappingTable.get(tableName, fieldName);
         if(StringUtils.isNotBlank(targetFieldName)){
             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/JoinInfo.java

@@ -20,6 +20,9 @@
 
 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;
 import com.google.common.base.Strings;
@@ -67,6 +70,18 @@ public class JoinInfo implements Serializable {
 
     private JoinType joinType;
 
+    private HashBiMap<String, String> fieldRefInfo = HashBiMap.create();
+
+    /**
+     * 左表需要查询的字段信息和output的时候对应的列名称
+     */
+    private Map<String, String> leftSelectFieldInfo = Maps.newHashMap();
+
+    /**
+     * 右表需要查询的字段信息和output的时候对应的列名称
+     */
+    private Map<String, String> rightSelectFieldInfo = Maps.newHashMap();
+
     public String getSideTableName(){
         if(leftIsSideTable){
             return leftTableAlias;
@@ -203,6 +218,43 @@ 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;
+    }
+
+    public void setLeftSelectFieldInfo(Map<String, String> leftSelectFieldInfo) {
+        this.leftSelectFieldInfo = leftSelectFieldInfo;
+    }
+
+    public Map<String, String> getRightSelectFieldInfo() {
+        return rightSelectFieldInfo;
+    }
+
+    public void setRightSelectFieldInfo(Map<String, String> rightSelectFieldInfo) {
+        this.rightSelectFieldInfo = rightSelectFieldInfo;
+    }
+
+    public HashBasedTable<String, String, String> getTableFieldRef(){
+        HashBasedTable<String, String, String> mappingTable = HashBasedTable.create();
+        getLeftSelectFieldInfo().forEach((key, value) -> {
+            mappingTable.put(getLeftTableAlias(), key, value);
+        });
+
+        getRightSelectFieldInfo().forEach((key, value) -> {
+            mappingTable.put(getRightTableAlias(), key, value);
+        });
+
+        return mappingTable;
+    }
+
     @Override
     public String toString() {
         return "JoinInfo{" +