Merge remote-tracking branch 'origin/1.5_v3.7.1' into 1.5_v3.7.2

Author: todd5167

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/parser/InsertSqlParser.java

@@ -26,6 +26,7 @@
 import org.apache.calcite.sql.SqlJoin;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOrderBy;
 import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
@@ -132,6 +133,10 @@ private static void parseNode(SqlNode sqlNode, SqlParseResult sqlParseResult){
                     parseNode(unionRight, sqlParseResult);
                 }
                 break;
+            case ORDER_BY:
+                SqlOrderBy sqlOrderBy  = (SqlOrderBy) sqlNode;
+                parseNode(sqlOrderBy.query, sqlParseResult);
+                break;
             default:
                 //do nothing
                 break;

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

@@ -31,6 +31,7 @@
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlOrderBy;
 import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
@@ -110,6 +111,20 @@ 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;
+
+            case ORDER_BY:
+                SqlOrderBy sqlOrderBy  = (SqlOrderBy) sqlNode;
+                parseSql(sqlOrderBy.query, sideTableSet, queueInfo);
         }
 
         return "";

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

@@ -35,6 +35,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.SqlSelect;
 import org.apache.calcite.sql.fun.SqlCase;
 import org.apache.calcite.sql.parser.SqlParseException;
@@ -240,11 +241,50 @@ private void replaceFieldName(SqlNode sqlNode, HashBasedTable<String, String, St
                 }
 
                 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;
+
+            case ORDER_BY:
+                SqlOrderBy sqlOrderBy  = (SqlOrderBy) sqlNode;
+                replaceFieldName(sqlOrderBy.query, mappingTable, targetTableName, tableAlias);
+                SqlNodeList orderFiledList = sqlOrderBy.orderList;
+                for (int i=0 ;i<orderFiledList.size();i++) {
+                    SqlNode replaceNode = replaceOrderByTableName(orderFiledList.get(i), tableAlias);
+                    orderFiledList.set(i, replaceNode);
+                }
             default:
                 break;
         }
     }
 
+    private SqlNode replaceOrderByTableName(SqlNode orderNode, String tableAlias) {
+        if(orderNode.getKind() == IDENTIFIER){
+            SqlIdentifier sqlIdentifier = (SqlIdentifier) orderNode;
+            if (sqlIdentifier.names.size() == 1) {
+                return orderNode;
+            }
+            return sqlIdentifier.setName(0, tableAlias);
+        } else if (orderNode instanceof  SqlBasicCall) {
+            SqlBasicCall sqlBasicCall = (SqlBasicCall) orderNode;
+            for(int i=0; i<sqlBasicCall.getOperandList().size(); i++){
+                SqlNode sqlNode = sqlBasicCall.getOperandList().get(i);
+                sqlBasicCall.getOperands()[i] = replaceOrderByTableName(sqlNode , tableAlias);
+            }
+            return sqlBasicCall;
+        } else {
+            return orderNode;
+        }
+    }
+
     private SqlNode replaceNodeInfo(SqlNode groupNode, HashBasedTable<String, String, String> mappingTable, String tableAlias){
         if(groupNode.getKind() == IDENTIFIER){
             SqlIdentifier sqlIdentifier = (SqlIdentifier) groupNode;
@@ -359,7 +399,7 @@ private SqlNode replaceSelectFieldName(SqlNode selectNode, HashBasedTable<String
             SqlIdentifier sqlIdentifier = (SqlIdentifier) selectNode;
 
             if(sqlIdentifier.names.size() == 1){
-                return null;
+                return selectNode;
             }
 
             String mappingFieldName = mappingTable.get(sqlIdentifier.getComponent(0).getSimple(), sqlIdentifier.getComponent(1).getSimple());

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

core/src/main/java/com/dtstack/flink/sql/util/JDBCUtils.java

@@ -0,0 +1,34 @@
+package com.dtstack.flink.sql.util;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.DriverManager;
+
+public class JDBCUtils {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ClassUtil.class);
+
+    public final static String lock_str = "jdbc_lock_str";
+
+    public static void forName(String clazz, ClassLoader classLoader)  {
+        synchronized (lock_str){
+            try {
+                Class.forName(clazz, true, classLoader);
+                DriverManager.setLoginTimeout(10);
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
+
+    public synchronized static void forName(String clazz) {
+        try {
+            Class<?> driverClass = Class.forName(clazz);
+            driverClass.newInstance();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+}