Managing State for Incremental Shapefile Updates

Almost every public vector source is published as a full replacement: the whole cadastre, the whole road network, every week, with no indication of what changed. Loading it as a replacement is simple and wrong — it rewrites four hundred thousand unchanged rows, invalidates every downstream cache and destroys any history. The alternative is to compute the difference yourself from a per-feature digest, which turns a full delivery into an incremental update and costs one hash column.

When to Use This Pattern#

• The source is a full snapshot with no change feed, which describes most open geodata.
• Downstream consumers cache on modification time, so rewriting unchanged rows has a real cost.
• The table is large enough that a full rewrite is disruptive — bloat, vacuum pressure, replica lag.
• Somebody will eventually ask what changed between two deliveries, and reconstructing it afterwards is impossible.

Complete Working Example#

Three parts: a stable feature identity, a digest over the fields that matter, and a batch-scoped upsert that soft-deletes what the delivery no longer contains.

from __future__ import annotations

import hashlib
from pathlib import Path

import fiona
from prefect import flow, task, get_run_logger
from shapely.geometry import shape


def feature_digest(props: dict, geom) -> str:
    """Everything that would make a downstream consumer care, and nothing else."""
    # WKB at fixed precision: a re-export that shifts a vertex by 1e-12 is not a change.
    wkb = shape(geom).normalize().wkb_hex
    fields = "|".join(f"{k}={props.get(k)!r}" for k in sorted(TRACKED_FIELDS))
    return hashlib.sha256(f"{fields}|{wkb}".encode()).hexdigest()[:32]


@task(retries=2, timeout_seconds=1800)
def stage_delivery(path: Path, batch: str) -> int:
    """Load the whole shapefile into a staging table. Cheap, unlogged, disposable."""
    with fiona.open(path) as src, pg_copy("stg_parcels") as sink:
        for feat in src:
            sink.write(
                {
                    "batch": batch,
                    "parcel_id": feat["properties"]["PARCEL_ID"],
                    "digest": feature_digest(feat["properties"], feat["geometry"]),
                    "props": feat["properties"],
                    "geom": shape(feat["geometry"]).wkb,
                }
            )
    return sink.count


MERGE = """
WITH changed AS (
    INSERT INTO parcels AS p (parcel_id, digest, props, geom, batch, valid_from)
    SELECT parcel_id, digest, props, geom, %(batch)s, now() FROM stg_parcels
    ON CONFLICT (parcel_id) DO UPDATE
    SET digest = EXCLUDED.digest, props = EXCLUDED.props, geom = EXCLUDED.geom,
        batch = EXCLUDED.batch, valid_from = now()
    WHERE p.digest IS DISTINCT FROM EXCLUDED.digest      -- the whole optimisation
    RETURNING p.parcel_id
),
touched AS (
    -- Rows the delivery contains but did not change still need their batch stamped,
    -- or the withdrawal step below will delete them.
    UPDATE parcels SET batch = %(batch)s
    WHERE parcel_id IN (SELECT parcel_id FROM stg_parcels) AND batch <> %(batch)s
    RETURNING parcel_id
),
withdrawn AS (
    UPDATE parcels SET valid_to = now()
    WHERE batch <> %(batch)s AND valid_to IS NULL
    RETURNING parcel_id
)
SELECT (SELECT count(*) FROM changed), (SELECT count(*) FROM withdrawn);
"""


@flow(name="parcel-delivery")
def ingest(path: Path, batch: str, published_at: datetime) -> None:
    total = stage_delivery(path, batch)
    changed, withdrawn = merge(batch)
    if withdrawn > total * 0.05:
        raise RuntimeError(f"{withdrawn} withdrawals on a {total}-feature delivery — refusing")
    get_run_logger().info("delivery %s: %d features, %d changed, %d withdrawn",
                          batch, total, changed, withdrawn)
    advance_watermark("parcels", published_at)

The geometry normalisation inside the digest is doing quiet but essential work. Vector exports are not stable at the bit level: a re-export from the same source system can reverse a ring’s winding order, reorder rings within a polygon or shift a coordinate in the fifteenth decimal place, none of which is a change anybody cares about. Hashing raw WKB would report all of them as edits and the pipeline would rewrite the whole table every week while appearing to be incremental. Normalising first, and rounding coordinates to the source’s actual precision, is what makes the digest mean “changed” rather than “re-exported”.

Parameter & Option Reference#

Verification & Testing#

def test_reexport_without_changes_writes_nothing(db, tmp_path) -> None:
    ingest(fixture("parcels_w1.shp"), batch="w1", published_at=T1)
    before = db.scalar("SELECT max(valid_from) FROM parcels")
    ingest(fixture("parcels_w1_reexported.shp"), batch="w2", published_at=T2)
    assert db.scalar("SELECT max(valid_from) FROM parcels") == before


def test_ring_order_is_not_a_change() -> None:
    a = feature_digest(PROPS, POLY)
    b = feature_digest(PROPS, reverse_rings(POLY))
    assert a == b, "digest is sensitive to winding order"


def test_truncated_delivery_is_refused(db, tmp_path) -> None:
    ingest(fixture("parcels_w1.shp"), batch="w1", published_at=T1)
    with pytest.raises(RuntimeError, match="refusing"):
        ingest(fixture("parcels_first_1000_only.shp"), batch="w2", published_at=T2)


def test_withdrawal_is_soft(db) -> None:
    ingest(fixture("parcels_minus_one.shp"), batch="w3", published_at=T3)
    row = db.one("SELECT valid_to FROM parcels WHERE parcel_id = %s", (GONE_ID,))
    assert row.valid_to is not None
    assert db.scalar("SELECT count(*) FROM parcels WHERE parcel_id = %s", (GONE_ID,)) == 1

The truncated-delivery test is the one that pays for itself. A partial download — an HTTP connection dropped at eighty per cent, a zip extracted before it finished writing — produces a perfectly valid shapefile containing a prefix of the features, and without a guard the merge dutifully withdraws the eighty thousand parcels that were not in it. Every check downstream passes, because the data is internally consistent; it is simply missing a fifth of the country. A percentage threshold on withdrawals costs one line and converts that into a refused run.

One consequence of computing the difference yourself is that you acquire a change feed the publisher never offered, and it is worth exploiting. The changed count per delivery, logged on every run, is a weekly time series of how much the source actually moves — and it is remarkably informative. A cadastre that normally changes twelve hundred parcels a week and suddenly changes forty thousand has had a bulk correction applied upstream, which is something worth knowing before a downstream consumer notices. A source that changes exactly zero features for three consecutive weeks has almost certainly stopped being maintained, which no error will ever tell you. Both readings come free from a number the merge already computes.

The same feed makes selective downstream work possible. Because the merge returns the identifiers it changed, the tiles that need rebuilding are the tiles intersecting those features — a spatial query against the changed set rather than the whole layer. That is the difference between a nightly re-render of a national tile pyramid and a re-render of the forty tiles a week’s parcel edits actually touched, and it falls out of the pattern rather than needing to be built.

Common Pitfalls#

• Hashing raw WKB. Winding order and floating-point noise then count as changes and the merge rewrites everything, incrementally.
• Including export metadata in the digest. An extract timestamp inside the attributes makes every feature differ on every delivery.
• Hard-deleting withdrawals. The one question people ask about historical data is when a feature disappeared, and a hard delete is the one answer you cannot reconstruct.
• Forgetting to stamp unchanged rows with the new batch. They then look absent from the delivery and get withdrawn — the subtlest bug in this whole pattern.
• No withdrawal guard. A truncated download becomes a mass deletion that every downstream check accepts as legitimate.
• Advancing the watermark from the run’s clock. Use the publisher’s timestamp; a run that starts late otherwise records that it ingested data that had not been published yet.

Frequently Asked Questions#

Related#

• State management in geospatial flows — the watermark this advances
• Making PostGIS upserts idempotent for feature loads — the write side in detail
• Streaming large GeoPackage loads into PostGIS — getting the delivery into staging
• Repairing invalid geometries before load — what to do before hashing a geometry
• Storing flow state in PostGIS versus object storage — where the batch record belongs

← State Management in Geospatial Flows