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

@@ -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;

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());