Applying Delta Loads to a PostGIS Target Idempotently Jump to heading#

# etl/apply.py — psycopg >=3.1 — Python 3.10+
import psycopg

STAGE_DDL = """
CREATE UNLOGGED TABLE IF NOT EXISTS parcels_stage (
    stable_id      text PRIMARY KEY,
    geom           geometry(Polygon, 25832) NOT NULL,
    land_use_code  text NOT NULL,
    ownership_status text,
    content_hash   text NOT NULL
);
TRUNCATE parcels_stage;
"""


def stage(conn: psycopg.Connection, rows) -> int:
    with conn.cursor() as cur:
        cur.execute(STAGE_DDL)
        with cur.copy(
            "COPY parcels_stage (stable_id, geom, land_use_code, ownership_status, content_hash) "
            "FROM STDIN (FORMAT BINARY)"
        ) as copy:
            for row in rows:
                copy.write_row(row)
        cur.execute("ANALYZE parcels_stage")     # the planner needs stats for the join below
        return cur.rowcount

-- PostgreSQL >= 16, PostGIS >= 3.4 — the whole insert-or-update, in one statement.
INSERT INTO parcels AS target
       (stable_id, geom, land_use_code, ownership_status, content_hash, updated_at, retired_at)
SELECT s.stable_id, s.geom, s.land_use_code, s.ownership_status, s.content_hash, now(), NULL
  FROM parcels_stage s
    ON CONFLICT (stable_id) DO UPDATE
   SET geom             = EXCLUDED.geom,
       land_use_code    = EXCLUDED.land_use_code,
       ownership_status = EXCLUDED.ownership_status,
       content_hash     = EXCLUDED.content_hash,
       updated_at       = now(),
       retired_at       = NULL                     -- resurrect a feature that came back
 WHERE target.content_hash IS DISTINCT FROM EXCLUDED.content_hash
    OR target.retired_at IS NOT NULL;

-- Features present in the target but not in this delivery.
UPDATE parcels t
   SET retired_at = now()
 WHERE t.retired_at IS NULL
   AND NOT EXISTS (SELECT 1 FROM parcels_stage s WHERE s.stable_id = t.stable_id);

CREATE OR REPLACE VIEW parcels_current AS
SELECT * FROM parcels WHERE retired_at IS NULL;

# etl/apply.py — psycopg >=3.1 — Python 3.10+
DELETE_CEILING = 0.05        # 5 % of live features in one run


def apply_delta(conn: psycopg.Connection, rows, policy: dict) -> dict:
    with conn.transaction():                 # everything below commits or none of it does
        with conn.cursor() as cur:
            cur.execute("SET LOCAL lock_timeout = '2s'")
            cur.execute("SET LOCAL statement_timeout = '120s'")

            staged = stage(conn, rows)
            cur.execute("SELECT count(*) FROM parcels WHERE retired_at IS NULL")
            (live_before,) = cur.fetchone()

            cur.execute(UPSERT_SQL)
            upserted = cur.rowcount

            cur.execute(SOFT_DELETE_SQL)
            retired = cur.rowcount

            if live_before and retired / live_before > policy.get("delete_ceiling", DELETE_CEILING):
                raise DeleteStorm(
                    f"{retired} of {live_before} features would be retired "
                    f"({retired / live_before:.1%}) — above the ceiling; aborting"
                )

            cur.execute(LEDGER_INSERT, {"staged": staged, "upserted": upserted,
                                        "retired": retired, "run_id": policy["run_id"]})
    return {"staged": staged, "upserted": upserted, "retired": retired}

CREATE TABLE IF NOT EXISTS delta_run_ledger (
    run_id        text PRIMARY KEY,
    layer         text NOT NULL,
    policy_version text NOT NULL,
    staged        bigint NOT NULL,
    upserted      bigint NOT NULL,
    retired       bigint NOT NULL,
    applied_at    timestamptz NOT NULL DEFAULT now()
);

-- Re-apply the identical delta and confirm nothing moved.
SELECT upserted, retired FROM delta_run_ledger ORDER BY applied_at DESC LIMIT 2;
--  upserted | retired
-- ----------+---------
--         0 |       0      ← second application
--      1918 |     142      ← first application

# tests/test_apply.py — pytest >=7
def test_apply_twice_is_a_no_op(target_db, delta):
    first = apply_delta(target_db, delta, POLICY | {"run_id": "r1"})
    snapshot = dump_table(target_db, "parcels")
    second = apply_delta(target_db, delta, POLICY | {"run_id": "r2"})
    assert second["upserted"] == 0 and second["retired"] == 0
    assert dump_table(target_db, "parcels") == snapshot


def test_delete_storm_rolls_everything_back(target_db, truncated_delta):
    before = dump_table(target_db, "parcels")
    with pytest.raises(DeleteStorm):
        apply_delta(target_db, truncated_delta, POLICY | {"run_id": "r3"})
    assert dump_table(target_db, "parcels") == before      # including the upserts


def test_resurrected_feature_keeps_its_identifier(target_db, delta_with_return):
    apply_delta(target_db, delta_with_return, POLICY | {"run_id": "r4"})
    row = fetch(target_db, "PARCEL-14-023-9")
    assert row["retired_at"] is None