Mapping EXPLAIN Costs to Real-World Latency Metrics

-- PostgreSQL: what the planner currently believes about I/O
SELECT name, setting, unit
FROM pg_settings
WHERE name IN ('random_page_cost', 'seq_page_cost',
               'effective_cache_size', 'cpu_tuple_cost');

-- Actual read pressure per statement
SELECT queryid, calls, mean_exec_time,
       shared_blks_hit, shared_blks_read,
       shared_blks_hit::float
         / NULLIF(shared_blks_hit + shared_blks_read, 0) AS hit_ratio
FROM pg_stat_statements
ORDER BY shared_blks_read DESC
LIMIT 20;

-- PostgreSQL: stale-statistics smell test
SELECT relname, n_live_tup, n_mod_since_analyze,
       last_analyze, last_autoanalyze
FROM pg_stat_user_tables
WHERE n_mod_since_analyze > 0.1 * GREATEST(n_live_tup, 1)
ORDER BY n_mod_since_analyze DESC;

# MySQL: histogram + cardinality currency for a suspect table
mysql -e "SELECT * FROM information_schema.column_statistics \
          WHERE table_name='orders'\G"
mysql -e "SHOW INDEX FROM orders" | awk '{print $3, $7}'

-- MySQL: reconstruct the execution profile EXPLAIN omits
SELECT DIGEST_TEXT,
       COUNT_STAR,
       AVG_TIMER_WAIT/1e9      AS avg_ms,
       SUM_ROWS_EXAMINED,
       SUM_ROWS_SENT
FROM performance_schema.events_statements_summary_by_digest
ORDER BY AVG_TIMER_WAIT DESC
LIMIT 20;

-- Confirm the gap between planned and launched workers
EXPLAIN (ANALYZE, BUFFERS, VERBOSE, FORMAT JSON)
SELECT ... ;   -- inspect "Workers Planned" vs "Workers Launched"

import asyncio
import math
from collections import deque
from dataclasses import dataclass
from typing import Deque, Dict, Optional

import asyncpg
import structlog
from opentelemetry import metrics, trace

log = structlog.get_logger("cost_latency_calibrator")
tracer = trace.get_tracer("queryplan.cost_latency")
meter = metrics.get_meter("queryplan.cost_latency")

CORRELATION_GAUGE = meter.create_gauge(
    "db_cost_latency_correlation",
    description="Rolling Pearson r between estimated cost and mean latency",
)
HIT_RATIO_GAUGE = meter.create_gauge(
    "db_buffer_hit_ratio",
    description="shared_blks_hit / (hit + read) for a tracked queryid",
)

CORRELATION_THRESHOLD = 0.75
MIN_SAMPLES = 30
WINDOW_SIZE = 1000


@dataclass(frozen=True)
class CostLatencySample:
    query_hash: str
    estimated_cost: float     # from EXPLAIN (FORMAT JSON), captured on a replica
    actual_latency_ms: float  # mean_exec_time from pg_stat_statements
    buffer_hit_ratio: float


class CostLatencyCalibrator:
    def __init__(self, pool: asyncpg.Pool) -> None:
        self._pool = pool
        self._windows: Dict[str, Deque[CostLatencySample]] = {}

    async def fetch_execution_stats(self, queryid: int) -> Optional[dict]:
        row = await self._pool.fetchrow(
            """
            SELECT mean_exec_time,
                   shared_blks_hit::float
                     / NULLIF(shared_blks_hit + shared_blks_read, 0) AS hit_ratio
            FROM pg_stat_statements
            WHERE queryid = $1
            """,
            queryid,
        )
        if row is None:
            return None
        return {
            "mean_exec_time_ms": float(row["mean_exec_time"]),
            "buffer_hit_ratio": float(row["hit_ratio"] or 0.0),
        }

    async def add_sample(
        self, query_hash: str, estimated_cost: float, queryid: int
    ) -> None:
        stats = await self.fetch_execution_stats(queryid)
        if stats is None:
            log.warning("no_stats_for_queryid", query_hash=query_hash, queryid=queryid)
            return
        window = self._windows.setdefault(query_hash, deque(maxlen=WINDOW_SIZE))
        window.append(
            CostLatencySample(
                query_hash=query_hash,
                estimated_cost=estimated_cost,
                actual_latency_ms=stats["mean_exec_time_ms"],
                buffer_hit_ratio=stats["buffer_hit_ratio"],
            )
        )
        HIT_RATIO_GAUGE.set(stats["buffer_hit_ratio"], {"query_hash": query_hash})

    def _pearson(self, window: Deque[CostLatencySample]) -> float:
        n = len(window)
        if n < MIN_SAMPLES:
            return 1.0  # insufficient data — assume stable
        xs = [s.estimated_cost for s in window]
        ys = [s.actual_latency_ms for s in window]
        mx, my = sum(xs) / n, sum(ys) / n
        cov = sum((x - mx) * (y - my) for x, y in zip(xs, ys))
        vx = math.sqrt(sum((x - mx) ** 2 for x in xs))
        vy = math.sqrt(sum((y - my) ** 2 for y in ys))
        if vx == 0.0 or vy == 0.0:
            return 1.0
        return cov / (vx * vy)

    async def evaluate_and_alert(self, query_hash: str) -> bool:
        with tracer.start_as_current_span("evaluate_correlation") as span:
            window = self._windows.get(query_hash, deque())
            corr = self._pearson(window)
            span.set_attribute("query_hash", query_hash)
            span.set_attribute("correlation", corr)
            CORRELATION_GAUGE.set(corr, {"query_hash": query_hash})
            if corr < CORRELATION_THRESHOLD:
                log.warning(
                    "cost_latency_regression",
                    query_hash=query_hash,
                    correlation=round(corr, 3),
                    threshold=CORRELATION_THRESHOLD,
                )
                return True
            return False


async def run_once(dsn: str, tracked: Dict[str, int]) -> None:
    pool = await asyncpg.create_pool(dsn, min_size=2, max_size=8)
    try:
        calibrator = CostLatencyCalibrator(pool)
        for query_hash, queryid in tracked.items():
            cost = await capture_explain_cost(pool, queryid)  # replica-side EXPLAIN
            await calibrator.add_sample(query_hash, cost, queryid)
            await calibrator.evaluate_and_alert(query_hash)
    finally:
        await pool.close()

groups:
  - name: query_cost_regression
    rules:
      - alert: CostLatencyMappingDegraded
        expr: db_cost_latency_correlation < 0.75
        for: 10m
        labels:
          severity: warning
          team: database-sre
        annotations:
          summary: "Optimizer cost model diverging from physical latency"
          description: "Query  correlation dropped to ; verify storage I/O and run ANALYZE."

-- Steer toward sequential scans when random I/O is genuinely slow
SET LOCAL random_page_cost = 12.0;
-- Refresh statistics that drove a bad estimate
ANALYZE VERBOSE orders (customer_id, status);

ANALYZE TABLE orders;
EXPLAIN FORMAT=JSON SELECT ... ;   -- read query_cost under cost_info

cluster_tier: gp3-16xlarge
random_page_cost: 12.0
seq_page_cost: 1.0
correlation_floor: 0.80
observed_slope_ms_per_cost: 0.34