union parse

Author: todd5167

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

@@ -232,6 +232,17 @@ 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;

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

@@ -174,7 +174,6 @@ public void writeRecord(Tuple2 tuple2) {
 
 
     private void insertWrite(Row row) {
-        System.out.println("接受到数据row:" +row );
         checkConnectionOpen(dbConn);
         try {
             if (batchInterval == 1) {
@@ -196,9 +195,8 @@ private void writeSingleRecord(Row row) {
         try {
             updatePreparedStmt(row, upload);
             upload.execute();
-            System.out.println("单条插入成功：" + row);
+            dbConn.commit();
         } catch (SQLException e) {
-            System.out.println("单条插入失败：" + row);
             LOG.error("record insert failed ..", row.toString());
             LOG.error("", e);
         }
@@ -209,7 +207,6 @@ private synchronized void submitExecuteBatch() {
             LOG.info("submitExecuteBatch start......");
             this.upload.executeBatch();
             dbConn.commit();
-            rows.forEach(row -> System.out.println("批量插入成功："+ row));
         } catch (SQLException e) {
             try {
                 dbConn.rollback();