Do FlinkSQL jobs that have Left joins have to sink into an upsert-kafka connector (or connector which understands change events)? We're seeing null values in our sink stream. As I was investigating, I changed our sink to be a kafka connector and that's when I received the following error:

org.apache.flink.table.api.TableException: Table sink 'default_catalog.default_database.enriched_topic' doesn't support consuming update and delete changes which is produced by node Join(joinType=[LeftOuterJoin], where=[((stringid = sitenameid)

Is it true that Flink SQL stream with a left join can produce delete events?

You can use the kafka

upsert

connector. I suppose you see null values because the matching records is missing on the right side? It emits insert, update and delete events in order to track changes. For example since its a dual-stream join, you might consume events from the left side but the matching record hasnt consumed yet for the right side. when the record with the matching key arrives from the right side, it will produce delete the previous record and insert the new one

Thanks @Giannis Polyzos. Why would it need a delete record? From what I understood, upsert connector just treats events that already exist as updates. We do use the upsert connector, I was just trying to understand why it would produce delete events. This is the description of the upsert connector:

"More precisely, the value in a data record is interpreted as an UPDATE of the last value for the same key, if any (if a corresponding key doesn’t exist yet, the update will be considered an INSERT). Using the table analogy, a data record in a changelog stream is interpreted as an UPSERT aka INSERT/UPDATE because any existing row with the same key is overwritten. Also, null values are interpreted in a special way: a record with a null value represents a “DELETE”.

Given the row is "overwritten", I don't see where delete makes sense here. If the right hand side doesn't yet exist, it will yield a record. when the right hand side exists, it will yield the record with those columns (and it will be an update, given the key already exists). But delete is still confusing to me. Can you elaborate a bit more. What am I missing? Thank you for your time.

Basically a delete record is like an update.. when the right doesn't exist it will produce for a given key a value with i.e [key1, lv1,lv2,null,null], when the value on the right side exist it will need to update into kafka that value - let's assume [key1, lv1, lv2, rv2, rv2]. Flink provides the rowtypes of [+Insert, -Update Before, +Update After, Delete]. So basically it needs to sent a delete in order to expire/delete the previous row (with the nulls) and update it with the current existing value. Does this help?

@Giannis Polyzos It does. Thanks for the explanation.

Yeah maybe my explanation was also poor. Basically Flink can have 3 different types of streams: • append-only streams: for insert only operations • retract streams: that can handle INSERT, UPDATE BEFORE, UPDATE AFTER and DELETE Operations • Upsert Streams: that can handle INSERT (if the keys doesn't exist), UPDATE and DELETE operations. In a jdbc sink for example an upsert stream will only insert (if the key doesn't exist), or update the row if it exists. same happens with kafka, but what i might made confusing in my previous explanation is that the upsert connector can also write DELETE data as Kafka messages with null values in order to mark them ready for deletion (i.e add a tombstone to that record), in order to expire previous message values.

@Giannis Polyzos Thank you! In a FlinkSQL query, where you have inner and left joins. I imagined that when a right side of the left join doesn't exist, it would emit a record and then, when the right side is available, emit another record for the same key. But we're seeing tombstone records. Why would it want to retract the previous record vs. just updating it?

Kafka is an immutable log right? this means that its not able to delete message, everything gets appended. In the case of an INNER join when there isn't a matching key yet records will be buffered until there is a matching key. In the case of the LEFT JOIN a record will be emitted with NULL values when the left side is not present and will be updated when a matching one arrives. On the kafka side now the way to "update" the records is to mark it with a tombstone for deletion. Basically if you have <key1, v1> and gets an update like <k1,v2> in order to delete the previous value (because the log is immutable) it issues a tombstone and the next time kafka performs a cleanup, it knows it can delete the previous records for that key. Basically it boils down at how Kafka works with updates.

@Giannis Polyzos. Wow, thanks again. We're learning a lot here. That makes total sense. The reason for this confusion was more because we use the upsert kafka connector as a way to sink to the database. So in my mind kafka was just a supporting role on the way to destination. Kafka of course doesn't support updates, so it makes sense for it to issue a retraction there. My concern is because the kafka topic has retractions, we can no longer really do upserts from that topic into a sink (i.e. Postgres) right? since that connector will delete and insert records, not perform updates. So the only way to support real upserts in say a RDBMS, is to write to it directly from the FlinkSQL?

It's not that you can't, its more about semantics and SLAs I would say. Basically you need to "wait" until your data is available (for example the left join will emit the result immediately even if the left side is not there yet). Eventually when it arrives it will be reflected in the database as well and you will have the latest value. Using a dual-stream join for example (regular join) will buffer both sides until there is a matching record. Another approach you can try if the records don't take too long to arrive is mini-batch so instead of emitting reach record one by one, it can buffer results and emit few records. For example if some keys are updated in a timely interval and the value arrives and the left join value gets updated, it could emit fewer results with the updated value. Not sure what the end goal is, but apache paimon (prev. flink table store) is another interesting project to explore, with strong upsert functionality, that allows querying intermediate results and creating a streaming datalakehouse on cheap storage like s3.

@Giannis Polyzos - i'm also slightly confused about this topic. the compaction would work anyway on the key thus the tombstone for a "-U" seems unnecessary. At least that should work for <key1, v1> => <k1,v2> updates I'm sure I'm missing something but can't really grasp what.

the upsert connector doesn't produce a

-U

record

according to this it does https://www.ververica.com/blog/streaming-modes-of-flink-kafka-connectors haven't checked in the record kinds myself but we see the tombstones for every value change at the moment. maybe important to know that we also consume from s kafka-upsert source.

can you point me to the reference?

in the article above there is the title Kafka Connector - Upsert Mode - right below we can see a gif

apologies this is an old thread and i thought it was about the upsert connector as a sink. as a sink it writes INSERT/UPDATE_AFTER and DELETE messages

that's what i got from the FLIP too yes - nevertheless i'm left wondering why every update of a value leads to a tombstone then - since if the compaction is activated, it could simply compact on the key instead of writing tombstone. was that a design decision?