Storing Flow State in PostGIS Versus Object Storage

The choice is usually presented as a database-versus-files question and it is not one. It is a question about the queries the state has to answer: if any of them is “which of these overlap a polygon”, the state belongs in PostGIS, because nothing else will answer it; if the state is large, opaque and read only by the key that wrote it, object storage is cheaper and simpler. Most pipelines need both, and the useful skill is putting each piece where its queries live rather than choosing one home for everything.

When to Use This Pattern#

• A pipeline keeps more than one kind of state — a completion record, a scratch artefact, a position marker.
• Somebody asks spatial questions of the pipeline — coverage by district, staleness by region.
• The fan-out writes state concurrently, which puts a hard requirement on the write path.
• Storage cost has started to matter, which usually means checkpoints are being kept in the wrong place.

Complete Working Example#

The same run, with each kind of state in the home its queries need.

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path

import psycopg
from prefect import flow, task


# ---- PostGIS: small, transactional, spatially queried ----------------------

LEDGER_UPSERT = """
INSERT INTO tile_ledger (layer, z, x, y, work_key, geom, built_at)
VALUES (%(layer)s, %(z)s, %(x)s, %(y)s, %(key)s,
        ST_TileEnvelope(%(z)s, %(x)s, %(y)s), now())
ON CONFLICT (layer, z, x, y) DO UPDATE
SET work_key = EXCLUDED.work_key, built_at = now()
WHERE tile_ledger.work_key IS DISTINCT FROM EXCLUDED.work_key;
"""

COVERAGE_BY_DISTRICT = """
SELECT d.name,
       count(l.*) FILTER (WHERE l.built_at > now() - interval '7 days') AS fresh,
       count(l.*)                                                       AS total
FROM   districts d
JOIN   tile_ledger l ON ST_Intersects(d.geom, l.geom)      -- the query that decides the home
WHERE  l.layer = %(layer)s
GROUP  BY d.name
ORDER  BY fresh::float / NULLIF(count(l.*), 0);
"""


# ---- Object storage: large, opaque, keyed, disposable ----------------------

@dataclass(frozen=True)
class Scratch:
    bucket: str
    run_id: str

    def key(self, tile: Tile) -> str:
        return f"ckpt/{self.run_id}/{tile.z}/{tile.x}/{tile.y}.tif"


@task(retries=2, tags=["tile"])
def build(tile: Tile, key: str, scratch: Scratch, conn_str: str) -> str:
    obj = scratch.key(tile)
    if head_object(scratch.bucket, obj) is not None:       # resume, by key only
        return obj
    put_object(scratch.bucket, obj, warp(tile))            # megabytes, never in a row
    with psycopg.connect(conn_str) as conn:                # the record, in the database
        conn.execute(LEDGER_UPSERT,
                     {"layer": "ortho", "z": tile.z, "x": tile.x, "y": tile.y, "key": key})
    return obj

The line that decides most cases is the spatial join in COVERAGE_BY_DISTRICT. That query is unremarkable in SQL and effectively impossible against object storage: answering it there means listing every key, parsing the tile index out of each one, computing an envelope, and testing it against a polygon you would have to fetch from somewhere else — in application code, over a network, for every object. The moment anybody wants coverage broken down by an administrative boundary, the ledger has to be in a spatial database, and it is much cheaper to have put it there at the start.

Parameter & Option Reference#

Verification & Testing#

def test_ledger_answers_a_spatial_question(db) -> None:
    seed_ledger(db, tiles_for("district-A", fresh=90, stale=10))
    rows = db.all(COVERAGE_BY_DISTRICT, {"layer": "ortho"})
    assert rows[0].name == "district-A" and rows[0].fresh == 90


def test_concurrent_writers_do_not_duplicate(db) -> None:
    with ThreadPoolExecutor(8) as pool:
        list(pool.map(lambda _: upsert_ledger(db, TILE, key="k1"), range(8)))
    assert db.scalar("SELECT count(*) FROM tile_ledger WHERE z=%s AND x=%s AND y=%s",
                     TILE) == 1


def test_checkpoints_expire(bucket) -> None:
    rule = get_lifecycle_rule(bucket, prefix="ckpt/")
    assert rule is not None and rule.days <= 3, "scratch prefix has no expiry"


def test_no_large_payload_in_a_row(db) -> None:
    biggest = db.scalar("SELECT max(pg_column_size(t.*)) FROM tile_ledger t")
    assert biggest < 4096, "something large has been stored in the ledger"

The last two tests exist because both failures are invisible until they are expensive. A missing lifecycle rule shows up as a storage line item three months later, by which time the bucket holds several terabytes of checkpoints from runs nobody remembers. A large payload in a ledger row shows up as a database that has quietly become the pipeline’s blob store, with backup times and replica lag to match. Both are one assertion each in CI, and both are much harder to unwind than to prevent.

There is a migration path worth knowing, because plenty of pipelines start with everything in object storage and only later discover they needed a database. It is easier than it looks, and the reason is that a well-built ledger is derivable: every object carries its work key, its source digest and its recipe version in metadata, so the ledger can be rebuilt by walking the output prefix once. That single property — the ledger is an index over the outputs, not the only record of them — turns what would be a data migration into a batch job that can be run twice with no consequence. Pipelines that stored the ledger as the primary record instead have a genuinely difficult migration ahead, because losing a row means losing the knowledge that an object exists.

The reverse direction almost never happens, and it is worth asking why. Nobody moves a ledger out of a spatial database once the coverage dashboards exist, because the queries those dashboards run have no equivalent anywhere else. What does happen is that large payloads accumulate in database rows and have to be moved out under pressure, usually the week a backup starts taking longer than the maintenance window. The asymmetry is a good argument for putting the small transactional state in PostGIS from the beginning and being disciplined about keeping everything else out of it.

Common Pitfalls#

• A ledger in object storage. It works until the first spatial question, at which point it has to be migrated under pressure with history to preserve.
• Checkpoints in the database. Row sizes in the tens of megabytes make backups, replication and vacuum everybody’s problem at once.
• No lifecycle rule on the scratch prefix. Nothing breaks; the bill grows forever.
• Relying on last-writer-wins for a completion record. Two concurrent writers in object storage produce one surviving answer, arbitrarily chosen — fine for a checkpoint, wrong for a ledger.
• Listing a prefix to compute coverage. It is a request per object and it gets slower every week; the same answer is one indexed query in PostGIS.
• One connection per mapped task. Four hundred tasks opening their own connection will exhaust the pool long before they exhaust the database; pool at the worker.

Frequently Asked Questions#

Related#

• State management in geospatial flows — the three kinds of state this page finds homes for
• Checkpointing large raster mosaics — what lives in the object-storage column
• Securing PostGIS connections in workflows — connecting safely from a fan-out
• Cutting egress costs with COG range reads — the other half of the object-storage bill
• Visualizing tile coverage gaps on a geomap — the query that decides the home, on a screen

← State Management in Geospatial Flows