DuckDB Spatial Analytical SQL for GIS

Resuming Failed Batch Spatial Jobs

A three-hour spatial job that fails at two hours and fifty minutes is a design problem rather than an operations problem, and the design decision that fixes it is where the checkpoint goes. This walkthrough, part of the batch processing pipelines reference, covers making a run restartable at partition granularity, the difference between a checkpoint that records intent and one that records completion, and how to make a rerun produce the same output as an uninterrupted run.

Root-Cause Analysis: why a failed batch job cannot simply be re-run

  • The output is partially written. A run that appended to one output file has left an unknown number of rows in it. Re-running appends again, so the result contains duplicates that no later query can distinguish from real ones.
  • There is no record of what completed. Without a completion log the only way to know what was done is to inspect the output, and inspecting the output is exactly what the partial write has made unreliable.
  • The checkpoint records intent, not completion. A log written when a partition starts cannot distinguish “finished” from “died halfway”, so a restart either repeats work or skips it, and neither is safe.
  • The work order is not deterministic. A run that processes partitions in whatever order the scheduler produced cannot be resumed by position, because position means something different on the next run.
  • The output is not idempotent. Even with a correct completion log, re-processing a partition must overwrite rather than add — and an append-only sink makes that impossible.

The distinguishing question is whether a partition can be re-processed safely. If it can, resumption is a set difference; if it cannot, no amount of logging helps.

Four checkpoint designs and what each survives No checkpoint means a full re-run; intent means ambiguity; completion means a set difference; completion plus per-partition output means idempotence. CHECKPOINT AFTER A FAILURE VERDICT none full re-run, plus cleanup the default, and the worst written when the partition starts cannot distinguish finished from died ambiguous written after the output is durable a set difference correct completion + per-partition output overwrite and continue idempotent

One line moves from the second row to the third, and it is the whole design.

Deterministic Configuration 

import duckdb

con = duckdb.connect("pipeline.duckdb")
con.execute("INSTALL spatial; LOAD spatial")
con.execute("SET memory_limit = '6GB'")
con.execute("SET threads = 4")
con.execute("SET temp_directory = '/var/tmp/duckdb_batch'")

# The completion log lives in the database, so it is transactional with
# nothing else and survives whatever killed the process.
con.execute("""
    CREATE TABLE IF NOT EXISTS run_log (
        run_id      VARCHAR,
        unit_id     VARCHAR,
        rows_out    BIGINT,
        finished_at TIMESTAMP,
        PRIMARY KEY (run_id, unit_id)
    )
""")

Optimized Execution Pattern

The pattern is a deterministic unit list, a per-unit output path, and a completion log consulted at startup to compute what remains.

# ANTI-PATTERN: one output file, no log. A failure leaves an unknown number
# of rows written and no way to know which partitions produced them.
for tile in tiles:
    con.execute(f"""
        COPY (SELECT ... FROM parcels WHERE tile_id = '{tile}')
        TO 'output.parquet' (FORMAT PARQUET, APPEND)
    """)
# PATTERN: per-unit output, completion recorded after the write, and the
# remaining work computed as a set difference at startup.
RUN_ID = "parcels-2024-08"

planned = [r[0] for r in con.execute(
    "SELECT DISTINCT tile_id FROM parcels ORDER BY tile_id"      # deterministic
).fetchall()]

done = {r[0] for r in con.execute(
    "SELECT unit_id FROM run_log WHERE run_id = ?", [RUN_ID]
).fetchall()}

for tile in [t for t in planned if t not in done]:
    n = con.execute(f"""
        COPY (SELECT parcel_id, zone_id, geom FROM parcels WHERE tile_id = '{tile}')
        TO 'out/tile_id={tile}' (FORMAT PARQUET, ROW_GROUP_SIZE 30000)
    """).fetchone()
    con.execute("INSERT INTO run_log VALUES (?, ?, ?, now())", [RUN_ID, tile, n])

The order of the last two statements is the entire correctness argument. Writing the log first records an intention; writing it after the COPY returns records a fact, and only a fact makes the set difference meaningful.

Three ways to make a unit re-processable A per-unit path is simplest, a staging path hides partial writes, and a delete-then-insert works only where both are transactional. APPROACH MECHANISM WHEN a per-unit output path a rerun overwrites that path the default stage, then move on success a partial write is never visible when partial output is dangerous delete the unit rows, reinsert needs both in one transaction when the sink is a table

Partitioned output gives you the first row for free, which is most of why it is worth having.

Making the rerun produce identical output

Restartability is not enough on its own: a resumed run should produce the same bytes an uninterrupted run would have. Two things threaten that in spatial work, and both are avoidable.

The first is ordering. A COPY with no ORDER BY writes rows in whatever order the scan produced, and that order can differ between runs with different thread counts. Sorting deterministically inside each unit makes the output byte-comparable, which is what allows a regression test to exist at all.

The second is floating-point non-determinism in aggregation. A sum over floats depends on the order of accumulation, so a run at four threads and a run at eight can differ in the last digits. Where the output is compared exactly, either round to a stated precision or sort before aggregating.

-- Deterministic within the unit: a stated order and a stated precision.
COPY (
    SELECT parcel_id, zone_id, ST_ReducePrecision(geom, 0.001) AS geom
    FROM parcels WHERE tile_id = 'T042'
    ORDER BY parcel_id
) TO 'out/tile_id=T042' (FORMAT PARQUET, ROW_GROUP_SIZE 30000);

Diagnostic Queries & Plan Validation

After a resumed run, two checks confirm the result is what an uninterrupted run would have produced.

-- 1. Every planned unit completed exactly once.
SELECT count(*) AS units_logged, count(DISTINCT unit_id) AS units_distinct
FROM run_log WHERE run_id = 'parcels-2024-08';

-- 2. The output row count matches the sum of the logged per-unit counts.
SELECT (SELECT sum(rows_out) FROM run_log WHERE run_id = 'parcels-2024-08') AS logged,
       (SELECT count(*) FROM read_parquet('out/**/*.parquet', hive_partitioning = true)) AS actual;

A logged count above the actual means a unit was logged without its output being durable, which points at the log being written too early. An actual count above the logged means a unit was written twice, which points at output that is not per-unit.

Four resumption failures and what each indicates Too many rows means non-per-unit output; too few means a premature checkpoint; different bytes mean non-determinism; repeated work means a non-deterministic plan. SYMPTOM INDICATES FIX more rows than an uninterrupted run output is not per-unit write per-unit paths fewer rows the checkpoint was written too early log after the write returns different bytes, same rows ordering or float accumulation ORDER BY and a stated precision work repeated the unit list is not deterministic order the plan

Two of the four are ordering and two are the checkpoint. None is the spatial SQL.

Sizing units so a failure is cheap

The cost of a failure is one unit of work, so unit size is a direct decision about how much a failure costs. Very large units mean a failure discards hours; very small ones mean per-unit overhead — a COPY, a log write, a file — dominates the run.

The workable target is a unit that takes a few minutes, which usually means splitting the natural geographic key by feature count rather than using it directly. A national dataset partitioned by tile has tiles that differ by two orders of magnitude in feature count, so “one tile” is a unit that takes seconds in the countryside and an hour in a city centre. Splitting the heavy tiles into equal-weight sub-units makes the failure cost uniform, which is what makes it plannable.

Geometry Validation & Fallback Routing

Where a unit fails repeatedly rather than transiently, quarantine it and continue rather than blocking the run.

# A unit that fails is recorded and skipped rather than stopping the run.
# The failures are then a work queue, and the other 400 units completed.
for tile in remaining:
    try:
        n = process(tile)
        con.execute("INSERT INTO run_log VALUES (?, ?, ?, now())", [RUN_ID, tile, n])
    except Exception as exc:
        con.execute(
            "INSERT INTO run_failures VALUES (?, ?, ?, now())", [RUN_ID, tile, str(exc)[:500]]
        )

Frequently Asked Questions

Where exactly should the checkpoint be written?

After the unit output is durable and never before. A log written at the start records an intention, so a restart cannot distinguish a finished unit from one that died halfway — and it must then either repeat work or skip it, both of which are wrong. Moving that one statement after the write is the whole design.

Why does per-unit output matter so much?

Because it makes re-processing idempotent. A rerun overwrites exactly the path that unit owns and nothing else, so the set-difference restart is safe by construction. An append-only sink cannot offer that, which is why a partial write there requires a manual cleanup before any retry.

How large should a unit be?

A few minutes of work, which usually means splitting the natural geographic key rather than using it directly — tiles in a national dataset differ by two orders of magnitude in feature count. Uniform unit cost is what makes the cost of a failure predictable.

Should a failed unit stop the run?

Usually not. Recording the failure and continuing means the other units complete and the failures become a work queue with a reason attached. A run that stops on the first bad unit turns one bad tile into a night of lost processing.

How do I know a resumed run matches an uninterrupted one?

Compare the logged per-unit row counts against the actual output count, and compare bytes if the output is meant to be reproducible. A logged count above the actual means the checkpoint was early; an actual count above the logged means a unit was written twice.

Does the log need to be in the database?

It needs to survive whatever killed the process, and a table in the pipeline database is the simplest thing that does. A file works too, provided the write is flushed; what does not work is holding the state in memory, which is exactly what the failure destroyed.

Up: Batch Processing Pipelines for DuckDB Spatial Workloads