duckdb/snapshot.sql

{{- /* snapshot.sql — diff-based snapshot for DuckDB                                    */ -}}
{{- /* Compares staging against active history using amf_hash_primary_key (PK identity) */ -}}
{{- /* and amf_hash_source_row_key (row content). Only touches rows that differ.        */ -}}
{{- /*                                                                                  */ -}}
{{- /* Statement 1 (CLOSE): close active rows that are changed or deleted.              */ -}}
{{- /* Statement 2 (INSERT): insert rows that are new or changed.                       */ -}}
{{- /* When nothing changed: zero writes. Cost proportional to delta, not table size.   */ -}}
{{- /* The CLOSE timestamp is the run start, not the event-time expression: closed rows  */ -}}
{{- /* include DELETES, which have no staging row (and no stg alias is in scope here).   */ -}}
{{- /* Under full the two statements compare staging with all active history in place.  */ -}}
{{- /* Under any other compare scope (ADR-161) they read the compare rows instead, which  */ -}}
{{- /* close only the keys that scope allows: a delta batch never closes an absent key.   */ -}}
{{- if eq .CompareScope "full" -}}
UPDATE "{{ .HistorySchema }}"."{{ .HistoryTable }}" hst
SET    amf_valid_to   = '{{ .RunStartedAt }}'::timestamptz,
       amf_updated_at = NOW()
WHERE  hst.amf_valid_to IS NULL
AND    NOT EXISTS (
    SELECT 1
    FROM   "{{ .StagingSchema }}"."{{ .StagingTable }}" stg
    WHERE  stg.amf_hash_primary_key    = hst.amf_hash_primary_key
    AND    stg.amf_hash_source_row_key = hst.amf_hash_source_row_key
);

INSERT INTO "{{ .HistorySchema }}"."{{ .HistoryTable }}" (
    {{ template "business_col_list" . }}, {{ template "amf_system_cols" . }}
)
SELECT {{ template "business_col_list" . }}, {{ template "amf_system_vals" . }}
FROM   "{{ .StagingSchema }}"."{{ .StagingTable }}" stg
WHERE  NOT EXISTS (
    SELECT 1
    FROM   "{{ .HistorySchema }}"."{{ .HistoryTable }}" hst
    WHERE  hst.amf_hash_primary_key    = stg.amf_hash_primary_key
    AND    hst.amf_hash_source_row_key = stg.amf_hash_source_row_key
    AND    hst.amf_valid_to IS NULL
)
{{- else -}}
{{ template "compare_prepare" . -}}
UPDATE "{{ .HistorySchema }}"."{{ .HistoryTable }}" hst
SET    amf_valid_to   = '{{ .RunStartedAt }}'::timestamptz,
       amf_updated_at = NOW()
FROM   {{ template "compare_rows" . }} stg
WHERE  stg.amf_cdc_action IN (2, 3)
  AND  hst.amf_hash_primary_key = stg.amf_hash_primary_key
  AND  hst.amf_valid_to IS NULL;

INSERT INTO "{{ .HistorySchema }}"."{{ .HistoryTable }}" (
    {{ template "business_col_list" . }}, {{ template "amf_system_cols" . }}
)
SELECT {{ template "business_col_list" . }}, {{ template "amf_system_vals" . }}
FROM   {{ template "compare_rows" . }} stg
WHERE  stg.amf_cdc_action IN (1, 2)
{{- template "compare_release" . }}
{{- end }}