Merge branch 'v1.5.0_dev_union' into 'v1.5.0_dev'

V1.5.0 dev union



See merge request !60

Author: zoudaokoulife

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

@@ -110,6 +110,16 @@ private Object parseSql(SqlNode sqlNode, Set<String> sideTableSet, Queue<Object>
                 aliasInfo.setAlias(alias.toString());
 
                 return aliasInfo;
+
+            case UNION:
+                SqlNode unionLeft = ((SqlBasicCall)sqlNode).getOperands()[0];
+                SqlNode unionRight = ((SqlBasicCall)sqlNode).getOperands()[1];
+
+                parseSql(unionLeft, sideTableSet, queueInfo);
+
+                parseSql(unionRight, sideTableSet, queueInfo);
+
+                break;
         }
 
         return "";

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

@@ -306,6 +306,16 @@ private void replaceFieldName(SqlNode sqlNode, HashBasedTable<String, String, St
                     throw new RuntimeException("---not deal type:" + sqlNode);
                 }
 
+                break;
+            case UNION:
+                SqlNode unionLeft = ((SqlBasicCall) sqlNode).getOperands()[0];
+
+                SqlNode unionRight = ((SqlBasicCall) sqlNode).getOperands()[1];
+
+                replaceFieldName(unionLeft, mappingTable, targetTableName, tableAlias);
+
+                replaceFieldName(unionRight, mappingTable, targetTableName, tableAlias);
+
                 break;
             default:
                 break;
@@ -439,39 +449,33 @@ private SqlNode replaceSelectFieldName(SqlNode selectNode, HashBasedTable<String
             return sqlIdentifier;
         }else if(selectNode.getKind() == LITERAL || selectNode.getKind() == LITERAL_CHAIN){//字面含义
             return selectNode;
-        }else if(selectNode.getKind() == OTHER_FUNCTION
+        }else if(  AGGREGATE.contains(selectNode.getKind())
+                || AVG_AGG_FUNCTIONS.contains(selectNode.getKind())
+                || COMPARISON.contains(selectNode.getKind())
+                || selectNode.getKind() == OTHER_FUNCTION
                 || selectNode.getKind() == DIVIDE
                 || selectNode.getKind() == CAST
-                || selectNode.getKind() == SUM
-                || selectNode.getKind() == AVG
-                || selectNode.getKind() == MAX
-                || selectNode.getKind() == MIN
                 || selectNode.getKind() == TRIM
                 || selectNode.getKind() == TIMES
                 || selectNode.getKind() == PLUS
-                || selectNode.getKind() == IN
+                || selectNode.getKind() == NOT_IN
                 || selectNode.getKind() == OR
                 || selectNode.getKind() == AND
-                || selectNode.getKind() == COUNT
-                || selectNode.getKind() == SUM0
-                || selectNode.getKind() == LEAD
-                || selectNode.getKind() == LAG
-                || selectNode.getKind() == EQUALS
-                || selectNode.getKind() == NOT_EQUALS
                 || selectNode.getKind() == MINUS
                 || selectNode.getKind() == TUMBLE
                 || selectNode.getKind() == TUMBLE_START
                 || selectNode.getKind() == TUMBLE_END
                 || selectNode.getKind() == SESSION
                 || selectNode.getKind() == SESSION_START
                 || selectNode.getKind() == SESSION_END
+                || selectNode.getKind() == HOP
+                || selectNode.getKind() == HOP_START
+                || selectNode.getKind() == HOP_END
                 || selectNode.getKind() == BETWEEN
                 || selectNode.getKind() == IS_NULL
                 || selectNode.getKind() == IS_NOT_NULL
-                || selectNode.getKind() == LESS_THAN
-                || selectNode.getKind() == GREATER_THAN
-                || selectNode.getKind() == LESS_THAN_OR_EQUAL
-                || selectNode.getKind() == GREATER_THAN_OR_EQUAL
+                || selectNode.getKind() == CONTAINS
+
                 ){
             SqlBasicCall sqlBasicCall = (SqlBasicCall) selectNode;
             for(int i=0; i<sqlBasicCall.getOperands().length; i++){

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/RetractJDBCOutputFormat.java

@@ -197,7 +197,8 @@ private void insertWrite(Row row) {
     private void writeSingleRecord(Row row) {
         try {
             updatePreparedStmt(row, upload);
-            upload.execute();
+            upload.executeUpdate();
+            dbConn.commit();
         } catch (SQLException e) {
             outDirtyRecords.inc();
             LOG.error("record insert failed ..", row.toString());