Measuring GiST Index Bloat With pgstattuple

-- Slow and indiscriminate: full-scans every GiST index, business hours or not
SELECT * FROM pgstattuple('idx_vehicle_tracks_position_gist');
SELECT * FROM pgstattuple('idx_trip_segments_route_gist');

-- Cheap structural probe: run on all GiST indexes
SELECT
    'idx_vehicle_tracks_position_gist'::text        AS index_name,
    pg_relation_size('idx_vehicle_tracks_position_gist') AS index_bytes,
    ps.leaf_pages,
    ps.empty_pages,
    ps.deleted_pages,
    ps.avg_leaf_density,
    ps.leaf_fragmentation
FROM pgstatindex('idx_vehicle_tracks_position_gist') ps;

-- Authoritative full scan: run only on flagged indexes
SELECT
    table_len            AS index_bytes,
    tuple_count          AS live_entries,
    dead_tuple_count,
    dead_tuple_percent,
    free_space,
    free_percent
FROM pgstattuple('idx_vehicle_tracks_position_gist');

SELECT
    i.indexrelid::regclass::text AS index_name,
    t.relname                    AS table_name,
    pg_relation_size(i.indexrelid) AS index_bytes
FROM pg_index i
JOIN pg_class t  ON t.oid  = i.indrelid
JOIN pg_class ic ON ic.oid = i.indexrelid
JOIN pg_am   am  ON am.oid = ic.relam
WHERE am.amname = 'gist'
  AND t.relname IN ('vehicle_tracks', 'trip_segments')
ORDER BY pg_relation_size(i.indexrelid) DESC;

from __future__ import annotations

import sys
from dataclasses import dataclass

import psycopg
from psycopg.rows import dict_row

# --- Thresholds. Flag an index only when it is BOTH fragmented AND sparse,
#     and above a minimum size so tiny indexes are never rebuilt. ---
LEAF_FRAGMENTATION_MAX = 30.0    # percent; above this the tree is disordered
AVG_LEAF_DENSITY_MIN = 55.0      # percent; below this leaves are half empty
MIN_INDEX_BYTES = 256 * 1024 * 1024  # ignore indexes smaller than 256 MB

ENUMERATE_GIST = """
    SELECT
        i.indexrelid::regclass::text AS index_name,
        t.relname                    AS table_name,
        pg_relation_size(i.indexrelid) AS index_bytes
    FROM pg_index i
    JOIN pg_class t  ON t.oid  = i.indrelid
    JOIN pg_class ic ON ic.oid = i.indexrelid
    JOIN pg_am   am  ON am.oid = ic.relam
    WHERE am.amname = 'gist'
      AND t.relname = ANY(%(tables)s)
    ORDER BY pg_relation_size(i.indexrelid) DESC;
"""

# %s for the index name is safe here because the value comes from the catalog
# enumeration above, not from user input. pgstatindex takes a regclass.
PROBE_INDEX = """
    SELECT avg_leaf_density, leaf_fragmentation
    FROM pgstatindex(%s);
"""

FULL_SCAN = """
    SELECT dead_tuple_percent, free_percent
    FROM pgstattuple(%s);
"""


@dataclass
class Candidate:
    index_name: str
    table_name: str
    index_bytes: int
    avg_leaf_density: float
    leaf_fragmentation: float
    free_percent: float

    @property
    def approx_reclaimable_bytes(self) -> int:
        return int(self.index_bytes * self.free_percent / 100.0)


def find_bloated_indexes(dsn: str, tables: list[str]) -> list[Candidate]:
    candidates: list[Candidate] = []
    with psycopg.connect(dsn, row_factory=dict_row) as conn:
        # Keep the audit read-only and non-blocking.
        conn.read_only = True
        with conn.cursor() as cur:
            cur.execute(ENUMERATE_GIST, {"tables": tables})
            indexes = cur.fetchall()

        for idx in indexes:
            name = idx["index_name"]
            size = idx["index_bytes"]
            if size < MIN_INDEX_BYTES:
                continue

            # Cheap probe first.
            with conn.cursor() as cur:
                cur.execute(PROBE_INDEX, (name,))
                probe = cur.fetchone()

            frag = float(probe["leaf_fragmentation"])
            density = float(probe["avg_leaf_density"])
            if frag <= LEAF_FRAGMENTATION_MAX or density >= AVG_LEAF_DENSITY_MIN:
                continue  # not bloated enough to justify the full scan

            # Escalate to the authoritative scan only for suspects.
            with conn.cursor() as cur:
                cur.execute(FULL_SCAN, (name,))
                scan = cur.fetchone()

            candidates.append(
                Candidate(
                    index_name=name,
                    table_name=idx["table_name"],
                    index_bytes=size,
                    avg_leaf_density=density,
                    leaf_fragmentation=frag,
                    free_percent=float(scan["free_percent"]),
                )
            )
    return candidates


def emit_reindex_plan(candidates: list[Candidate]) -> str:
    if not candidates:
        return "-- No GiST indexes exceeded the bloat thresholds.\n"

    lines = ["-- REINDEX plan (review before running; each rebuild is online but I/O heavy)"]
    total_reclaim = 0
    for c in candidates:
        mb = c.index_bytes / (1024 * 1024)
        reclaim_mb = c.approx_reclaimable_bytes / (1024 * 1024)
        total_reclaim += c.approx_reclaimable_bytes
        lines.append(
            f"-- {c.index_name} on {c.table_name}: "
            f"{mb:.0f} MB, frag={c.leaf_fragmentation:.1f}%, "
            f"density={c.avg_leaf_density:.1f}%, free={c.free_percent:.1f}% "
            f"(~{reclaim_mb:.0f} MB reclaimable)"
        )
        lines.append(f"REINDEX INDEX CONCURRENTLY {c.index_name};")
    lines.append(f"-- Estimated total reclaim: {total_reclaim / (1024 * 1024):.0f} MB")
    return "\n".join(lines) + "\n"


if __name__ == "__main__":
    dsn = "postgresql://monitor:secret@replica.internal:5432/fleet"
    flagged = find_bloated_indexes(dsn, ["vehicle_tracks", "trip_segments"])
    sys.stdout.write(emit_reindex_plan(flagged))

with conn.cursor() as cur:
    cur.execute("SET statement_timeout = '600s'")

-- Before
SELECT avg_leaf_density, leaf_fragmentation FROM pgstatindex('idx_vehicle_tracks_position_gist');
SELECT free_percent FROM pgstattuple('idx_vehicle_tracks_position_gist');

-- Execute the plan
REINDEX INDEX CONCURRENTLY idx_vehicle_tracks_position_gist;

-- After: density should climb toward 90, fragmentation and free_percent should collapse
SELECT avg_leaf_density, leaf_fragmentation FROM pgstatindex('idx_vehicle_tracks_position_gist');
SELECT free_percent FROM pgstattuple('idx_vehicle_tracks_position_gist');