Hello,

I'm creating a Lakeflow Spark declarative pipeline to migrate a process that was previously handled by a custom pipeline. There is one component that is a bit complicated for me: we have a table that incrementally adds records but, at the same time, deletes records based on the data's lifecycle.

The process is actually very simple. We have a historical table with an integer column that expresses the DATE in YYYYMM format. For example, if records have a 2-year lifecycle and the current ingested data is 202604, then we need to DELETE data prior to 202406.

Initially, we tried using sinks with ForEachSink, but the problem is that we can't track dependencies between bronze -> silver -> because the sinks are flow-based and produce tables that are not managed by Databricks. As a result, we cannot guarantee that a bronze process runs before a silver process.

Next, we tried with CDC, which gets new records with an 'INSERT' value in a column. From the previous stage, we got the records to be deleted and masked them with 'DELETE' in the columns. Finally, we used unionByName to get the source table and use it in the CDC, but this failed due to a circular dependency.

There is an alternative of creating a support_table with the keys of the target table to avoid the circular dependency, but I think this is an anti-pattern for LSDP.

What would be a suitable solution for these deletion cases, and how can I execute this task using Lakeflow Spark Declarative Pipelines?