TomRenish
New Contributor III

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!