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

Incremental Spatial Loading with Watermark Timestamps

The Problem: A Delta Load Is Only Safe If the Watermark Moves Correctly

An incremental spatial load reads a stored high-water mark, fetches everything newer, writes it, and records a new mark. The concept is simple; the failure modes are subtle, and every one of them corrupts the dataset silently rather than crashing.

  • The watermark advances past data that was never loaded. If the pipeline writes the new mark from the wall clock and then crashes mid-write, the next run starts after records that never landed — a permanent, invisible gap in the spatial record.
  • Late-arriving scenes are skipped forever. Querying with > watermark exactly means any record the source backdated or published after the last run falls below the lower bound and is never seen again.
  • The overlap window duplicates geometries. The fix for late data — re-reading a recent slice — floods the target with duplicate polygons unless the write is idempotent on a stable key.
  • Naive local timestamps drift. A watermark stored as naive local time and compared against UTC source timestamps quietly shifts the window by the UTC offset, dropping or double-loading an hour of data at every run.

Version and Environment Compatibility

Component Version Note
Python 3.10+ Timezone-aware datetime, `X
GeoPandas >=0.14 to_postgis() write path used by the upsert
Shapely >=2.0 Vectorized shapely.is_valid(arr) for pre-load validation
SQLAlchemy >=2.0 engine.begin() transactional scope for atomic advance 
pip install "geopandas>=0.14" "shapely>=2.0" "sqlalchemy>=2.0"

How the Watermark Advances Across Runs

The correctness of the whole pattern lives in the ordering of four operations and in choosing the new watermark from data, not the clock. The sequence below traces one run against the source and the state table.

Watermark incremental-load sequence Four ordered steps between a state table, the pipeline, and the source. Step one reads the stored watermark. Step two queries the source for records newer than the watermark minus an overlap. Step three upserts the batch into the target on a stable key. Step four writes a new watermark, derived from the batch maximum timestamp, back to the state table within the same transaction as the load. state table pipeline source + target 1. read watermark W 2. fetch rows > (W − overlap) delta batch (max ts = W′) 3. upsert on scene_id 4. write W′ (same txn as load)

The incremental_load Function

The function below is the complete pattern in one place: it reads the watermark, fetches the delta through a caller-supplied fetch function, validates geometry with the array API, upserts on a stable key, and advances the watermark — derived from the batch maximum — inside the same transaction as the load. The fetch_delta callable is injected so the same loader works against a STAC search, a portal download, or a database change feed.

from __future__ import annotations

import logging
from datetime import datetime, timedelta, timezone
from typing import Callable

import geopandas as gpd
import shapely
from sqlalchemy import text
from sqlalchemy.engine import Engine

logger = logging.getLogger(__name__)

# A record must expose a source timestamp column the watermark is measured against.
FetchDelta = Callable[[datetime], gpd.GeoDataFrame]


def incremental_load(
    engine: Engine,
    source_id: str,
    fetch_delta: FetchDelta,
    target_table: str,
    *,
    ts_col: str = "updated_at",
    key: str = "scene_id",
    overlap: timedelta = timedelta(hours=48),
    floor: datetime = datetime(2020, 1, 1, tzinfo=timezone.utc),
) -> int:
    """Load only records newer than the stored watermark, then advance it atomically.

    Returns the number of rows upserted. A run that finds no new records is a
    cheap no-op that leaves the watermark untouched.
    """
    # 1. Read the stored high-water mark (timezone-aware UTC), or the floor on first run.
    watermark = _read_watermark(engine, source_id, floor)
    lower_bound = watermark - overlap  # overlap re-reads late arrivals; upsert dedupes
    logger.info(
        "incremental_load | source=%s | watermark=%s | lower_bound=%s",
        source_id, watermark.isoformat(), lower_bound.isoformat(),
    )

    # 2. Fetch the delta. The caller scopes extraction to records >= lower_bound.
    gdf = fetch_delta(lower_bound)
    if gdf.empty:
        logger.info("incremental_load | source=%s | no new records", source_id)
        return 0

    # Normalize the source timestamp column to UTC so comparisons never drift.
    gdf[ts_col] = gpd.pd.to_datetime(gdf[ts_col], utc=True)

    # 3. Validate geometry with the vectorized Shapely 2.0 API — never per-row apply.
    valid = shapely.is_valid(gdf.geometry.values) & ~gdf.geometry.is_empty.values
    dead = gdf.loc[~valid]
    if len(dead):
        logger.warning(
            "incremental_load | source=%s | quarantining %d invalid geometries",
            source_id, len(dead),
        )
    gdf = gdf.loc[valid].copy()
    if gdf.empty:
        logger.warning("incremental_load | source=%s | all records invalid", source_id)
        return 0

    # New watermark comes from the DATA, not the clock — the max timestamp observed.
    new_watermark: datetime = gdf[ts_col].max().to_pydatetime()

    # 4. Upsert and advance in a single transaction: a crash rolls back BOTH.
    ids = gdf[key].unique().tolist()
    with engine.begin() as conn:
        conn.execute(
            text(f"DELETE FROM {target_table} WHERE {key} = ANY(:ids)"),
            {"ids": ids},
        )
        gdf.to_postgis(target_table, conn, if_exists="append", index=False)
        conn.execute(
            text(
                "INSERT INTO etl_watermark (source_id, watermark_ts) "
                "VALUES (:sid, :ts) "
                "ON CONFLICT (source_id) DO UPDATE SET watermark_ts = EXCLUDED.watermark_ts"
            ),
            {"sid": source_id, "ts": new_watermark},
        )

    logger.info(
        "incremental_load | source=%s | upserted=%d | advanced_to=%s",
        source_id, len(gdf), new_watermark.isoformat(),
    )
    return len(gdf)


def _read_watermark(engine: Engine, source_id: str, floor: datetime) -> datetime:
    """Fetch the stored watermark, normalized to timezone-aware UTC, or the floor."""
    with engine.connect() as conn:
        row = conn.execute(
            text("SELECT watermark_ts FROM etl_watermark WHERE source_id = :sid"),
            {"sid": source_id},
        ).fetchone()
    if row is None or row[0] is None:
        return floor
    ts: datetime = row[0]
    return ts if ts.tzinfo else ts.replace(tzinfo=timezone.utc)

Key Implementation Notes

  • The new watermark is gdf[ts_col].max(), never datetime.now(). Anchoring it to the maximum timestamp actually loaded guarantees the mark never overtakes data the source has yet to publish. If a source’s clock and yours disagree, the data-derived value stays correct; a wall-clock value silently opens a gap.
  • The overlap is subtracted at query time, not stored. The watermark itself always records the true high-water mark. Only the lower bound of the next fetch is watermark - overlap, so widening or narrowing the overlap is a one-line change that never requires rewriting stored state.
  • Load and advance share one engine.begin() block. This is the crash-safety guarantee: if the process dies after the upsert but before the watermark write, the transaction rolls back the upsert too, and the next run repeats the delta harmlessly. Splitting them into two transactions reintroduces the gap this pattern exists to prevent.
  • Idempotency comes from the key, not from luck. DELETE ... WHERE scene_id = ANY(:ids) followed by an append makes the batch a set-replace on scene_id. Re-reading an overlapping record overwrites its own row rather than adding a second geometry. Choose a key that is stable across source revisions — a scene ID or tile ID, not a row number.
  • An empty delta is the common case and must be free. On a healthy schedule most runs find nothing new. Returning early on gdf.empty before opening any transaction keeps those runs to a single indexed range query.
  • Validation splits rather than raises. Quarantining invalid geometries with the array API lets the good records advance the watermark; a single malformed feature must not freeze the whole stream.

Troubleshooting Watermark Loads

Failure Root Cause Fix
Permanent gap after a crash Watermark advanced in a separate step from the load Wrap upsert and watermark write in one engine.begin() transaction
Recent records never appear Fetch used > watermark with no overlap Query from watermark - overlap; let the upsert dedupe re-reads
Duplicate polygons every run Overlap re-reads records into a plain insert Upsert on a stable scene_id/tile_id with delete-then-insert or ON CONFLICT
Window shifts by a fixed offset each run Naive-local watermark compared to UTC source timestamps Store and compare watermarks as timezone-aware UTC; normalize the source column with utc=True
Watermark stops advancing Every fetched record fails geometry validation Inspect the dead-letter batch; repair upstream before the load, not after

Integration Note

This loader is the reusable core of the scheduling and incremental spatial loads workflow: the scheduler decides when to call it and supplies the fetch_delta closure that scopes extraction to the source, while incremental_load owns the watermark discipline. Because the whole body is a single atomic upsert-and-advance, it slots directly into an orchestrated task without extra checkpointing — the same guarantees that make writing idempotent spatial ETL tasks in Airflow safe under retry apply here unchanged, since a retried incremental_load re-reads the same delta and converges on the same target state.