Managing Async Transactions for Bulk Geometry Writes

← Back to Async PostGIS Transaction Patterns

Load tens of thousands of geometries into PostGIS from an async service without holding one enormous transaction that bloats WAL, blocks vacuum, and pins a backend for minutes.

# Context & when to use

A row-at-a-time INSERT loop is the wrong tool the moment a load exceeds a few hundred geometries: every statement is its own round-trip, and if you wrap the whole thing in one transaction to make it atomic you now hold locks and generate unflushed WAL for the entire load. The general transaction discipline — short scopes, explicit boundaries — is covered in the parent async PostGIS transaction patterns guide; this page is the bulk-specific corollary: how to move thousands of rows fast while keeping each transaction small.

Use this pattern for scheduled imports, backfills, and derived-geometry rebuilds that run inside your own async process. If the data arrives as an uploaded file over HTTP, do not run it on the request path at all — hand it to a worker as described in async bulk uploads with Celery. The two techniques compose: the worker uses exactly the chunked-copy approach below.

The core idea is asyncpg’s binary COPY protocol (copy_records_to_table), which streams rows into PostgreSQL far faster than parameterised INSERTs, combined with chunked commits so no single transaction grows unbounded. For very large loads, you also defer the GiST index and rebuild it once at the end — maintaining a spatial index while inserting is dramatically slower than building it in one pass.

# Data flow: staged copy, chunked commit, deferred index

# Runnable implementation

This loads geometries through a raw asyncpg connection (the COPY protocol is an asyncpg feature, so we borrow the driver connection from the SQLAlchemy engine rather than going through the ORM). Geometries arrive as Shapely objects, are encoded to WKB, and stream into an unindexed staging table in chunks. The GiST index and ANALYZE come last.

# app/ingest/bulk_geometry.py
import asyncio
from typing import Iterable, Sequence
from shapely.geometry.base import BaseGeometry
from shapely import wkb as shp_wkb
from sqlalchemy.ext.asyncio import AsyncEngine

CHUNK_SIZE = 5_000          # rows per COPY + COMMIT — see the parameters table
TARGET_SRID = 4326


def _to_records(geoms: Iterable[BaseGeometry], srid: int) -> list[tuple]:
    """Encode each geometry to WKB bytes for the binary COPY stream."""
    return [(shp_wkb.dumps(g), srid) for g in geoms]


async def bulk_load(engine: AsyncEngine, geoms: Sequence[BaseGeometry]) -> int:
    total = 0
    # Borrow the raw asyncpg connection: copy_records_to_table is driver-level.
    async with engine.connect() as sa_conn:
        raw = await sa_conn.get_raw_connection()
        pg = raw.driver_connection          # the underlying asyncpg.Connection

        # 1. Fresh, UNLOGGED, UNINDEXED staging table — fast to fill, cheap to drop.
        await pg.execute("""
            DROP TABLE IF EXISTS parcels_staging;
            CREATE UNLOGGED TABLE parcels_staging (geom_wkb bytea, srid int);
        """)

        # 2. Stream WKB records in chunks, one transaction per chunk.
        records = _to_records(geoms, TARGET_SRID)
        for start in range(0, len(records), CHUNK_SIZE):
            chunk = records[start:start + CHUNK_SIZE]
            async with pg.transaction():                    # BEGIN ... COMMIT per chunk
                await pg.copy_records_to_table(
                    "parcels_staging",
                    records=chunk,
                    columns=["geom_wkb", "srid"],
                )
            total += len(chunk)

        # 3. Promote into the real table, converting WKB -> geometry ONCE.
        #    ST_GeomFromWKB parses the binary; ST_SetSRID stamps the CRS.
        async with pg.transaction():
            await pg.execute(f"""
                INSERT INTO parcels (geom)
                SELECT ST_SetSRID(ST_GeomFromWKB(geom_wkb), {TARGET_SRID})
                FROM parcels_staging
                WHERE geom_wkb IS NOT NULL;
            """)

        # 4. Build the spatial index AFTER the data lands — one pass, not per-insert.
        async with pg.transaction():
            await pg.execute(
                "CREATE INDEX IF NOT EXISTS idx_parcels_geom ON parcels USING GIST (geom);"
            )

        # 5. Refresh planner statistics; ANALYZE cannot run inside a transaction block
        #    when combined with VACUUM, so run it on its own.
        await pg.execute("ANALYZE parcels;")

        await pg.execute("DROP TABLE IF EXISTS parcels_staging;")
    return total


if __name__ == "__main__":
    # demo: load 50k random points
    from shapely.geometry import Point
    import random
    from app.database import engine
    pts = [Point(random.uniform(-180, 180), random.uniform(-85, 85)) for _ in range(50_000)]
    print(asyncio.run(bulk_load(engine, pts)), "rows loaded")

If the target table already exists and is indexed, the alternative is to DROP INDEX before the load and recreate it after — for an existing table, CREATE INDEX CONCURRENTLY avoids taking an ACCESS EXCLUSIVE lock but runs slower and cannot be inside a transaction. For a fresh load into a new table, a plain CREATE INDEX at the end is fastest.

For datasets small enough that a single INSERT ... VALUES with executemany is simpler, asyncpg’s executemany still beats a Python loop — but it is bounded by the 65,535-parameter limit per statement (see the gotchas), so chunk it too.

# Key parameters & options

# Gotchas & failure modes

• One giant transaction bloats WAL and holds locks. Wrapping the entire load in a single async with pg.transaction() means nothing commits until the end: WAL grows unbounded, autovacuum cannot reclaim anything the load touched, and every lock is held for the full duration. Symptom: pg_stat_activity shows one backend with a multi-minute xact_start and pg_wal growing fast. Fix: commit per chunk, as above. This is the same “keep transactions short” rule from the parent guide, applied to loads.

• Parameter limit on executemany. PostgreSQL’s wire protocol caps a single statement at 65,535 bound parameters. A multi-row INSERT with 3 columns overflows at ~21,845 rows and fails with asyncpg.exceptions.ProtocolViolationError or a silently truncated batch. Fix: use copy_records_to_table (no per-value parameters) or chunk executemany well under the limit.

• SRID mismatch produces unindexable geometry. WKB does not encode an SRID, so ST_GeomFromWKB(geom_wkb) yields SRID=0. If the target column is declared geometry(Point, 4326), the insert raises Geometry SRID (0) does not match column SRID (4326); if the column is untyped, the rows land with SRID=0 and later bounding-box index queries using ST_Intersects silently return nothing. Fix: always wrap with ST_SetSRID(..., 4326) during promotion.

• Invalid geometries abort the whole promotion. A single self-intersecting polygon can make a downstream constraint or ST_MakeValid-free insert fail, rolling back the entire promotion INSERT. Fix: filter or repair during promotion — WHERE ST_IsValid(ST_GeomFromWKB(geom_wkb)) or wrap the geometry in ST_MakeValid(...).

• Forgetting ANALYZE leaves the planner blind. Right after a bulk load the table’s statistics still say it is empty, so the planner picks sequential scans over the new GiST index. Symptom: the first queries after a load are inexplicably slow. Fix: run ANALYZE before serving reads. Confirm plans with reading EXPLAIN ANALYZE for spatial query optimization.

# Verification

Confirm the row count, the SRID, and that the index exists and is used:

# Row count and SRID sanity
psql "$POSTGIS_DSN" -c "SELECT count(*), min(ST_SRID(geom)), max(ST_SRID(geom)) FROM parcels;"

-- Index present and chosen by the planner after ANALYZE
EXPLAIN (ANALYZE, BUFFERS)
SELECT id FROM parcels
WHERE geom && ST_MakeEnvelope(-10, -10, 10, 10, 4326);
-- Expect: Index Scan / Bitmap Index Scan using idx_parcels_geom (never Seq Scan)

Time the load to catch a runaway single transaction — a healthy chunked load shows steady progress and stable pg_wal size:

SELECT pid, now() - xact_start AS tx_age, state
FROM pg_stat_activity
WHERE query ILIKE '%parcels_staging%';
-- tx_age should reset each chunk, never climb for the whole load

# Related

• Async PostGIS Transaction Patterns — the transaction-boundary rules this bulk pattern specialises
• Handling Deadlocks in Concurrent Spatial Updates — what to do when bulk writers contend with live updates
• Async Bulk Uploads with Celery — run large imports off the request path in a worker

← Back to Async PostGIS Transaction Patterns