This guide is part of Scheduling & Incremental Spatial Loads, within the broader Orchestrating Spatial ETL Pipelines reference.

Change Data Capture for PostGIS Source Tables

A watermark read tells you what appeared or changed. It cannot tell you what disappeared, and for a spatial target that must mirror its source — parcels that get merged, incidents that get retracted — that gap is the difference between a mirror and an accumulation.

Why Timestamps Are Not Enough

  • Deletes are invisible. A removed row leaves no trace a WHERE updated_at > x query can see.
  • Updates lose their history. Only the latest state is readable, so a correction and the value it replaced are indistinguishable.
  • Bulk edits produce identical timestamps. A migration touching a million rows leaves them all at one instant, and a watermark cannot page through them safely.
  • Clock skew moves the boundary. Timestamps set by different clients drift, which is what the overlap window in a watermark read exists to absorb.

Version and Environment Compatibility

Component Version Note
PostgreSQL >=13 Logical replication with pgoutput; generated identity columns
PostGIS >=3.3 ST_AsBinary in triggers, geometry equality operators
SQLAlchemy >=2.0 Connection and transaction handling
GeoPandas >=1.0 Applying the change stream to a target frame 
pip install "sqlalchemy>=2.0" "psycopg[binary]>=3.1" "geopandas>=1.0"

Two Ways to Capture, and What Each Costs

Trigger-based capture versus logical replication Two capture mechanisms. A trigger fires inside the writing transaction and appends a row to an audit table, which costs write latency on the source but needs only ordinary table permissions. Logical replication reads the write-ahead log through a replication slot, adding no latency to writers but requiring elevated privileges and careful slot management, since an unconsumed slot retains WAL until the disk fills. trigger + audit table runs inside the writer's transaction needs only table privileges costs a little write latency audit table needs its own retention the pragmatic default logical replication reads the write-ahead log no latency added to writers needs replication privileges an unconsumed slot fills the disk right at volume, with an owner

The Audit Trigger

CREATE TABLE parcels_changes (
    change_id   bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    changed_at  timestamptz NOT NULL DEFAULT now(),
    operation   char(1)     NOT NULL CHECK (operation IN ('I', 'U', 'D')),
    parcel_id   text        NOT NULL,
    geom_wkb    bytea,                    -- NULL for a delete
    attributes  jsonb
);

CREATE INDEX ON parcels_changes (change_id);

CREATE OR REPLACE FUNCTION capture_parcel_change() RETURNS trigger AS $$
BEGIN
    IF (TG_OP = 'DELETE') THEN
        INSERT INTO parcels_changes (operation, parcel_id, geom_wkb, attributes)
        VALUES ('D', OLD.parcel_id, NULL, to_jsonb(OLD) - 'geom');
        RETURN OLD;
    END IF;

    INSERT INTO parcels_changes (operation, parcel_id, geom_wkb, attributes)
    VALUES (
        CASE TG_OP WHEN 'INSERT' THEN 'I' ELSE 'U' END,
        NEW.parcel_id,
        ST_AsBinary(NEW.geom),
        to_jsonb(NEW) - 'geom'
    );
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER parcels_audit
AFTER INSERT OR UPDATE OR DELETE ON parcels
FOR EACH ROW EXECUTE FUNCTION capture_parcel_change();

Consuming the Change Stream

from __future__ import annotations

import logging

import geopandas as gpd
import pandas as pd
import shapely
import sqlalchemy

logger = logging.getLogger(__name__)


def read_changes(engine: sqlalchemy.Engine, since_change_id: int,
                 batch: int = 100_000) -> tuple[gpd.GeoDataFrame, int]:
    """Read one batch of changes above the last consumed identity value."""
    query = sqlalchemy.text(
        """
        SELECT change_id, operation, parcel_id, geom_wkb, attributes
        FROM parcels_changes
        WHERE change_id > :since
        ORDER BY change_id
        LIMIT :batch
        """
    )
    with engine.connect() as connection:
        frame = pd.read_sql(query, connection, params={"since": since_change_id, "batch": batch})

    if frame.empty:
        return gpd.GeoDataFrame(geometry=[], crs=4326), since_change_id

    geometry = [shapely.from_wkb(value) if value is not None else None
                for value in frame.pop("geom_wkb")]
    changes = gpd.GeoDataFrame(frame, geometry=gpd.GeoSeries(geometry, crs=4326))

    high_water = int(changes["change_id"].max())
    counts = changes["operation"].value_counts().to_dict()
    logger.info("read %d changes up to %d: %s", len(changes), high_water, counts)
    return changes, high_water


def apply_changes(target: gpd.GeoDataFrame, changes: gpd.GeoDataFrame,
                  key: str = "parcel_id") -> gpd.GeoDataFrame:
    """Replay a change batch onto a target frame, in change order."""
    result = target.set_index(key, drop=False)

    for _, change in changes.sort_values("change_id").iterrows():
        identifier = change[key]
        if change["operation"] == "D":
            result = result.drop(index=identifier, errors="ignore")
        else:
            row = {key: identifier, "geometry": change.geometry, **(change["attributes"] or {})}
            result.loc[identifier] = pd.Series(row)

    return gpd.GeoDataFrame(result.reset_index(drop=True), crs=target.crs)

Key Implementation Notes

  • The cursor is an identity value, not a timestamp. A monotonic sequence has no clock skew, no ties and no ambiguity about inclusivity, which removes every boundary problem a timestamp watermark has.
  • Changes are applied in change_id order. Out-of-order application lets an update overtake the insert it depends on, producing a row that briefly exists with the wrong content and sometimes permanently.
  • Deletes carry no geometry. The NULL is deliberate: a delete needs only the key, and storing the removed geometry doubles the log’s size for information the target already has.
  • Attributes go to jsonb minus the geometry column. The change log then survives a source schema change without a migration of its own.
  • The audit table needs its own retention. It grows monotonically and will eventually exceed the table it audits; expire consumed rows on a schedule, keeping a margin beyond the slowest consumer.
  • A single-row apply loop is fine at these volumes and not at all volumes. Above a few hundred thousand changes per batch, collapse the batch to the last operation per key first and apply once.
Collapsing a change batch before applying it A batch holds an insert, an update and a further update for parcel A17, plus a delete for parcel B22. Collapsing to the last operation per key leaves one upsert for A17 and one delete for B22, so the target is written twice rather than four times and the result is identical. raw batch, in change_id order 1041 · I · A17 1042 · U · A17 1043 · U · A17 1044 · D · B22 collapse applied upsert A17 · state at 1043 delete B22 Collapsing also makes the apply idempotent: replaying the same batch produces the same target state regardless of how far the previous attempt got.

Reconciling Against a Snapshot

A change stream drifts. A trigger disabled during a maintenance window, a bulk load that bypassed it, a consumer that skipped a batch — each leaves the target subtly wrong in a way the stream itself cannot reveal.

What a change feed cannot see Four kinds of source operation and whether each produces change rows. Ordinary inserts, updates and deletes are captured. A COPY load fires row triggers in PostgreSQL but is often run with triggers disabled during maintenance. A TRUNCATE fires no row-level trigger at all, so a table emptied that way leaves the feed silent while the target keeps every row. source operation change rows produced INSERT / UPDATE / DELETE yes — the normal path COPY with triggers disabled none — a routine maintenance habit TRUNCATE none — no row-level trigger fires at all trigger dropped in a migration none — and nothing fails to announce it

The remedy is a periodic full comparison, cheap enough to run weekly: count rows on both sides, and compare a checksum aggregated over the key and a geometry hash. Where they disagree, a full reload of the affected partition is usually faster than diagnosing the divergence.

Two failure modes are worth expecting. Bulk operations that use COPY or TRUNCATE bypass row-level triggers entirely, so a source team’s routine maintenance can empty a target silently. And a trigger dropped during a schema migration is rarely restored, because nothing fails when it is missing — the stream simply goes quiet, which is why the reconciliation should alert on an unusually empty batch as well as on a mismatch.

Troubleshooting a CDC Feed

Symptom Likely cause Fix
Target keeps rows the source deleted Trigger missing the DELETE event Attach for INSERT OR UPDATE OR DELETE
Change stream went silent Trigger dropped in a migration Reconcile on a schedule; alert on empty batches
Source writes slowed noticeably Audit insert inside a hot transaction Move to logical replication, or narrow what is captured
Disk filling on the source Unconsumed replication slot retaining WAL Monitor slot lag; drop abandoned slots
Target has a row that never existed Changes applied out of order Sort by change_id before applying
Audit table larger than the source No retention on the change log Expire consumed rows past the slowest consumer

Integration Note

The consumed change_id is a watermark like any other and should be advanced only after the target commits — the ordering argued for in incremental spatial loading with watermark timestamps. Keep the cursor in the target database so that the write and the advance share a transaction, and the whole pipeline becomes exactly-once without any distributed coordination.

Geometry Changes Are Not Like Attribute Changes

A CDC feed treats every update the same, and for spatial data that hides a distinction worth surfacing.

An attribute-only update — an owner name corrected, a status flag flipped — leaves the geometry byte-identical. The target can apply it as a cheap column write, and no spatial index needs touching.

A geometry update rewrites the shape, which invalidates any cached derivation: area, centroid, tile assignment, spatial join results, and the index entry itself. On a large target these are the expensive changes, and knowing which updates are which turns a full re-derivation into a targeted one.

Distinguishing them costs one comparison in the trigger: hash the WKB and store it, so a consumer can tell whether the geometry moved without comparing geometries itself. The flag then lets the apply step skip re-tiling for the ninety percent of updates that are administrative rather than spatial — which, on a parcel dataset, is usually most of them.