Hello, I have few questions about Flink’s dynamic tables: • Flink’s dynamic tables can be converted into an upsert stream (requires a unique key). ◦ But docs mentioned: “Please note that only append and retract streams are supported when converting a dynamic table into a `DataStream`”. • Flink’s dynamic tables can be written to an external system. ◦ From slide 17 here, looks like we can save ~50% of traffic if downstream system supports upserting. ◦

WITH ('connector'='kafka', 'format' = 'debezium-json')

supports: ▪︎ insertion, update before, update after, and delete ◦ What does the community use as an external system for dynamic tables? • There are query restrictions (state size for continuous queries evaluated on unbounded streams that need to update perv. emitted results.) ◦ Does Flink has to maintain this state even when dynamic table is written to external system? ◦ Any benchmark for state size for different queries? Thanks