[fix-32416][core]add window early trigger feat

Author: simenliuxing

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

@@ -56,6 +56,9 @@ public class ConfigConstrant {
     // default 200ms
     public static final String AUTO_WATERMARK_INTERVAL_KEY = "autoWatermarkInterval";
 
+    // window early trigger
+    public static final String EARLY_TRIGGER = "early.trigger";
+
     public static final String SQL_TTL_MINTIME = "sql.ttl.min";
     public static final String SQL_TTL_MAXTIME = "sql.ttl.max";
 

core/src/main/java/com/dtstack/flink/sql/environment/StreamEnvConfigManager.java

@@ -121,6 +121,21 @@ public static void streamExecutionEnvironmentConfig(StreamExecutionEnvironment s
         }
     }
 
+    /**
+     * 设置TableEnvironment window提前触发
+     * @param tableEnv
+     * @param confProperties
+     */
+    public static void streamTableEnvironmentEarlyTriggerConfig(TableEnvironment tableEnv, Properties confProperties) {
+        confProperties = PropertiesUtils.propertiesTrim(confProperties);
+        String triggerTime = confProperties.getProperty(ConfigConstrant.EARLY_TRIGGER);
+        if (StringUtils.isNumeric(triggerTime)) {
+            TableConfig qConfig = tableEnv.getConfig();
+            qConfig.getConfiguration().setString("table.exec.emit.early-fire.enabled", "true");
+            qConfig.getConfiguration().setString("table.exec.emit.early-fire.delay", triggerTime+"s");
+        }
+    }
+
     /**
      * 设置TableEnvironment状态超时时间
      * @param tableEnv

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -379,6 +379,7 @@ public static StreamTableEnvironment getStreamTableEnv(StreamExecutionEnvironmen
 
         StreamTableEnvironment tableEnv = StreamTableEnvironmentImpl.create(env, settings, tableConfig);
         StreamEnvConfigManager.streamTableEnvironmentStateTTLConfig(tableEnv, confProperties);
+        StreamEnvConfigManager.streamTableEnvironmentEarlyTriggerConfig(tableEnv, confProperties);
         return tableEnv;
     }
 

localTest/src/main/java/com/dtstack/flink/sql/localTest/LocalTest.java

@@ -50,13 +50,14 @@ public static void main(String[] args) throws Exception {
         setLogLevel("INFO");
 
         List<String> propertiesList = new ArrayList<>();
-        String sqlPath = "/Users/wtz/dtstack/job/flinkStreamSQL/sql/TestDemo/JoinDemoTwo.sql";
+        String sqlPath = "/Users/chuixue/Desktop/tmp/sqlFile.sql";
         Map<String, Object> conf = new HashMap<>();
         JSONObject properties = new JSONObject();
 
         //其他参数配置
         properties.put("time.characteristic", "eventTime");
         properties.put("timezone", TimeZone.getDefault());
+        properties.put("early.trigger", "1");
 
         // 任务配置参数
         conf.put("-sql", URLEncoder.encode(readSQL(sqlPath), StandardCharsets.UTF_8.name()));