CREATE INDEX CONCURRENTLY on Large Spatial Tables

-- Wrong at this scale: freezes every write for the whole multi-hour build
CREATE INDEX idx_land_parcels_geom
ON land_parcels
USING GIST (geom);

# Wrong: CREATE INDEX CONCURRENTLY errors inside an implicit transaction
import psycopg

with psycopg.connect("dbname=cadastre") as conn:   # autocommit is False by default
    with conn.cursor() as cur:
        cur.execute(
            "CREATE INDEX CONCURRENTLY idx_land_parcels_geom "
            "ON land_parcels USING GIST (geom)"
        )
    # -> psycopg.errors.ActiveSqlTransaction:
    #    CREATE INDEX CONCURRENTLY cannot run inside a transaction block

"""Zero-downtime GiST build on a 100M-row land_parcels table (EPSG:32633).

Launches CREATE INDEX CONCURRENTLY on an autocommit connection and monitors
pg_stat_progress_create_index + pg_index.indisvalid from a second connection.
"""
import time
import logging
import psycopg

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

DSN = "host=db-primary dbname=cadastre user=migrator"
TABLE = "land_parcels"
INDEX = "idx_land_parcels_geom"


def drop_invalid_index(conn: psycopg.Connection) -> None:
    """Remove a leftover INVALID index from a prior failed build, if any."""
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT c.relname
            FROM pg_index i
            JOIN pg_class c ON c.oid = i.indexrelid
            WHERE i.indrelid = %s::regclass
              AND c.relname = %s
              AND i.indisvalid = false
            """,
            (TABLE, INDEX),
        )
        if cur.fetchone():
            log.warning("dropping leftover INVALID index %s", INDEX)
            cur.execute(f"DROP INDEX CONCURRENTLY {INDEX}")


def launch_build(conn: psycopg.Connection) -> None:
    """Issue the concurrent build. conn MUST be autocommit."""
    with conn.cursor() as cur:
        cur.execute("SET maintenance_work_mem = '2GB'")
        cur.execute("SET max_parallel_maintenance_workers = 4")
        log.info("starting CREATE INDEX CONCURRENTLY on %s", TABLE)
        cur.execute(
            f"CREATE INDEX CONCURRENTLY {INDEX} "
            f"ON {TABLE} USING GIST (geom)"
        )
        log.info("build statement returned")


def poll_progress(monitor: psycopg.Connection) -> None:
    """Print live phase/percent until no build is in flight."""
    while True:
        with monitor.cursor() as cur:
            cur.execute(
                """
                SELECT phase, blocks_done, blocks_total,
                       round(100.0 * blocks_done / NULLIF(blocks_total, 0), 1) AS pct
                FROM pg_stat_progress_create_index
                WHERE relid = %s::regclass
                """,
                (TABLE,),
            )
            row = cur.fetchone()
        if row is None:
            break
        phase, done, total, pct = row
        log.info("phase=%s blocks=%s/%s (%s%%)", phase, done, total, pct)
        time.sleep(10)


def confirm_valid(monitor: psycopg.Connection) -> bool:
    with monitor.cursor() as cur:
        cur.execute(
            """
            SELECT i.indisvalid, i.indisready
            FROM pg_index i
            JOIN pg_class c ON c.oid = i.indexrelid
            WHERE i.indrelid = %s::regclass AND c.relname = %s
            """,
            (TABLE, INDEX),
        )
        row = cur.fetchone()
    return bool(row and row[0] and row[1])


def main() -> None:
    # Separate monitor connection so polling never shares the build's session.
    with psycopg.connect(DSN, autocommit=True) as monitor:
        with psycopg.connect(DSN, autocommit=True) as builder:
            drop_invalid_index(builder)
            # Run the (blocking-in-Python) build; poll from the monitor meanwhile
            # by using a background thread or, more simply, launch the build and
            # poll after it returns. Here we poll after launch returns because
            # the statement blocks this connection until the build finishes.
            launch_build(builder)
        # Once launch returns, the build is done or failed; verify.
        if confirm_valid(monitor):
            with monitor.cursor() as cur:
                cur.execute(f"ANALYZE {TABLE}")
            log.info("index %s is VALID; table analyzed", INDEX)
        else:
            log.error("index %s is INVALID; run drop_invalid_index and retry", INDEX)


if __name__ == "__main__":
    main()

-- Run repeatedly from a second session while the build runs
SELECT phase, blocks_done, blocks_total,
       round(100.0 * blocks_done / NULLIF(blocks_total, 0), 1) AS pct,
       (now() - a.query_start) AS elapsed
FROM pg_stat_progress_create_index p
JOIN pg_stat_activity a USING (pid)
WHERE p.relid = 'land_parcels'::regclass;

-- 1. Index is valid and ready
SELECT c.relname, i.indisvalid, i.indisready
FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid
WHERE i.indrelid = 'land_parcels'::regclass
  AND c.relname = 'idx_land_parcels_geom';        -- both true

-- 2. Index size is sane for the row count
SELECT pg_size_pretty(pg_relation_size('idx_land_parcels_geom'));

-- 3. Planner uses it against a same-SRID envelope (EPSG:32633)
EXPLAIN (ANALYZE, BUFFERS)
SELECT parcel_id
FROM land_parcels
WHERE geom && ST_MakeEnvelope(380000, 5810000, 385000, 5815000, 32633);
-- Expect Bitmap Index Scan on idx_land_parcels_geom, not Seq Scan

import psycopg

with psycopg.connect(DSN, autocommit=True) as conn, conn.cursor() as cur:
    cur.execute(
        """
        SELECT i.indisvalid
        FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid
        WHERE i.indrelid = 'land_parcels'::regclass
          AND c.relname = 'idx_land_parcels_geom'
        """
    )
    row = cur.fetchone()
    assert row and row[0], "GiST index is missing or INVALID"
    print("idx_land_parcels_geom is valid")