重构 sql语句解析逻辑；

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

Author: xuchao

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

@@ -57,6 +57,7 @@ public void parseSql(String sql, SqlTree sqlTree) {
                 .configBuilder()
                 .setLex(Lex.MYSQL)
                 .build();
+
         SqlParser sqlParser = SqlParser.create(sql,config);
         SqlNode sqlNode = null;
         try {

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

@@ -731,9 +731,9 @@ private boolean checkAndRemoveWhereCondition(Set<String> fromTableNameSet,
             Set<String> conditionRefTableNameSet = Sets.newHashSet();
 
             fieldInfos.forEach(fieldInfo -> {
-                String[] splitInfo = StringUtils.split(fieldInfo, ",");
+                String[] splitInfo = StringUtils.split(fieldInfo, ".");
                 if(splitInfo.length == 2){
-                    conditionRefTableNameSet.add(splitInfo[1]);
+                    conditionRefTableNameSet.add(splitInfo[0]);
                 }
             });
 

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

@@ -55,6 +55,8 @@ public abstract class SideInfo implements Serializable{
 
     protected String sideSelectFields = "";
 
+    protected Map<Integer, String> sideSelectFieldsType = Maps.newHashMap();
+
     protected JoinType joinType;
 
     //key:Returns the value of the position, value: the ref field index​in the input table
@@ -84,15 +86,17 @@ public void parseSelectFields(JoinInfo joinInfo){
         String sideTableName = joinInfo.getSideTableName();
         String nonSideTableName = joinInfo.getNonSideTable();
         List<String> fields = Lists.newArrayList();
+        int sideTableFieldIndex = 0;
 
-        int sideIndex = 0;
         for( int i=0; i<outFieldInfoList.size(); i++){
             FieldInfo fieldInfo = outFieldInfoList.get(i);
             if(fieldInfo.getTable().equalsIgnoreCase(sideTableName)){
-                fields.add(sideTableInfo.getPhysicalFields().getOrDefault(fieldInfo.getFieldName(), fieldInfo.getFieldName()));
-                sideFieldIndex.put(i, sideIndex);
-                sideFieldNameIndex.put(i, fieldInfo.getFieldName());
-                sideIndex++;
+                String sideFieldName = sideTableInfo.getPhysicalFields().getOrDefault(fieldInfo.getFieldName(), fieldInfo.getFieldName());
+                fields.add(sideFieldName);
+                sideSelectFieldsType.put(sideTableFieldIndex, getTargetFieldType(sideFieldName));
+                sideFieldIndex.put(i, sideTableFieldIndex);
+                sideFieldNameIndex.put(i, sideFieldName);
+                sideTableFieldIndex++;
             }else if(fieldInfo.getTable().equalsIgnoreCase(nonSideTableName)){
                 int nonSideIndex = rowTypeInfo.getFieldIndex(fieldInfo.getFieldName());
                 inFieldIndex.put(i, nonSideIndex);
@@ -108,6 +112,16 @@ public void parseSelectFields(JoinInfo joinInfo){
         sideSelectFields = String.join(",", fields);
     }
 
+    public String getTargetFieldType(String fieldName){
+        int fieldIndex = sideTableInfo.getFieldList().indexOf(fieldName);
+        if(fieldIndex == -1){
+            throw new RuntimeException(sideTableInfo.getName() + "can't find field: " + fieldName);
+        }
+
+        return sideTableInfo.getFieldTypes()[fieldIndex];
+    }
+
+
     public void dealOneEqualCon(SqlNode sqlNode, String sideTableName){
         if(!SqlKind.COMPARISON.contains(sqlNode.getKind())){
             throw new RuntimeException("not compare operator.");
@@ -255,4 +269,16 @@ public Map<Integer, String> getSideFieldNameIndex() {
     public void setSideFieldNameIndex(Map<Integer, String> sideFieldNameIndex) {
         this.sideFieldNameIndex = sideFieldNameIndex;
     }
+
+    public Map<Integer, String> getSideSelectFieldsType() {
+        return sideSelectFieldsType;
+    }
+
+    public void setSideSelectFieldsType(Map<Integer, String> sideSelectFieldsType) {
+        this.sideSelectFieldsType = sideSelectFieldsType;
+    }
+
+    public String getSelectSideFieldType(int index){
+        return sideSelectFieldsType.get(index);
+    }
 }

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

@@ -394,7 +394,6 @@ private void joinFun(Object pollObj,
         HashBasedTable<String, String, String> mappingTable = ((JoinInfo) pollObj).getTableFieldRef();
 
         //获取两个表的所有字段
-        //TODO 抽取
         List<FieldInfo> sideJoinFieldInfo = ParserJoinField.getRowTypeInfo(joinInfo.getSelectNode(), joinScope, true);
         //通过join的查询字段信息过滤出需要的字段信息
         sideJoinFieldInfo.removeIf(tmpFieldInfo -> mappingTable.get(tmpFieldInfo.getTable(), tmpFieldInfo.getFieldName()) == null);

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

@@ -26,6 +26,7 @@
 import com.google.common.collect.HashBasedTable;
 import com.google.common.collect.HashBiMap;
 import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
 import com.typesafe.config.ConfigException;
 import org.apache.calcite.sql.SqlAsOperator;
 import org.apache.calcite.sql.SqlBasicCall;
@@ -200,7 +201,14 @@ public static SqlBasicCall buildAsNodeByJoinInfo(JoinInfo joinInfo, SqlNode sqlN
         String joinLeftTableAlias = joinInfo.getLeftTableAlias();
         joinLeftTableName = Strings.isNullOrEmpty(joinLeftTableName) ? joinLeftTableAlias : joinLeftTableName;
         String newTableName = buildInternalTableName(joinLeftTableName, SPLIT, joinInfo.getRightTableName());
-        String newTableAlias = !StringUtils.isEmpty(tableAlias) ? tableAlias : buildInternalTableName(joinInfo.getLeftTableAlias(), SPLIT, joinInfo.getRightTableAlias());
+        String lefTbAlias = joinInfo.getLeftTableAlias();
+        if(Strings.isNullOrEmpty(lefTbAlias)){
+            Set<String> fromTableSet = Sets.newHashSet();
+            TableUtils.getFromTableInfo(joinInfo.getLeftNode(), fromTableSet);
+            lefTbAlias = StringUtils.join(fromTableSet, "_");
+        }
+
+        String newTableAlias = !StringUtils.isEmpty(tableAlias) ? tableAlias : buildInternalTableName(lefTbAlias, SPLIT, joinInfo.getRightTableAlias());
 
         if (null == sqlNode0) {
             sqlNode0 = new SqlIdentifier(newTableName, null, sqlParserPos);

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java

@@ -201,7 +201,6 @@ protected List<Row> getRows(Row inputRow, List<JsonArray> cacheContent, List<Jso
     public Row fillData(Row input, Object line) {
         JsonArray jsonArray = (JsonArray) line;
         Row row = new Row(sideInfo.getOutFieldInfoList().size());
-        String[] fields = sideInfo.getSideTableInfo().getFieldTypes();
         for (Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()) {
             Object obj = input.getField(entry.getValue());
             boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
@@ -216,7 +215,8 @@ public Row fillData(Row input, Object line) {
             if (jsonArray == null) {
                 row.setField(entry.getKey(), null);
             } else {
-                Object object = SwitchUtil.getTarget(jsonArray.getValue(entry.getValue()), fields[entry.getValue()]);
+                String fieldType = sideInfo.getSelectSideFieldType(entry.getValue());
+                Object object = SwitchUtil.getTarget(jsonArray.getValue(entry.getValue()), fieldType);
                 row.setField(entry.getKey(), object);
             }
         }