Merge remote-tracking branch 'origin/1.5_v3.8.0_beta_1.0' into 1.5_v3.8.0

todd5167 · todd5167 · commit 96ac939fb2c4 · 2019-09-25T14:35:44.000+08:00
# Conflicts:
#	core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java
#	rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/RetractJDBCOutputFormat.java
diff --git a/core/src/main/java/com/dtstack/flink/sql/Main.java b/core/src/main/java/com/dtstack/flink/sql/Main.java
@@ -22,7 +22,12 @@
 
 import com.dtstack.flink.sql.classloader.DtClassLoader;
 import com.dtstack.flink.sql.enums.ECacheType;
-import com.dtstack.flink.sql.parser.*;
+import com.dtstack.flink.sql.exec.FlinkSQLExec;
+import com.dtstack.flink.sql.parser.CreateFuncParser;
+import com.dtstack.flink.sql.parser.CreateTmpTableParser;
+import com.dtstack.flink.sql.parser.InsertSqlParser;
+import com.dtstack.flink.sql.parser.SqlParser;
+import com.dtstack.flink.sql.parser.SqlTree;
 import com.dtstack.flink.sql.side.SideSqlExec;
 import com.dtstack.flink.sql.side.SideTableInfo;
 import com.dtstack.flink.sql.table.SourceTableInfo;
@@ -204,7 +209,7 @@ public static void main(String[] args) throws Exception {
                         //sql-dimensional table contains the dimension table of execution
                         sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache);
                     }else{
-                        tableEnv.sqlUpdate(result.getExecSql());
+                        FlinkSQLExec.sqlUpdate(tableEnv, result.getExecSql());
                         if(LOG.isInfoEnabled()){
                             LOG.info("exec sql: " + result.getExecSql());
                         }
diff --git a/core/src/main/java/com/dtstack/flink/sql/exec/FlinkSQLExec.java b/core/src/main/java/com/dtstack/flink/sql/exec/FlinkSQLExec.java
@@ -0,0 +1,59 @@
+package com.dtstack.flink.sql.exec;
+
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlInsert;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.calcite.FlinkPlannerImpl;
+import org.apache.flink.table.plan.logical.LogicalRelNode;
+import org.apache.flink.table.plan.schema.TableSinkTable;
+
+import java.lang.reflect.Method;
+
+/**
+ * @description:  mapping by name when insert into sink table
+ * @author: maqi
+ * @create: 2019/08/15 11:09
+ */
+public class FlinkSQLExec {
+
+    public static void sqlUpdate(StreamTableEnvironment tableEnv, String stmt) throws Exception {
+
+        FlinkPlannerImpl planner = new FlinkPlannerImpl(tableEnv.getFrameworkConfig(), tableEnv.getPlanner(), tableEnv.getTypeFactory());
+        SqlNode insert = planner.parse(stmt);
+
+        if (!(insert instanceof SqlInsert)) {
+            throw new TableException(
+                    "Unsupported SQL query! sqlUpdate() only accepts SQL statements of type INSERT.");
+        }
+        SqlNode query = ((SqlInsert) insert).getSource();
+
+        SqlNode validatedQuery = planner.validate(query);
+
+        Table queryResult = new Table(tableEnv, new LogicalRelNode(planner.rel(validatedQuery).rel));
+        String targetTableName = ((SqlIdentifier) ((SqlInsert) insert).getTargetTable()).names.get(0);
+
+        Method method = TableEnvironment.class.getDeclaredMethod("getTable", String.class);
+        method.setAccessible(true);
+
+        TableSinkTable targetTable = (TableSinkTable) method.invoke(tableEnv, targetTableName);
+        String[] fieldNames = targetTable.tableSink().getFieldNames();
+
+        Table newTable = null;
+
+        try {
+            newTable = queryResult.select(String.join(",", fieldNames));
+        } catch (Exception e) {
+            throw new ValidationException(
+                    "Field name of query result and registered TableSink "+targetTableName +" do not match.\n" +
+                    "Query result schema: " + String.join(",", queryResult.getSchema().getColumnNames()) + "\n" +
+                    "TableSink schema: " + String.join(",", fieldNames));
+        }
+
+        tableEnv.insertInto(newTable, targetTableName, tableEnv.queryConfig());
+    }
+}
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/SideInfo.java b/core/src/main/java/com/dtstack/flink/sql/side/SideInfo.java
@@ -89,7 +89,7 @@ public void parseSelectFields(JoinInfo joinInfo){
         for( int i=0; i<outFieldInfoList.size(); i++){
             FieldInfo fieldInfo = outFieldInfoList.get(i);
             if(fieldInfo.getTable().equalsIgnoreCase(sideTableName)){
-                fields.add(fieldInfo.getFieldName());
+                fields.add(sideTableInfo.getPhysicalFields().getOrDefault(fieldInfo.getFieldName(), fieldInfo.getFieldName()));
                 sideFieldIndex.put(i, sideIndex);
                 sideFieldNameIndex.put(i, fieldInfo.getFieldName());
                 sideIndex++;
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java b/core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java
@@ -21,11 +21,13 @@
 package com.dtstack.flink.sql.side;
 
 import com.dtstack.flink.sql.enums.ECacheType;
+import com.dtstack.flink.sql.exec.FlinkSQLExec;
 import com.dtstack.flink.sql.parser.CreateTmpTableParser;
 import com.dtstack.flink.sql.side.operator.SideAsyncOperator;
 import com.dtstack.flink.sql.side.operator.SideWithAllCacheOperator;
 import com.dtstack.flink.sql.util.ClassUtil;
 import com.dtstack.flink.sql.util.ParseUtils;
+import org.apache.calcite.sql.SqlAsOperator;
 import org.apache.calcite.sql.SqlBasicCall;
 import org.apache.calcite.sql.SqlDataTypeSpec;
 import org.apache.calcite.sql.SqlIdentifier;
@@ -35,7 +37,7 @@
 import org.apache.calcite.sql.SqlLiteral;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlNodeList;
-import org.apache.calcite.sql.SqlOrderBy;
+import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.fun.SqlCase;
 import org.apache.calcite.sql.parser.SqlParseException;
@@ -100,13 +102,16 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
 
                 if(preIsSideJoin){
                     preIsSideJoin = false;
+                    List<String> fieldNames = null;
                     for(FieldReplaceInfo replaceInfo : replaceInfoList){
+                        fieldNames = Lists.newArrayList();
                         replaceFieldName(pollSqlNode, replaceInfo.getMappingTable(), replaceInfo.getTargetTableName(), replaceInfo.getTargetTableAlias());
+                        addAliasForFieldNode(pollSqlNode, fieldNames, replaceInfo.getMappingTable());
                     }
                 }
 
                 if(pollSqlNode.getKind() == INSERT){
-                    tableEnv.sqlUpdate(pollSqlNode.toString());
+                    FlinkSQLExec.sqlUpdate(tableEnv, pollSqlNode.toString());
                     if(LOG.isInfoEnabled()){
                         LOG.info("exec sql: " + pollSqlNode.toString());
                     }
@@ -125,6 +130,68 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
 
     }
 
+
+    private void addAliasForFieldNode(SqlNode pollSqlNode, List<String> fieldList, HashBasedTable<String, String, String> mappingTable) {
+        SqlKind sqlKind = pollSqlNode.getKind();
+        switch (sqlKind) {
+            case INSERT:
+                SqlNode source = ((SqlInsert) pollSqlNode).getSource();
+                addAliasForFieldNode(source, fieldList, mappingTable);
+                break;
+
+            case AS:
+                addAliasForFieldNode(((SqlBasicCall) pollSqlNode).getOperands()[0], fieldList, mappingTable);
+                break;
+
+            case SELECT:
+
+                SqlNodeList selectList = ((SqlSelect) pollSqlNode).getSelectList();
+
+                selectList.getList().forEach(node -> {
+                    if (node.getKind() == IDENTIFIER) {
+                        SqlIdentifier sqlIdentifier = (SqlIdentifier) node;
+                        if (sqlIdentifier.names.size() == 1) {
+                            return;
+                        }
+                        // save real field
+                        String fieldName = sqlIdentifier.names.get(1);
+                        if (!fieldName.endsWith("0") || fieldName.endsWith("0") && mappingTable.columnMap().containsKey(fieldName)) {
+                            fieldList.add(fieldName);
+                        }
+
+                    }
+                });
+
+                for (int i = 0; i < selectList.getList().size(); i++) {
+                    SqlNode node = selectList.get(i);
+                    if (node.getKind() == IDENTIFIER) {
+                        SqlIdentifier sqlIdentifier = (SqlIdentifier) node;
+                        if (sqlIdentifier.names.size() == 1) {
+                            return;
+                        }
+
+                        String name = sqlIdentifier.names.get(1);
+                        // avoid real field pv0 convert pv
+                        if (name.endsWith("0") &&  !fieldList.contains(name) && !fieldList.contains(name.substring(0, name.length() - 1))) {
+                            SqlOperator operator = new SqlAsOperator();
+                            SqlParserPos sqlParserPos = new SqlParserPos(0, 0);
+
+                            SqlIdentifier sqlIdentifierAlias = new SqlIdentifier(name.substring(0, name.length() - 1), null, sqlParserPos);
+                            SqlNode[] sqlNodes = new SqlNode[2];
+                            sqlNodes[0] = sqlIdentifier;
+                            sqlNodes[1] = sqlIdentifierAlias;
+                            SqlBasicCall sqlBasicCall = new SqlBasicCall(operator, sqlNodes, sqlParserPos);
+
+                            selectList.set(i, sqlBasicCall);
+                        }
+
+                    }
+                }
+                break;
+        }
+    }
+
+
     public AliasInfo parseASNode(SqlNode sqlNode) throws SqlParseException {
         SqlKind sqlKind = sqlNode.getKind();
         if(sqlKind != AS){
@@ -305,7 +372,7 @@ private SqlNode replaceNodeInfo(SqlNode groupNode, HashBasedTable<String, String
         }
     }
 
-    public SqlNode filterNodeWithTargetName(SqlNode sqlNode, String targetTableName){
+    public SqlNode filterNodeWithTargetName(SqlNode sqlNode, String targetTableName) {
 
         SqlKind sqlKind = sqlNode.getKind();
         switch (sqlKind){
@@ -342,7 +409,7 @@ public SqlNode filterNodeWithTargetName(SqlNode sqlNode, String targetTableName)
     }
 
 
-    public void setLocalSqlPluginPath(String localSqlPluginPath){
+    public void setLocalSqlPluginPath(String localSqlPluginPath) {
         this.localSqlPluginPath = localSqlPluginPath;
     }
 
@@ -386,12 +453,12 @@ private List<SqlNode> replaceSelectStarFieldName(SqlNode selectNode, HashBasedTa
         }
     }
 
-    private SqlNode replaceSelectFieldName(SqlNode selectNode, HashBasedTable<String, String, String> mappingTable, String tableAlias){
-        if(selectNode.getKind() == AS){
-            SqlNode leftNode = ((SqlBasicCall)selectNode).getOperands()[0];
+    private SqlNode replaceSelectFieldName(SqlNode selectNode, HashBasedTable<String, String, String> mappingTable, String tableAlias) {
+        if (selectNode.getKind() == AS) {
+            SqlNode leftNode = ((SqlBasicCall) selectNode).getOperands()[0];
             SqlNode replaceNode = replaceSelectFieldName(leftNode, mappingTable, tableAlias);
-            if(replaceNode != null){
-                ((SqlBasicCall)selectNode).getOperands()[0] = replaceNode;
+            if (replaceNode != null) {
+                ((SqlBasicCall) selectNode).getOperands()[0] = replaceNode;
             }
 
             return selectNode;
@@ -470,15 +537,15 @@ private SqlNode replaceSelectFieldName(SqlNode selectNode, HashBasedTable<String
             for(int i=0; i<whenOperands.size(); i++){
                 SqlNode oneOperand = whenOperands.get(i);
                 SqlNode replaceNode = replaceSelectFieldName(oneOperand, mappingTable, tableAlias);
-                if(replaceNode != null){
+                if (replaceNode != null) {
                     whenOperands.set(i, replaceNode);
                 }
             }
 
             for(int i=0; i<thenOperands.size(); i++){
                 SqlNode oneOperand = thenOperands.get(i);
                 SqlNode replaceNode = replaceSelectFieldName(oneOperand, mappingTable, tableAlias);
-                if(replaceNode != null){
+                if (replaceNode != null) {
                     thenOperands.set(i, replaceNode);
                 }
             }
@@ -495,19 +562,27 @@ private SqlNode replaceSelectFieldName(SqlNode selectNode, HashBasedTable<String
 
     /**
      * Analyzing conditions are very join the dimension tables include all equivalent conditions (i.e., dimension table is the primary key definition
+     *
      * @return
      */
-    private boolean checkJoinCondition(SqlNode conditionNode, String sideTableAlias,  List<String> primaryKeys){
-
-        List<String> conditionFields = getConditionFields(conditionNode, sideTableAlias);
-        if(CollectionUtils.isEqualCollection(conditionFields, primaryKeys)){
+    private boolean checkJoinCondition(SqlNode conditionNode, String sideTableAlias, SideTableInfo sideTableInfo) {
+        List<String> conditionFields = getConditionFields(conditionNode, sideTableAlias, sideTableInfo);
+        if(CollectionUtils.isEqualCollection(conditionFields, convertPrimaryAlias(sideTableInfo))){
             return true;
         }
 
         return false;
     }
 
-    public List<String> getConditionFields(SqlNode conditionNode, String specifyTableName){
+    private List<String> convertPrimaryAlias(SideTableInfo sideTableInfo) {
+        List<String> res = Lists.newArrayList();
+        sideTableInfo.getPrimaryKeys().forEach(field -> {
+            res.add(sideTableInfo.getPhysicalFields().getOrDefault(field, field));
+        });
+        return res;
+    }
+
+    public List<String> getConditionFields(SqlNode conditionNode, String specifyTableName, SideTableInfo sideTableInfo){
         List<SqlNode> sqlNodeList = Lists.newArrayList();
         ParseUtils.parseAnd(conditionNode, sqlNodeList);
         List<String> conditionFields = Lists.newArrayList();
@@ -530,7 +605,7 @@ public List<String> getConditionFields(SqlNode conditionNode, String specifyTabl
             }else{
                 throw new RuntimeException(String.format("side table:%s join condition is wrong", specifyTableName));
             }
-
+            tableCol = sideTableInfo.getPhysicalFields().getOrDefault(tableCol, tableCol);
             conditionFields.add(tableCol);
         }
 
@@ -561,8 +636,11 @@ public void registerTmpTable(CreateTmpTableParser.SqlParserResult result,
 
                 if(preIsSideJoin){
                     preIsSideJoin = false;
-                    for(FieldReplaceInfo replaceInfo : replaceInfoList){
+                    List<String> fieldNames = null;
+                    for (FieldReplaceInfo replaceInfo : replaceInfoList) {
+                        fieldNames = Lists.newArrayList();
                         replaceFieldName(pollSqlNode, replaceInfo.getMappingTable(), replaceInfo.getTargetTableName(), replaceInfo.getTargetTableAlias());
+                        addAliasForFieldNode(pollSqlNode, fieldNames, replaceInfo.getMappingTable());
                     }
                 }
 
@@ -598,6 +676,7 @@ public void registerTmpTable(CreateTmpTableParser.SqlParserResult result,
             }
         }
     }
+
     private void joinFun(Object pollObj, Map<String, Table> localTableCache,
                          Map<String, SideTableInfo> sideTableMap, StreamTableEnvironment tableEnv,
                          List<FieldReplaceInfo> replaceInfoList) throws Exception{
@@ -624,7 +703,7 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
             throw new RuntimeException("can't not find side table:" + joinInfo.getRightTableName());
         }
 
-        if(!checkJoinCondition(joinInfo.getCondition(), joinInfo.getRightTableAlias(), sideTableInfo.getPrimaryKeys())){
+        if(!checkJoinCondition(joinInfo.getCondition(), joinInfo.getRightTableAlias(), sideTableInfo)){
             throw new RuntimeException("ON condition must contain all equal fields!!!");
         }
 
@@ -650,7 +729,7 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
 
         //join side table before keyby ===> Reducing the size of each dimension table cache of async
         if(sideTableInfo.isPartitionedJoin()){
-            List<String> leftJoinColList = getConditionFields(joinInfo.getCondition(), joinInfo.getLeftTableAlias());
+            List<String> leftJoinColList = getConditionFields(joinInfo.getCondition(), joinInfo.getLeftTableAlias(), sideTableInfo);
             String[] leftJoinColArr = new String[leftJoinColList.size()];
             leftJoinColArr = leftJoinColList.toArray(leftJoinColArr);
             adaptStream = adaptStream.keyBy(leftJoinColArr);
@@ -681,12 +760,11 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
         }
     }
 
-    private boolean checkFieldsInfo(CreateTmpTableParser.SqlParserResult result, Table table){
+    private boolean checkFieldsInfo(CreateTmpTableParser.SqlParserResult result, Table table) {
         List<String> fieldNames = new LinkedList<>();
         String fieldsInfo = result.getFieldsInfoStr();
         String[] fields = fieldsInfo.split(",");
-        for (int i=0; i < fields.length; i++)
-        {
+        for (int i = 0; i < fields.length; i++) {
             String[] filed = fields[i].split("\\s");
             if (filed.length < 2 || fields.length != table.getSchema().getColumnNames().length){
                 return false;
diff --git a/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncSideInfo.java b/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncSideInfo.java
@@ -56,7 +56,7 @@ public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
 
         sqlCondition = "select ${selectField} from ${tableName} where ";
         for (int i = 0; i < equalFieldList.size(); i++) {
-            String equalField = equalFieldList.get(i);
+            String equalField = sideTableInfo.getPhysicalFields().getOrDefault(equalFieldList.get(i), equalFieldList.get(i));
 
             sqlCondition += dealLowerFiled(equalField) + "=? ";
             if (i != equalFieldList.size() - 1) {
diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncSideInfo.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncSideInfo.java
@@ -66,7 +66,7 @@ public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
 
         sqlCondition = "select ${selectField} from ${tableName} where ";
         for (int i = 0; i < equalFieldList.size(); i++) {
-            String equalField = equalFieldList.get(i);
+            String equalField = sideTableInfo.getPhysicalFields().getOrDefault(equalFieldList.get(i), equalFieldList.get(i));
 
             sqlCondition += equalField + "=? ";
             if (i != equalFieldList.size() - 1) {
