postgres/scd1.sql

{{- /* scd1.sql — upsert in place; physically delete records no longer in staging */ -}}
{{- /* The leading UPDATE overwrites changed keys, so the run reports them as updated: a  */ -}}
{{- /* driver returns one count per statement, and the upsert alone would report every  */ -}}
{{- /* overwrite as an insert (connectors#240). The upsert then finds those keys equal   */ -}}
{{- /* and only inserts new ones. A key the batch carries twice is left to the upsert,  */ -}}
{{- /* which handles it exactly as before, refusing a second write to one row; when it  */ -}}
{{- /* does write such a key, the row counts as inserted, as every upsert row did.      */ -}}
UPDATE "{{ .HistorySchema }}"."{{ .HistoryTable }}" hst
SET    {{ range $i, $col := .AttributeCols }}{{ if $i }},
       {{ end }}"{{ $col }}" = stg."{{ $col }}"{{ end }},
       amf_run_id = {{ .RunID }},
       amf_inserted_at = NOW(),
       amf_updated_at = NOW(),
       amf_valid_from = {{ template "valid_from_expr" . }},
       amf_hash_source_row_key = stg.amf_hash_source_row_key
FROM   "{{ .StagingSchema }}"."{{ .StagingTable }}" stg
WHERE  hst.amf_hash_primary_key = stg.amf_hash_primary_key
  AND  hst.amf_valid_to IS NULL
  AND  hst.amf_hash_source_row_key IS DISTINCT FROM stg.amf_hash_source_row_key
  AND  stg.amf_hash_primary_key IN (
           SELECT amf_hash_primary_key
           FROM   "{{ .StagingSchema }}"."{{ .StagingTable }}"
           GROUP  BY amf_hash_primary_key
           HAVING count(*) = 1
       );

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
ON CONFLICT (amf_hash_primary_key, amf_valid_to) DO UPDATE
    SET {{ range $i, $col := .AttributeCols }}{{ if $i }},
        {{ end }}"{{ $col }}" = EXCLUDED."{{ $col }}"{{ end }},
        amf_run_id = EXCLUDED.amf_run_id,
        amf_inserted_at = EXCLUDED.amf_inserted_at,
        amf_updated_at = EXCLUDED.amf_updated_at,
        amf_valid_from = EXCLUDED.amf_valid_from,
        amf_hash_source_row_key = EXCLUDED.amf_hash_source_row_key
    WHERE "{{ .HistorySchema }}"."{{ .HistoryTable }}".amf_hash_source_row_key IS DISTINCT FROM EXCLUDED.amf_hash_source_row_key;

{{ template "absent_key_delete" . }}