Incremental OSM Loads into DuckDB

A daily reload of a country extract is an hour of work to change a fraction of a percent of rows. The alternative is not complicated, but it has one requirement people underestimate: the load must be safe to run twice.

Prerequisites

Conceptual minimum

A change file lists created, modified and deleted elements. Turning that into a table update needs three steps, and the middle one is the subtle part.

Collect the affected features. Not just the changed elements: a moved node changes the geometry of every way referencing it, and a changed way changes every relation containing it. The affected set is the changed elements plus their dependents, which requires the reverse references.

Rebuild, do not patch. For each affected feature, rebuild its row from current data rather than applying a delta to the stored row. Rebuilding is idempotent and delta application is not, and the cost difference is negligible because the affected set is small.

Apply atomically. Upserts and deletes for one change file go in one transaction, with the sequence number recorded in the same transaction. That is what makes the load safe to retry: either the data and the state both advanced, or neither did.

The idempotence requirement is worth stating plainly: running the same change file twice must leave the table identical. Retries happen, and a load that double-counts on retry is worse than one that fails.

The four stages of applying one change file Four stages. The collect stage reads the change file and gathers the identifiers of changed elements plus every feature that depends on them, since a moved node changes the geometry of the ways using it. The rebuild stage reconstructs each affected feature from current data rather than applying a delta, which makes the operation idempotent. The apply stage performs upserts and deletions in a single transaction. The record stage writes the new replication sequence number inside that same transaction, so data and state advance together or not at all. Collect, rebuild, apply, record — one transaction collect changed plus dependents reverse references rebuild from current data never a delta apply upsert and delete one transaction record sequence in the same both or neither Recording the sequence outside the transaction is how a pipeline ends up believing it applied a change file it did not.
Rebuilding rather than patching is what makes a retry harmless, which matters because retries are routine.

Runnable solution

python 
from __future__ import annotations

import logging
from collections.abc import Iterable
from dataclasses import dataclass, field
from pathlib import Path

import duckdb

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger("osm.warehouse.incremental")


@dataclass
class ChangeSet:
    sequence: int
    created: set[tuple[str, int]] = field(default_factory=set)
    modified: set[tuple[str, int]] = field(default_factory=set)
    deleted: set[tuple[str, int]] = field(default_factory=set)

    @property
    def touched(self) -> set[tuple[str, int]]:
        return self.created | self.modified | self.deleted


def dependents(conn: duckdb.DuckDBPyConnection,
               touched: set[tuple[str, int]]) -> set[tuple[str, int]]:
    """Features whose geometry depends on a changed element.

    A moved node changes every way that references it, and a changed way
    changes every relation containing it. Missing this leaves stale geometry
    on features the change file never mentions.
    """
    nodes = [i for t, i in touched if t == "node"]
    ways = [i for t, i in touched if t == "way"]
    affected: set[tuple[str, int]] = set()

    if nodes:
        rows = conn.execute(
            "SELECT DISTINCT 'way', way_id FROM way_node WHERE node_id IN "
            "(SELECT UNNEST(?))", [nodes]).fetchall()
        affected.update((t, i) for t, i in rows)
    if ways or affected:
        candidate_ways = ways + [i for t, i in affected if t == "way"]
        rows = conn.execute(
            "SELECT DISTINCT 'relation', relation_id FROM relation_member "
            "WHERE member_type = 'way' AND member_id IN (SELECT UNNEST(?))",
            [candidate_ways]).fetchall()
        affected.update((t, i) for t, i in rows)

    logger.info("%d touched element(s) affect %d additional feature(s)",
                len(touched), len(affected - touched))
    return affected


def apply_change(conn: duckdb.DuckDBPyConnection, change: ChangeSet,
                 rebuild: callable) -> None:
    """Apply one change file atomically, including the sequence number."""
    expected = conn.execute(
        "SELECT sequence FROM replication_state").fetchone()[0]
    if change.sequence <= expected:
        # Already applied. Idempotence means this is a no-op, not an error.
        logger.info("sequence %d already applied (state is %d); skipping",
                    change.sequence, expected)
        return
    if change.sequence != expected + 1:
        raise ValueError(f"sequence gap: state is {expected}, file is "
                         f"{change.sequence}; catch up before applying")

    affected = change.created | change.modified
    affected |= dependents(conn, change.touched) - change.deleted
    rows = [rebuild(osm_type, osm_id) for osm_type, osm_id in sorted(affected)]
    rows = [r for r in rows if r is not None]

    conn.execute("BEGIN TRANSACTION")
    try:
        if rows:
            conn.executemany("""
                INSERT INTO fact_feature
                  (feature_key, snapshot_key, class_key, area_key, osm_type,
                   osm_id, osm_version, name, geom, area_m2, length_m, tags)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT (feature_key) DO UPDATE SET
                  class_key = excluded.class_key, area_key = excluded.area_key,
                  osm_version = excluded.osm_version, name = excluded.name,
                  geom = excluded.geom, area_m2 = excluded.area_m2,
                  length_m = excluded.length_m, tags = excluded.tags
            """, rows)
        if change.deleted:
            conn.executemany(
                "DELETE FROM fact_feature WHERE osm_type = ? AND osm_id = ?",
                sorted(change.deleted))
        # The sequence advances INSIDE the transaction: data and state together.
        conn.execute("UPDATE replication_state SET sequence = ?",
                     [change.sequence])
        conn.execute("COMMIT")
    except Exception:
        conn.execute("ROLLBACK")
        raise

    logger.info("sequence %d: %d upsert(s), %d deletion(s)",
                change.sequence, len(rows), len(change.deleted))


if __name__ == "__main__":
    logger.info("rebuild affected features from current data, never from deltas")

Step-by-step walkthrough

  1. Check the sequence before doing anything. An already-applied file is skipped rather than failing, because a retry after a network error is routine and should be harmless.
  2. Refuse a gap. A file whose sequence is more than one ahead means diffs were missed, and applying it leaves the table silently inconsistent with the map.
  3. Expand to dependents. A moved node changes every way that uses it, and a change file does not mention those ways. Skipping this leaves stale geometry on features nothing appeared to touch.
  4. Exclude deleted features from the rebuild set. A dependent that was itself deleted should not be reconstructed.
  5. Rebuild whole rows. Reconstructing from current data makes the operation idempotent; applying deltas to stored rows does not, and the affected set is small enough that the difference in cost is immaterial.
  6. Use an upsert, not an insert. A created element may already be present if a previous run partially succeeded, and an upsert handles both cases identically.
  7. Advance the sequence inside the transaction. This is the single most important line: data and state commit together, so a crash leaves both at the old value and the retry is clean.
  8. Sort before applying. Deterministic ordering makes two runs over the same input produce identical write patterns, which matters when comparing logs.
Rows touched by one minutely change file, by stage Four counts for a single minutely diff over a country-sized table. The change file itself names a few thousand elements. Expanding to ways whose nodes moved roughly triples that. Adding relations containing changed ways adds a smaller increment. The final upsert count remains a tiny fraction of the table's tens of millions of rows, which is the entire argument for loading incrementally rather than reloading. A minutely diff against a table of tens of millions Elements in the change file about 3,100 Plus dependent ways about 9,400 Plus dependent relations about 10,200 Rows in the table tens of millions The dependent expansion roughly triples the work and is not optional: skipping it leaves stale geometry nothing appears to have touched.
Even after expansion the affected set is a rounding error against the table, which is why a full reload is so wasteful.
Three ways an incremental load diverges from the map without erroring Three panels. Skipping dependent expansion leaves ways whose nodes moved carrying stale geometry, which nothing in the change file pointed at and no error reports. Writing the sequence outside the transaction lets data and state disagree after a crash, so a change is either applied twice or skipped forever. Applying across a sequence gap leaves some features at a recent state and others frozen at an older one, producing a table describing a moment the map never passed through. Three divergences, none of which raises No dependent expansion Ways keep old geometry Nothing pointed at them No error anywhere Found by a visual check State outside the transaction Crash between the two Applied twice, or never Depends on the order Found much later Gap applied anyway Some rows recent Others frozen A state that never existed Impossible to reason about Each is prevented by one line, and each is discovered weeks later by somebody comparing the table against the live map.
The third is the worst because there is no consistent moment the table corresponds to, so no comparison can diagnose it.

Verification

  • Re-applying is a no-op. Run the same change file twice and confirm the second run skips and the table is byte-identical.
  • A gap is refused. Skip a sequence deliberately and confirm the load raises rather than proceeding.
  • Dependents are updated. Move a node in a test diff and confirm the ways referencing it have new geometry.
  • State and data agree. After an interrupted load, the recorded sequence must correspond to the data actually present.
  • Counts reconcile. Compare the table’s feature count against the same count from a full reload of the caught-up extract.

Common errors and fixes

Symptom Root cause One-line fix
Stale geometry after a node move Dependents not expanded Resolve reverse references for changed nodes and ways
Retry double-counts Deltas applied rather than rows rebuilt Rebuild whole rows from current data
State ahead of the data Sequence written outside the transaction Update the sequence inside the same transaction
Load fails on a re-run Insert used instead of upsert Use an upsert keyed on the feature key
Silent divergence from the map A sequence gap applied anyway Refuse anything other than the next sequence
Deleted features reappear Dependents rebuilt after deletion Exclude deleted elements from the rebuild set
Write patterns differ between runs Unordered application Sort the affected set before applying

Specification reference

A change file contains create, modify and delete blocks describing element-level changes between two replication states, identified by a sequence number. Applying one to a derived dataset requires resolving which derived rows depend on the changed elements, since the file describes elements rather than features. See Applying .osc Change Files with osmium for the file format and Replication Sequence Numbers & State for the state model.

Frequently Asked Questions

Why rebuild features rather than patching them?

Because rebuilding is idempotent and patching is not. Applying the same delta twice double-applies it, which turns a routine retry into silent corruption; rebuilding a row from current data produces the same result however many times it runs. The cost argument that favours patching does not apply here, because the affected set after one change file is a few thousand rows against a table of tens of millions.

Why does a change file not mention every affected feature?

Because it describes elements and your table describes features. When a mapper drags a node, the change file contains that node and nothing else — but every way whose geometry includes it now has different geometry, and every relation containing those ways may too. Resolving those reverse references is the step that turns an element-level change into a feature-level one, and skipping it leaves stale geometry that nothing appears to have touched.

Why must the sequence number be updated inside the transaction?

So that the data and the record of what has been applied cannot disagree. If the sequence is written after the commit, a crash in between leaves data applied and state unaware, and the retry applies it again. If it is written before, a failed commit leaves state ahead of the data and the change is silently skipped forever. Inside the transaction, both advance or neither does.

What should happen when a sequence is skipped?

The load should refuse. A gap means diffs were missed, and applying a later one leaves the table describing a state the map never passed through — some features updated to a recent state and others frozen at an older one, with nothing indicating which. Catching up through the missing sequences, or reloading from a fresh extract, are the only two correct responses.

Up one level: Modelling OSM for Analytics Warehouses.