So, I'm completely confused about this behavior in flink SQL. If I do a straight

insert into upsert_result select * from topic_a;

It will publish the messages to the upsert_result. If a record that has already been streamed, changes in topic_a (topic a can be another upsert topic or a cdc connector), it will just add a message to the upsert_results topic (no retraction). Since it's an update. Now, when a query uses a left join, this behavior is different...

insert into upsert_result select ... from topic_a left join topic_b on topic_a.b_id = topic_b.id

Then, if initially the right hand side isn't available, it'll publish a message with null values for the right hand side (as expected), but once that data is available, it will first issue a retraction (null record) and then publish a new message with all the values. This seems weird to me. Why would it behave in one way (real upsert) when a single table changes, but perform retract/insert on outer joins? CC: @Giannis Polyzos

Because the semantics change when you apply a join: it’s not an append stream, but a changelog stream

Why isn't it an upsert stream? Why wouldn't it treat the change caused by resolving the right hand size of the join as an upsert (update) to the previous record which didn't resolve it?

In an unbounded stream with a join, you never know if there's anything coming in from your sources in the future that could impact an earlier made decision. That's why the result is a changelog, to be ready for retractions etc in the future. It also means that every decision that is made at that point in time, requires that state to be kept in Flink.

@Ilya Sterin i don't think you lose semantics, it's more of how different operations are encoded in terms of Flink. The end result should be the same depending on the operation, it's just somewhat tricky to get used to the semantics of handling data streams instead of static data. as @Martijn Visser stated

In an unbounded stream with a join, you never know if there's anything coming in from your sources in the future that could impact an earlier made decision.

now combine that with the downstream system like kafka versus a database with strong upsert support, different operations and downstream sinks need different ways for handling things.

ok, I think it's starting to make more sense now. 1. In the case of join, when the right hand side arrives, it will retract the previous message and create a new one because in that case it's not really an update, rather it's saying, we just got to the right hand side, so forget the previous message, here is the right one. This is where retraction/insert makes sense, it treats it as forget the previous, here is the correct answer in this point in time. 2. When some attribute changes, that can be interpreted as a real update, thus it doesn't issue a retraction, and instead just pushes a new message as an upsert.

glad we could help 😄