@LasseL 

1. Enable Change Data Capture (CDC):
Enable CDC before deleting data to ensure Delta tables track inserts, updates, and deletes. This allows downstream pipelines to handle deletions correctly.

ALTER TABLE your_table SET TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true');

2. Delete Old Data:
Delete rows older than one month based on the measurement_timestamp column.

DELETE FROM your_table WHERE measurement_timestamp < current_date() - INTERVAL 1 MONTH;

3. Vacuum the Table:
Cleanup deleted data and free up storage by running a VACUUM command.

VACUUM your_table RETAIN 7 HOURS;

4. Downstream Query Considerations:
Use table_changes to process incremental changes in downstream pipelines.
• Batch Query Example:

SELECT *
FROM table_changes('your_table')
WHERE _change_type IN ('insert', 'update');

• Streaming Query with DLT Example:

@dlt.view
def source_table_changes():
   return spark.readStream.format('delta') \
       .option('readChangeFeed', 'true') \
       .table('your_table') \
       .where("_change_type IN ('insert', 'update')")

dlt.create_streaming_table("downstream_table")
dlt.apply_changes(
   target="downstream_table",
   source="source_table_changes",
   keys=['unique_key'],  # Primary key column(s)
   sequence_by='measurement_timestamp'  # Sequence column for updates
)

Summary

1. Enable CDC on the source table using delta.enableChangeDataFeed.
2. Periodically delete old data and vacuum the source table to manage storage.
3. Use apply_changes or table_changes to process updates, inserts, and deletes into downstream tables.
4. Rely on Delta Live Tables (DLT) to ensure consistency and incremental processing.