It's amazing how therapeutic it can be to write out your problems. After replying here, the solution came to me. The secret sauce is in getting everything done *before* you run the dlt.apply_changes() engine. After that, all bets are off because the engine seemingly stops worrying about tracking CDC. So before you run apply changes...
make a simple table that takes in only your source data's primary key, or make one via concats as necessary. Then do something like this:
#########################################################
##define the target table's IDs
@dlt.table(comment = "Target for CDC ingestion.",
schema="""
sk_workorder BIGINT GENERATED ALWAYS AS IDENTITY (START WITH 1 INCREMENT BY 1),
workorder_id STRING
"""
)
def stg_workorder_ids():
return (dlt.read_stream("stg_workorders_hist_bronze")
.select('workorder_id')
.distinct()
)
#########################################################
This gets you a 1:1 relationship between your proprietary key and the SK (surrogate key). Once I built this, I joined it back to my history_silver table, *then* passed that through the apply_changes engine. Seems to work. GL!
