Recommended approach to process CDC data

[almoehi]

Hi,

I am trying to figure out what the best approach would be for handling postgres CDC data. I currently have it in a append-only raw table (in duckdb) which contains initial snapshot data and subsequent update/insert/delete events - identified by a flag column.

What is the recommended approach to process CDC data with sqlmesh?

Incremental by unique key and when matched ?

I was hoping to use duckdb but it seems like that is not supported with when matched.

Any other suggestions ?

Thanks & Cheers!

[StuffbyYuki]

Sorry I'm late to join this discussion but I think it's simplest and flexible to use incremental by time.

[cmgoffena13]

@almoehi -- you're on the right track.

The INCREMENTAL_BY_UNIQUE_KEY model automatically performs a MERGE on the table. You just want to make sure you have the same grain, so 1 change per record (no duplicate IDs) so the MERGE doesn't fail. I would recommend using a Row Number function to filter your CDC data to always get the last CDC Change record for the row. The MERGE will properly handle INSERT/UPDATEs automatically, and you can just put something like this in your query that updates the table when it matches a record for DELETE:

CASE WHEN cdc_operation = 'DELETE' THEN 0 ELSE 1 END is_active

You can use SQLMesh time macros with the INCREMENTAL_BY_UNIQUE_KEY model, so you can pull only the latest cdc changes from the table as well (like the last days worth if its a daily model).

The only caveat is (I'm guessing) your CDC records probably have all of the fields as NULL on a DELETE. This pattern will update all the fields to NULL if it matches on a DELETE. I don't know a good way around this if you can't use the WHEN MATCHED config unfortunately.

Hope that helps