Hey all, I have a more conceptual question on how we could handle some specific use cases with Flink. We have multiple Flink source tables ingested via Kafka in the Debezium changelog format. These tables are keyed, and per key, we are only interested in the latest event (similar to Kafka log compaction). I now want to join these tables to create events that are enriched with all information from these source tables. Let me provide an example:

CREATE TABLE Person (
    Id int,
    Firstname string,
    Lastname string,
    DateOfBirth date,
    primary key (Id) not enforced
) WITH ( ... )

CREATE TABLE Address (
    Id int,
    Street string,
    ZipCode string,
    City string,
    Country string,
    PersonId int,
    primary key (Id) not enforced
) WITH ( ... )

It doesn't matter what gets updated (

Person

or

Address

), for both source triggers I want to emit a new combined

PersonUpdate

event that contains information about the person and its corresponding addresses:

CREATE TABLE PersonUpdate (
    Id int,
    Firstname string,
    Lastname string,
    DateOfBirth date,
    Addresses ARRAY<ROW(Street string, City string, ZipCode string, Country string)>,
    primary key (Id) not enforced
) WITH (...)

One option I read about would be to use temporal joins, having the right side of the join doing lookups on for example the Address table (getting all addresses for a specific person X). However, as far as I understood the semantics of temporal joins, the join wouldn’t trigger if there is only an update on the Address table (e.g., an address is added with PersonId X)? This brings me to the question, whether in such cases I would need to duplicate the same query, but this time doing the temporal lookup on the Person table? But what if I need to join multiple tables to get all the enrichment info for my event? Would I need to duplicate the query multiple times and flip the join directions to set all the required triggers? Another alternative I currently see would be to rely on the DataStream API by connecting all involved source streams, creating KeyedStreams and then do the state work on my own (e.g., in

.process(...)

) using ValueState or MapState etc.. Would that be a better performing or recommended solution compared to the one using temporal joins? Or did I miss any other alternative? What I clearly want to avoid are lookup joins doing the lookup on the database again.

Hey! Regarding your second approach, I gave a talk about that , probably relevant here.

Thanks for sharing the link to the talk, this is definitely helpful! It also proves that we are not the only ones considering keeping state forever for being able to emit events on late updates 🙂 In retrospective, would you still follow that route, or would you consider something else? Savepoint recovery is probably still an open challenge for ever growing states?

The best solution to this problem is changing your source to be a series of denormalized Domain Events with Outbox pattern instead of raw CDC streams. This eliminates the need to perform most of the joins in the first place. This typically means talking to other teams and lots of negotiations, but it’s worth it 🙂