Merge branch '1.5_3.6.1_union' into '1.5_v3.6.1'

union parse



See merge request !61

Author: zoudaokoulife

core/src/main/java/com/dtstack/flink/sql/metric/MetricConstant.java

@@ -45,6 +45,8 @@ public class MetricConstant {
 
     public static final String DT_NUM_RECORDS_OUT = "dtNumRecordsOut";
 
+    public static final String DT_NUM_DIRTY_RECORDS_OUT = "dtNumDirtyRecordsOut";
+
     public static final String DT_NUM_RECORDS_OUT_RATE = "dtNumRecordsOutRate";
 
     public static final String DT_EVENT_DELAY_GAUGE = "dtEventDelay";

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;

core/src/main/java/com/dtstack/flink/sql/sink/MetricOutputFormat.java

@@ -34,8 +34,11 @@ public abstract  class MetricOutputFormat extends  RichOutputFormat<Tuple2>{
 
      protected transient Meter outRecordsRate;
 
+    protected  transient Counter outDirtyRecords;
+
      public void initMetric() {
         outRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_OUT);
+        outDirtyRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_DIRTY_RECORDS_OUT);
         outRecordsRate = getRuntimeContext().getMetricGroup().meter(MetricConstant.DT_NUM_RECORDS_OUT_RATE, new MeterView(outRecords, 20));
      }
 

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

@@ -49,6 +49,9 @@ public class RetractJDBCOutputFormat extends MetricOutputFormat {
 
     private static final Logger LOG = LoggerFactory.getLogger(RetractJDBCOutputFormat.class);
 
+    private static int dirtyDataPrintFrequency = 1000;
+    private static int receiveDataPrintFrequency = 1000;
+
     private String username;
     private String password;
     private String drivername;
@@ -164,8 +167,11 @@ public void writeRecord(Tuple2 tuple2) {
         }
 
         if (retract) {
-            insertWrite(row);
             outRecords.inc();
+            if (outRecords.getCount() % receiveDataPrintFrequency == 0) {
+                LOG.info("Receive data : {}", row);
+            }
+            insertWrite(row);
         } else {
             //do nothing
         }
@@ -174,7 +180,6 @@ public void writeRecord(Tuple2 tuple2) {
 
 
     private void insertWrite(Row row) {
-        System.out.println("接受到数据row:" +row );
         checkConnectionOpen(dbConn);
         try {
             if (batchInterval == 1) {
@@ -196,20 +201,20 @@ 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);
+            outDirtyRecords.inc();
+            if (outDirtyRecords.getCount() % dirtyDataPrintFrequency == 0) {
+                LOG.error("record insert failed ..", row.toString());
+                LOG.error("", e);
+            }
         }
     }
 
     private synchronized void submitExecuteBatch() {
         try {
-            LOG.info("submitExecuteBatch start......");
             this.upload.executeBatch();
             dbConn.commit();
-            rows.forEach(row -> System.out.println("批量插入成功："+ row));
         } catch (SQLException e) {
             try {
                 dbConn.rollback();