Hi, has there been any changes to state handling with Flink SQL? Anything planned?
I didn't find it at https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sql.html. Recently I ran into problems when trying to restore the state after changes that I thought wouldn't change the execution plan. This is what I did basically: SingleOutputStreamOperator kafkaStream = createKafkaStream(env); registerTable("events1", kafkaStream, tableEnv); Table result1 = tableEnv.sql(sql1); addSink(result1); registerTable("events2", kafkaStream, tableEnv); Table result2 = tableEnv.sql(sql2); addSink(result2); Now, even though I created two different tables from kafkaStream, they end up in the same StreamTableEnvironment and Flink optimizes the plan combining common parts of sql1 & sql2 :) That's of course nice when stream DAG doesn't need to be evolved.. But if I'd want to force a "split" after the kafkaStream, is there any way to do so without having to read from kafka multiple times? If I now add this: registerTable("events3", kafkaStream, tableEnv); Table result3 = tableEnv.sql(sql3); addSink(result3); ..restoring from savepoint fails, because Flink pushes for example filters (WHERE conditions) from this new query into the first common block of the SQL execution plan, and adding another branch to the DAG changing the common block, too. So even though this change looked like I'm only adding something new, it also touched the existing parts of the DAG. I ended up using java API where I define operators explicitly, even setting uid on each operator to ensure savepoint compatibility. I would hope to be able to use SQL and Table API instead, if it was also possible to restore state after this kind of incremental changes. -- Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
