Composite GiST Indexes for Geometry and Timestamp Columns

-- Naive: two separate indexes
CREATE INDEX idx_sensor_geom ON sensor_events USING GIST (location_geom);
CREATE INDEX idx_sensor_ts   ON sensor_events USING BTREE (recorded_at);

# Python: naively fetching all rows that match, then filtering in Python
import psycopg2
from shapely.geometry import box
from shapely import wkb

conn = psycopg2.connect(dsn="postgresql://user:pass@localhost/gisdb")
cur = conn.cursor()

# Fetches every geometry hit; timestamp filter happens client-side
cur.execute(
    "SELECT id, location_geom, recorded_at FROM sensor_events "
    "WHERE location_geom && ST_MakeEnvelope(%s, %s, %s, %s, 4326)",
    (-74.1, 40.6, -73.7, 40.9)
)
rows = cur.fetchall()  # may pull millions of rows across the network
filtered = [r for r in rows if start <= r[2] <= end]  # in-memory filter

-- Requires superuser or pg_extension_owner privilege
CREATE EXTENSION IF NOT EXISTS btree_gist;

-- Verify installation
SELECT extname, extversion
FROM pg_extension
WHERE extname = 'btree_gist';

-- Create composite GiST index: geometry column first, then timestamp
-- Geometry-first is optimal when spatial predicates are more selective
CREATE INDEX CONCURRENTLY idx_events_geom_ts
ON public.sensor_events
USING GIST (location_geom, recorded_at);

-- Alternatively, timestamp-first for narrow time-window workloads
-- (e.g. IoT ingestion where queries always target the last 15 minutes)
-- CREATE INDEX CONCURRENTLY idx_events_ts_geom
-- ON public.sensor_events
-- USING GIST (recorded_at, location_geom);

"""
composite_spatial_temporal.py
------------------------------
Query sensor_events using a composite GiST index on (location_geom, recorded_at).
Requires: sqlalchemy>=2.0, asyncpg, geoalchemy2, shapely
"""
from __future__ import annotations

from datetime import datetime
from typing import AsyncIterator

import asyncpg
from geoalchemy2 import Geometry
from geoalchemy2.functions import ST_MakeEnvelope
from sqlalchemy import Column, DateTime, Integer, Float, select, and_
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import DeclarativeBase


# ---------------------------------------------------------------------------
# ORM model
# ---------------------------------------------------------------------------

class Base(DeclarativeBase):
    pass


class SensorEvent(Base):
    __tablename__ = "sensor_events"

    id           = Column(Integer, primary_key=True)
    location_geom = Column(Geometry(geometry_type="POINT", srid=4326))
    recorded_at  = Column(DateTime(timezone=True), nullable=False)
    value        = Column(Float)


# ---------------------------------------------------------------------------
# SQLAlchemy 2.0 async — for FastAPI / Starlette services
# ---------------------------------------------------------------------------

async def fetch_events_sqlalchemy(
    session: AsyncSession,
    min_lon: float,
    min_lat: float,
    max_lon: float,
    max_lat: float,
    start_ts: datetime,
    end_ts: datetime,
    limit: int = 500,
) -> list[SensorEvent]:
    """
    Uses the composite GiST index on (location_geom, recorded_at).
    ST_MakeEnvelope generates the bounding box with explicit SRID=4326
    so the planner avoids an implicit cast that would bypass the index.
    """
    stmt = (
        select(SensorEvent)
        .where(
            and_(
                SensorEvent.location_geom.op("&&")(
                    ST_MakeEnvelope(min_lon, min_lat, max_lon, max_lat, 4326)
                ),
                SensorEvent.recorded_at.between(start_ts, end_ts),
            )
        )
        .order_by(SensorEvent.recorded_at.desc())
        .limit(limit)
    )
    result = await session.execute(stmt)
    return list(result.scalars().all())


# ---------------------------------------------------------------------------
# asyncpg — for pipeline / ETL code that needs maximum throughput
# Use a server-side cursor (cursor()) to avoid materialising the full result
# ---------------------------------------------------------------------------

async def stream_events_asyncpg(
    pool: asyncpg.Pool,
    min_lon: float,
    min_lat: float,
    max_lon: float,
    max_lat: float,
    start_ts: datetime,
    end_ts: datetime,
    prefetch: int = 200,
) -> AsyncIterator[asyncpg.Record]:
    """
    Streams rows via a server-side cursor; never pulls the full result set
    into memory. prefetch controls the wire batch size per round-trip.
    """
    sql = """
        SELECT id, location_geom::text AS geom_wkt, recorded_at, value
        FROM sensor_events
        WHERE location_geom && ST_MakeEnvelope($1, $2, $3, $4, 4326)
          AND recorded_at BETWEEN $5 AND $6
        ORDER BY recorded_at DESC
    """
    async with pool.acquire() as conn:
        async with conn.transaction():
            async for record in conn.cursor(
                sql,
                min_lon, min_lat, max_lon, max_lat,
                start_ts, end_ts,
                prefetch=prefetch,
            ):
                yield record

async with pool.acquire() as conn:
    await conn.execute("SET work_mem = '32MB'")
    await conn.execute("SET random_page_cost = 1.1")
    rows = await conn.fetch(
        "SELECT id, recorded_at FROM sensor_events "
        "WHERE location_geom && ST_MakeEnvelope($1,$2,$3,$4,4326) "
        "  AND recorded_at BETWEEN $5 AND $6",
        min_lon, min_lat, max_lon, max_lat, start_ts, end_ts,
    )

-- 1. Check the index exists and has been scanned
SELECT indexrelname, idx_scan, idx_tup_read, idx_tup_fetch
FROM pg_stat_user_indexes
WHERE relname = 'sensor_events'
  AND indexrelname = 'idx_events_geom_ts';

-- 2. Verify the planner chooses an Index Scan (not Bitmap Heap Scan or Seq Scan)
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT id, recorded_at
FROM sensor_events
WHERE location_geom && ST_MakeEnvelope(-74.05, 40.70, -73.90, 40.80, 4326)
  AND recorded_at BETWEEN '2024-01-01'::timestamptz
                      AND '2024-01-02'::timestamptz;

# Python timing comparison: before vs. after the composite index
import time, asyncpg, asyncio

async def benchmark(pool: asyncpg.Pool) -> None:
    sql = (
        "SELECT COUNT(*) FROM sensor_events "
        "WHERE location_geom && ST_MakeEnvelope(-74.05,40.70,-73.90,40.80,4326) "
        "  AND recorded_at BETWEEN '2024-01-01' AND '2024-01-02'"
    )
    t0 = time.perf_counter()
    async with pool.acquire() as conn:
        count = await conn.fetchval(sql)
    elapsed = (time.perf_counter() - t0) * 1000
    print(f"Rows: {count}  |  Query time: {elapsed:.1f} ms")