Merge branch '1.8_v3.8.1' into v1.8.0_dev

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/constrant/ConfigConstrant.java

@@ -29,15 +29,21 @@
  */
 public class ConfigConstrant {
 
-    public static final String FLINK_CHECKPOINT_INTERVAL_KEY = "sql.checkpoint.interval";
+    public static final String SQL_CHECKPOINT_INTERVAL_KEY = "sql.checkpoint.interval";
+    // 兼容上层
+    public static final String FLINK_CHECKPOINT_INTERVAL_KEY = "flink.checkpoint.interval";
 
     public static final String FLINK_CHECKPOINT_MODE_KEY = "sql.checkpoint.mode";
 
     public static final String FLINK_CHECKPOINT_TIMEOUT_KEY = "sql.checkpoint.timeout";
 
     public static final String FLINK_MAXCONCURRENTCHECKPOINTS_KEY = "sql.max.concurrent.checkpoints";
 
-    public static final String FLINK_CHECKPOINT_CLEANUPMODE_KEY = "sql.checkpoint.cleanup.mode";
+    public static final String SQL_CHECKPOINT_CLEANUPMODE_KEY = "sql.checkpoint.cleanup.mode";
+
+    public static final String FLINK_CHECKPOINT_CLEANUPMODE_KEY = "flink.checkpoint.cleanup.mode";
+
+
 
     public static final String FLINK_CHECKPOINT_DATAURI_KEY = "flinkCheckpointDataURI";
 

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

@@ -68,12 +68,14 @@ public static void openCheckpoint(StreamExecutionEnvironment env, Properties pro
         }
 
         //设置了时间间隔才表明开启了checkpoint
-        if(properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_INTERVAL_KEY) == null){
+        if(properties.getProperty(ConfigConstrant.SQL_CHECKPOINT_INTERVAL_KEY) == null && properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_INTERVAL_KEY) == null){
             return;
         }else{
-            Long interval = Long.valueOf(properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_INTERVAL_KEY));
+            Long sql_interval = Long.valueOf(properties.getProperty(ConfigConstrant.SQL_CHECKPOINT_INTERVAL_KEY,"0"));
+            Long flink_interval = Long.valueOf(properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_INTERVAL_KEY, "0"));
+            long checkpointInterval = Math.max(sql_interval, flink_interval);
             //start checkpoint every ${interval}
-            env.enableCheckpointing(interval);
+            env.enableCheckpointing(checkpointInterval);
         }
 
         String checkMode = properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_MODE_KEY);
@@ -101,7 +103,14 @@ public static void openCheckpoint(StreamExecutionEnvironment env, Properties pro
             env.getCheckpointConfig().setMaxConcurrentCheckpoints(maxConcurrCheckpoints);
         }
 
-        String cleanupModeStr = properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_CLEANUPMODE_KEY);
+        Boolean sqlCleanMode = MathUtil.getBoolean(properties.getProperty(ConfigConstrant.SQL_CHECKPOINT_CLEANUPMODE_KEY), false);
+        Boolean flinkCleanMode = MathUtil.getBoolean(properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_CLEANUPMODE_KEY), false);
+
+        String cleanupModeStr = "false";
+        if (sqlCleanMode || flinkCleanMode ){
+            cleanupModeStr = "true";
+        }
+
         if ("true".equalsIgnoreCase(cleanupModeStr)){
             env.getCheckpointConfig().enableExternalizedCheckpoints(
                     CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncSideInfo.java

@@ -72,7 +72,7 @@ public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
         for (int i = 0; i < equalFieldList.size(); i++) {
             String equalField = sideTableInfo.getPhysicalFields().getOrDefault(equalFieldList.get(i), equalFieldList.get(i));
 
-            sqlCondition += equalField + "\t" + sqlJoinCompareOperate.get(i) + " ？";
+            sqlCondition += equalField + " " + sqlJoinCompareOperate.get(i) + " ? ";
             if (i != equalFieldList.size() - 1) {
                 sqlCondition += " and ";
             }