Handling Session Timeouts During Bulk Spatial Inserts

Bulk spatial inserts are one of the most reliable timeout triggers in session management for spatial data: large geometry payloads, GiST index maintenance, and WAL amplification routinely push a single INSERT past PostgreSQL’s default statement_timeout. This page solves that exact scenario — a Python ETL pipeline that loads thousands of geometry rows in one transaction and starts receiving sqlalchemy.exc.OperationalError: canceling statement due to statement timeout. The fix is not simply raising the threshold; it is decoupling transaction boundaries from connection lifetimes through deterministic chunking, transaction-scoped timeouts, and exponential backoff with idempotent conflict resolution.

Why the Naive Approach Fails

The instinctive approach is to collect all records, call session.bulk_insert_mappings() once, and commit. With scalar data this is fine. With PostGIS geometries it breaks for three compounding reasons.

Geometry parse cost accumulates. Each row requires WKT/WKB parsing into PostgreSQL’s internal GEOMETRY binary representation, bounding-box computation, and GiST index insertion — operations that are CPU-bound and proportional to vertex count. Five thousand simple points are manageable; five thousand complex building footprints are not.

The GiST index update is transactional. Unlike B-tree indexes that amortise updates across pages, GiST index optimisation forces every inserted row’s bounding box into the index tree within the same transaction. All pending index pages are held in shared memory until COMMIT, growing WAL volume and holding row-level locks.

Statement timeout fires on the query, not the commit. PostgreSQL starts the statement_timeout clock when the INSERT statement begins executing on the server. A 30-second default catches even moderate spatial loads.

Here is the fragile pattern that fails under production volume:

python
# BROKEN — single transaction for all rows, no timeout control
from sqlalchemy.orm import Session

def naive_bulk_insert(session: Session, records: list[dict], table) -> None:
    session.bulk_insert_mappings(table, records)   # may run for minutes
    session.commit()                               # never reached if timeout fires

When statement_timeout fires, psycopg2 raises QueryCanceled, SQLAlchemy wraps it in OperationalError, and the session is left in an aborted transaction state. Subsequent calls raise InvalidRequestError: Can't reconnect until invalid transaction is rolled back — a cascade that stops the entire pipeline.

Production-Ready Implementation

The following implementation chunks the dataset into fixed-size batches, scopes statement_timeout to each individual transaction with SET LOCAL, wraps each chunk in exponential backoff, and uses ON CONFLICT DO NOTHING to make retries safe. It is a single, copy-paste-ready module.

python
import time
import logging
from sqlalchemy import create_engine, text
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session
from sqlalchemy.exc import OperationalError

# ── Configuration ──────────────────────────────────────────────────────────────
CHUNK_SIZE        = 500        # rows per batch; reduce to 100–250 for complex polygons
MAX_RETRIES       = 3
BASE_DELAY_S      = 1.0        # seconds; doubles each retry (1 → 2 → 4)
STATEMENT_TIMEOUT = "60000"    # ms applied with SET LOCAL per transaction

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger(__name__)

# ── Engine (pool_pre_ping guards against stale connections after network blips) ─
engine = create_engine(
    "postgresql+psycopg2://user:pass@localhost:5432/gis_db",
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,   # validates connection before checkout
    pool_recycle=300,     # recycle idle connections every 5 minutes
)

# ── Idempotent chunked insert ──────────────────────────────────────────────────
def bulk_insert_spatial(
    session: Session,
    table,                # SQLAlchemy Table or mapped class __table__
    records: list[dict],  # each dict must include an "id" key and a "geom" WKB/WKT value
    unique_key: str = "id",
) -> int:
    """
    Insert spatial records in chunks.  Returns total rows accepted by the database.
    Each geometry dict must include SRID-qualified WKT, e.g.:
        {"id": 42, "geom": "SRID=4326;POINT(-73.985 40.748)", "name": "Empire State"}
    """
    total_inserted = 0
    total_chunks   = (len(records) + CHUNK_SIZE - 1) // CHUNK_SIZE

    for chunk_idx, offset in enumerate(range(0, len(records), CHUNK_SIZE)):
        chunk = records[offset : offset + CHUNK_SIZE]

        for attempt in range(MAX_RETRIES):
            try:
                # SET LOCAL applies only to the current transaction — resets on COMMIT/ROLLBACK
                session.execute(
                    text(f"SET LOCAL statement_timeout = '{STATEMENT_TIMEOUT}'")
                )

                # pg_insert from sqlalchemy.dialects.postgresql gives us on_conflict_do_nothing
                stmt = (
                    pg_insert(table)
                    .values(chunk)
                    .on_conflict_do_nothing(index_elements=[unique_key])
                )
                result = session.execute(stmt)
                session.commit()

                accepted = result.rowcount if result.rowcount >= 0 else len(chunk)
                total_inserted += accepted
                log.info(
                    "Chunk %d/%d committed: %d rows accepted",
                    chunk_idx + 1, total_chunks, accepted,
                )
                break  # success — exit retry loop

            except OperationalError as exc:
                session.rollback()  # clear poisoned transaction before retry
                msg = str(exc).lower()

                if (
                    "canceling statement due to statement timeout" in msg
                    or "terminating connection due to idle-in-transaction timeout" in msg
                    or "server closed the connection unexpectedly" in msg
                ):
                    delay = BASE_DELAY_S * (2 ** attempt)
                    log.warning(
                        "Chunk %d/%d timed out (attempt %d/%d). Retrying in %.1fs.",
                        chunk_idx + 1, total_chunks,
                        attempt + 1, MAX_RETRIES,
                        delay,
                    )
                    time.sleep(delay)
                else:
                    raise  # non-transient error — propagate immediately

        else:
            # All retries exhausted for this chunk
            log.error(
                "Chunk %d/%d failed after %d attempts — skipping.",
                chunk_idx + 1, total_chunks, MAX_RETRIES,
            )

    return total_inserted

Why each decision matters

SET LOCAL statement_timeout — The LOCAL modifier confines the GUC change to the current transaction. PostgreSQL restores the session default automatically on COMMIT or ROLLBACK, so no manual cleanup is required and the connection pool’s baseline behaviour is never altered.

pg_insert(...).on_conflict_do_nothing() — When a chunk is retried after a partial failure, some rows may already exist. Without conflict handling, the re-insert raises UniqueViolation and aborts the transaction. DO NOTHING discards the duplicate silently, and rowcount accurately reflects only the new rows accepted.

session.rollback() before retry — After any OperationalError SQLAlchemy marks the session as having an invalid transaction. Calling rollback() clears that state and returns the connection to a usable condition. Skipping this step guarantees InvalidRequestError on the next execute().

Exponential backoff — Retrying immediately after a timeout often hits the same overloaded state. Doubling the delay (1 s → 2 s → 4 s) gives PostgreSQL time to drain its WAL buffer and free shared memory pages.

Configuration and Tuning Knobs

Chunk size vs geometry complexity

Chunk size is not a single universal value — it is a function of average geometry vertex count and projected CRS complexity.

Geometry type Typical vertex count Recommended CHUNK_SIZE
Points (SRID=4326) 1 1000
Simple polygons 10–50 500
Building footprints 50–500 250
Road network multilinestrings 500–5000 50–100
Cadastral parcels with holes 1000+ 25–50

Monitor pg_stat_statements for the average execution time of your insert statements. If average chunk commit time exceeds 40 seconds on STATEMENT_TIMEOUT = "60000", halve CHUNK_SIZE.

Pool settings

python
engine = create_engine(
    DATABASE_URL,
    pool_size=10,          # steady-state connections
    max_overflow=5,        # burst connections during spike loads
    pool_pre_ping=True,    # sends SELECT 1 before checkout to detect dead connections
    pool_recycle=300,      # prevent OS-level TCP timeout on idle connections (5 min)
)

pool_pre_ping adds ~0.5 ms per checkout but prevents connection already closed errors that appear when the PostgreSQL server restarts or the network path drops idle sockets. For bulk ingestion jobs running in background workers this overhead is irrelevant.

WAL and checkpoint pressure

Bulk spatial inserts generate heavy WAL. Each row produces a WAL record for the heap page and another for the GiST index leaf page. Consider:

  • max_wal_size = 4GB — reduces checkpoint frequency during ingestion bursts; set in postgresql.conf or via ALTER SYSTEM.
  • synchronous_commit = off — acceptable for ETL pipelines where a crash replays from the source; do not use for write-ahead financial or audit data.
  • checkpoint_completion_target = 0.9 — spreads checkpoint I/O over 90% of the checkpoint interval, reducing write spikes that compete with insert throughput.

Apply these as session-level GUCs if you cannot change the server configuration:

python
# Applied once per session, before chunked inserts begin
with Session(engine) as session:
    session.execute(text("SET synchronous_commit = off"))
    session.execute(text("SET max_wal_size = '4GB'"))  # requires superuser
    bulk_insert_spatial(session, MyTable.__table__, records)

Transaction-scoped idle timeout

PgBouncer and AWS RDS Proxy enforce idle_in_transaction_session_timeout independently of statement_timeout. If your environment sets this below 30 seconds, chunked inserts can be killed between the SET LOCAL and the INSERT statement. Detect this with:

sql
SHOW idle_in_transaction_session_timeout;

If the value is below 5000 (5 seconds), raise it at the database level or negotiate with your DBA — there is no client-side workaround.

Verification Steps

After running bulk_insert_spatial, confirm correctness with two checks.

Row count assertion:

python
from sqlalchemy import func, select

def verify_insert_count(session: Session, table, expected: int) -> bool:
    actual = session.execute(select(func.count()).select_from(table)).scalar_one()
    if actual < expected:
        log.warning("Expected >= %d rows, found %d", expected, actual)
        return False
    log.info("Verification passed: %d rows in table", actual)
    return True

No idle-in-transaction sessions remain:

sql
SELECT pid, state, now() - query_start AS duration, left(query, 80) AS query
FROM pg_stat_activity
WHERE datname = 'gis_db'
  AND state = 'idle in transaction'
ORDER BY duration DESC;

This query should return zero rows immediately after a completed bulk run. Any row here indicates a session that did not commit or roll back cleanly — a potential source of autovacuum blocking on your geometry tables.

GiST index health after insert:

sql
SELECT
    indexname,
    pg_size_pretty(pg_relation_size(indexname::regclass)) AS index_size,
    idx_scan,
    idx_tup_read,
    idx_tup_fetch
FROM pg_stat_user_indexes
WHERE tablename = 'your_spatial_table';

If idx_scan is zero for a table that has been queried since the insert, GiST index optimisation has likely been bypassed — run VACUUM ANALYZE your_spatial_table to refresh planner statistics.

Chunk Flow Diagram

Chunked spatial insert flow with retry and commit per batch Flow diagram showing a list of geometry records being split into chunks, each chunk going through SET LOCAL timeout, bulk INSERT with ON CONFLICT DO NOTHING, a COMMIT, and on OperationalError a rollback and exponential backoff retry before the next chunk. records[] N geometries chunk chunk[i] 500 rows SET LOCAL timeout INSERT … ON CONFLICT COMMIT next chunk Timeout Error? OperationalError ROLLBACK yes backoff × 2ⁿ retry same chunk no more chunks no

Gotchas Checklist

  • SET LOCAL only works inside an explicit transaction. If autocommit=True is set on the connection or DBAPI layer, there is no enclosing transaction and SET LOCAL silently behaves as SET SESSION, polluting the pool-level default for every subsequent checkout.

  • rowcount can return -1 with psycopg2 for multi-row inserts with ON CONFLICT. Always guard with result.rowcount if result.rowcount >= 0 else len(chunk) rather than relying on the raw value for accurate accounting.

  • GiST index bloat accumulates after high-throughput insert/delete cycles. Run VACUUM ANALYZE your_spatial_table after large ingestion jobs. Bloated GiST pages degrade bounding-box filtering performance because the planner overestimates index cardinality. Monitor bloat with pgstattuple from the pgstattuple extension.

  • WKB geometry strings must carry explicit SRID prefixes. If geom values are passed as plain WKT without SRID=4326;, PostGIS inserts them with SRID 0, which breaks ST_DWithin radius searches and any downstream reprojection queries silently. Always use SRID=<srid>; prefixed extended WKT or pass geoalchemy2.WKTElement(wkt, srid=4326).

  • PgBouncer transaction-mode pooling intercepts SET LOCAL. In transaction-mode PgBouncer, connections are returned to the pool between statements. SET LOCAL issued in one statement may not be visible when the next statement acquires a different physical connection. Use session-mode pooling for bulk spatial jobs, or pass the timeout as a connect_args option on the engine instead.