[FLINK-18540] Unnecessary retract messages when setIdleStateRetentionTime before converting dynamic table to retract stream

• Type: Improvement
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.10.0
• Component/s: Table SQL / Planner, Table SQL / Runtime
• Labels: auto-deprioritized-critical auto-deprioritized-major auto-deprioritized-minor
• Assignee: Unassigned
• Reporter: WeiqiangSun
• Created: 09/Jul/20 09:49
• Updated: 15/Dec/21 22:38

Description

Unnecessary retract messages when setIdleStateRetentionTime before converting dynamic table to retract stream。  

 

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

EnvironmentSettings es = EnvironmentSettings.newInstance().inStreamingMode().build();
StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(env,es);

DataStream<Integer> ds = env.fromElements(5,4,3,2,1);

// if do not setIdleStateRetentionTime, it will output only one message as expected like '(true,5)' 
// if do setIdleStateRetentionTime, it will output unnecessary retract messages like '(false,5)/(true,5)'
fsTableEnv.getConfig().setIdleStateRetentionTime(Time.hours(1000000),Time.hours(10000000));
fsTableEnv.createTemporaryView("test",ds,"id");
Table test = fsTableEnv.sqlQuery("select max(id) from test");
fsTableEnv.toRetractStream(test,Row.class).printToErr();