Merge branch 'hotfix_1.8_3.9.x_21763_mergedDev' into 'v1.8.0_dev'

Hotfix 1.8 3.9.x 21763 merged dev

多表连续join 支持

See merge request !219

Author: todd5167

core/pom.xml

@@ -116,6 +116,12 @@
             <version>${flink.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.12</version>
+        </dependency>
+
     </dependencies>
 
     <build>

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -181,8 +181,12 @@ private static void sqlTranslation(String localSqlPluginPath, StreamTableEnviron
                         //sql-dimensional table contains the dimension table of execution
                         sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig);
                     }else{
+                        System.out.println("----------exec sql without dimension join-----------" );
+                        System.out.println("----------real sql exec is--------------------------");
+                        System.out.println(result.getExecSql());
                         FlinkSQLExec.sqlUpdate(tableEnv, result.getExecSql(), queryConfig);
                         if(LOG.isInfoEnabled()){
+                            System.out.println();
                             LOG.info("exec sql: " + result.getExecSql());
                         }
                     }

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

@@ -21,9 +21,10 @@
 package com.dtstack.flink.sql.side;
 
 import com.google.common.collect.HashBasedTable;
+import org.apache.commons.lang3.StringUtils;
 
 /**
- * Reason:
+ * 用于记录转换之后的表和原来表直接字段的关联关系
  * Date: 2018/8/30
  * Company: www.dtstack.com
  * @author xuchao
@@ -37,6 +38,8 @@ public class FieldReplaceInfo {
 
     private String targetTableAlias = null;
 
+    private FieldReplaceInfo preNode = null;
+
     public void setMappingTable(HashBasedTable<String, String, String> mappingTable) {
         this.mappingTable = mappingTable;
     }
@@ -57,7 +60,39 @@ public String getTargetTableAlias() {
         return targetTableAlias;
     }
 
+    public FieldReplaceInfo getPreNode() {
+        return preNode;
+    }
+
+    public void setPreNode(FieldReplaceInfo preNode) {
+        this.preNode = preNode;
+    }
+
     public void setTargetTableAlias(String targetTableAlias) {
         this.targetTableAlias = targetTableAlias;
     }
+
+    /**
+     * 根据原始的tableName + fieldName 获取转换之后的fieldName
+     * @param tableName
+     * @param fieldName
+     * @return
+     */
+    public String getTargetFieldName(String tableName, String fieldName){
+        String targetFieldName = mappingTable.get(tableName, fieldName);
+        if(StringUtils.isNotBlank(targetFieldName)){
+            return targetFieldName;
+        }
+
+        if(preNode == null){
+            return null;
+        }
+
+        String preNodeTargetFieldName = preNode.getTargetFieldName(tableName, fieldName);
+        if(StringUtils.isBlank(preNodeTargetFieldName)){
+            return null;
+        }
+
+        return mappingTable.get(preNode.getTargetTableAlias(), preNodeTargetFieldName);
+    }
 }

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

@@ -41,8 +41,8 @@ public class JoinInfo implements Serializable {
 
     //左表是否是维表
     private boolean leftIsSideTable;
-    //左表是 转换后的中间表
-    private boolean leftIsMidTable;
+
+    private boolean leftIsTmpTable = false;
 
     //右表是否是维表
     private boolean rightIsSideTable;
@@ -66,8 +66,6 @@ public class JoinInfo implements Serializable {
     private SqlNode selectNode;
 
     private JoinType joinType;
-    // 左边是中间转换表，做表映射关系，给替换属性名称使用
-    private Map<String, String> leftTabMapping;
 
     public String getSideTableName(){
         if(leftIsSideTable){
@@ -92,21 +90,6 @@ public String getNewTableName(){
         return leftStr + "_" + rightTableName;
     }
 
-    public boolean isLeftIsMidTable() {
-        return leftIsMidTable;
-    }
-
-    public void setLeftIsMidTable(boolean leftIsMidTable) {
-        this.leftIsMidTable = leftIsMidTable;
-    }
-
-    public Map<String, String> getLeftTabMapping() {
-        return leftTabMapping;
-    }
-
-    public void setLeftTabMapping(Map<String, String> leftTabMapping) {
-        this.leftTabMapping = leftTabMapping;
-    }
 
     public String getNewTableAlias(){
         return leftTableAlias + "_" + rightTableAlias;
@@ -211,4 +194,12 @@ public JoinType getJoinType() {
     public void setJoinType(JoinType joinType) {
         this.joinType = joinType;
     }
+
+    public boolean isLeftIsTmpTable() {
+        return leftIsTmpTable;
+    }
+
+    public void setLeftIsTmpTable(boolean leftIsTmpTable) {
+        this.leftIsTmpTable = leftIsTmpTable;
+    }
 }

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

@@ -51,21 +51,20 @@ public class ParserJoinField {
      */
     public static List<FieldInfo> getRowTypeInfo(SqlNode sqlNode, JoinScope scope, boolean getAll){
 
-        if(sqlNode.getKind() != SqlKind.SELECT){
-            throw new RuntimeException("------not select node--------\n" + sqlNode.toString());
-        }
-
         List<FieldInfo> fieldInfoList = Lists.newArrayList();
         if(getAll){
             return getAllField(scope);
         }
 
+        if(sqlNode.getKind() != SqlKind.SELECT){
+            throw new RuntimeException("------not select node--------\n" + sqlNode.toString());
+        }
+
         SqlSelect sqlSelect = (SqlSelect)sqlNode;
         SqlNodeList sqlNodeList = sqlSelect.getSelectList();
         for(SqlNode fieldNode : sqlNodeList.getList()){
             SqlIdentifier identifier = (SqlIdentifier)fieldNode;
             if(!identifier.isStar()) {
-                System.out.println(identifier);
                 String tableName = identifier.getComponent(0).getSimple();
                 String fieldName = identifier.getComponent(1).getSimple();
                 TypeInformation<?> type = scope.getFieldType(tableName, fieldName);
@@ -76,7 +75,6 @@ public static List<FieldInfo> getRowTypeInfo(SqlNode sqlNode, JoinScope scope, b
                 fieldInfoList.add(fieldInfo);
             } else {
                 //处理
-                System.out.println("----------");
                 int identifierSize = identifier.names.size();
 
                 switch(identifierSize) {
@@ -108,6 +106,7 @@ public static List<FieldInfo> getRowTypeInfo(SqlNode sqlNode, JoinScope scope, b
         return fieldInfoList;
     }
 
+    //TODO 丢弃多余的PROCTIME
     private static List<FieldInfo> getAllField(JoinScope scope){
         Iterator prefixId = scope.getChildren().iterator();
         List<FieldInfo> fieldInfoList = Lists.newArrayList();