Statistical Distance Algorithms for Plan Drift Detection

Statistical-distance scoring measures how far a candidate plan’s cost, latency, and cardinality distributions have moved from an anchored baseline, collapsing a whole distribution shift into a single normalized drift score between 0 and 1. This guide specifies the distribution contract, the exact binning and smoothing math behind Jensen–Shannon divergence and the population stability index, a production asyncio scorer, and the precise numeric bands that turn a drift score into a PASS, WARN, or BLOCK routing flag.

The stage owns one narrow responsibility: given a bag of per-fingerprint samples for a metric, decide how anomalous the candidate distribution is relative to the anchored reference, and hand that decision downstream as a bounded score plus a flag. It does not capture plans, does not compute per-query cost deltas, and does not gate deploys itself. Where a scalar comparator asks “did the point estimate move,” this stage asks the sharper question “did the shape of the workload move” — the difference between a query that got slower on average and one whose tail collapsed into a bimodal cliff.

Architectural Boundaries #

Upstream, the scorer consumes windows of raw samples keyed by the deterministic plan fingerprint. The anchored baseline distribution lives in a read-mostly store; the candidate window arrives in the payload from the capture layer. Both are collections of numbers — total optimizer cost per execution, observed latency in milliseconds, or estimated-versus-actual cardinality ratios — never pre-aggregated means. Aggregation destroys exactly the tail information these measures exist to detect, so the boundary contract is samples-in, never summary statistics.

Downstream, the stage emits a DriftScore: the normalized divergence in [0.0, 1.0], the raw population stability index, and a routing flag. That payload fans out to the CI gate and the alert router. It deliberately overlaps with, but never replaces, the scalar path in Tracking Cost Deltas Across Baseline Versions: a query can pass the mean-delta check while failing the distribution check when variance widens without the average moving. Structural detectors such as graph-diff scoring for execution-plan trees are cross-referenced by the router when a high drift score coincides with an operator-tree change, which is a far stronger regression signal than either alone.

The scorer must be pure with respect to its two inputs — the candidate window and the anchored baseline row. Identical inputs must yield an identical score on any worker, in any epoch. The only I/O is the read-only baseline fetch; there is no write on the scoring path, no wall-clock dependency, and no cooldown state. That purity is what lets a suspicious BLOCK be replayed byte-for-byte during an incident review.

Deterministic Routing and Schema Enforcement #

The candidate payload is validated against a strict contract before any binning runs. A malformed sample array coerced to an empty distribution is the single most common cause of a false PASS, so the scorer rejects rather than guesses:

{
  "$id": "https://queryplan.org/schemas/plan-distribution-sample.json",
  "type": "object",
  "additionalProperties": false,
  "required": ["query_hash", "metric", "baseline_version", "candidate_samples"],
  "properties": {
    "query_hash":        { "type": "string", "pattern": "^[a-f0-9]{64}$" },
    "metric":            { "enum": ["cost", "latency", "cardinality"] },
    "baseline_version":  { "type": "string", "minLength": 1 },
    "candidate_samples": { "type": "array", "items": { "type": "number", "minimum": 0 }, "minItems": 1 },
    "captured_at":       { "type": "string", "format": "date-time" }
  }
}

Both distributions are quantized onto one shared grid so their probability vectors are index-aligned. The grid spans the union support of baseline and candidate, and each value maps to a bin with the deterministic formula bin_width = (support_hi − support_lo) / n_bins followed by bin_index = min(int((value − support_lo) / bin_width), n_bins − 1). Using the union support — rather than the baseline range alone — guarantees that a candidate outlier lands in the top bin instead of silently clamping and understating divergence.

Because every sample for a fingerprint must be scored against the same anchored baseline row, sharding uses a stable key derived directly from the 64-character SHA-256 hash: partition_key = int(query_hash[-8:], 16) % SHARD_COUNT. All windows for one fingerprint land on one worker and therefore read one baseline snapshot, eliminating cross-shard grid mismatch without any coordination. The measures the scorer computes, in strict order, are:

1. Jensen–Shannon divergence (base 2) — the symmetric, always-finite midpoint divergence, bounded in [0.0, 1.0]; this is the normalized drift score the flag is keyed on.
2. Population stability index — the natural-log stability metric, retained as a second opinion because it is far more sensitive to a single bin whose mass empties or explodes.
3. Wasserstein (earth-mover) distance — computed in one dimension from the sorted samples as an interpretable, unit-carrying diagnostic (milliseconds moved), surfaced in metadata but never used for gating because it is unbounded.
4. Kullback–Leibler divergence — the raw asymmetric term retained only inside the JS computation; it is never emitted alone because it diverges to infinity against any empty bin.

Production-Ready Implementation #

The scorer below is an asyncio service. It fetches the anchored baseline through an asyncpg pool, instruments every score with OpenTelemetry spans and metrics, and emits structured logs through structlog. The scoring path is pure; the only I/O is the read-only baseline fetch.

from __future__ import annotations

import math
import os
from dataclasses import dataclass, field
from typing import Literal, Sequence

import asyncpg
import structlog
from opentelemetry import metrics, trace

log = structlog.get_logger(__name__)
tracer = trace.get_tracer(__name__)
meter = metrics.get_meter(__name__)

DRIFT_SCORE = meter.create_histogram("plan_drift_score", unit="1")
DRIFT_VERDICT_TOTAL = meter.create_counter("plan_drift_verdict_total", unit="1")
DEGENERATE_TOTAL = meter.create_counter("plan_drift_degenerate_total", unit="1")

# --- Fixed scoring constants (mirror drift_scoring.yaml) ------------------
N_BINS = 32
SMOOTHING_EPSILON = 1e-9
MIN_SAMPLES = 200
JS_WARN_THRESHOLD = 0.1
JS_BLOCK_THRESHOLD = 0.25
PSI_WARN_THRESHOLD = 0.1
PSI_BLOCK_THRESHOLD = 0.25

Metric = Literal["cost", "latency", "cardinality"]


@dataclass(frozen=True)
class DistributionSample:
    query_hash: str
    metric: Metric
    baseline_version: str
    candidate_samples: tuple[float, ...]


@dataclass
class DriftScore:
    query_hash: str
    metric: Metric
    js_divergence: float
    psi: float
    drift_score: float          # normalized 0..1 (base-2 JS divergence)
    routing_flag: Literal["PASS", "WARN", "BLOCK"]
    metadata: dict = field(default_factory=dict)


def _bin_edges(lo: float, hi: float, n_bins: int = N_BINS) -> list[float]:
    hi = hi if hi > lo else lo + 1.0
    width = (hi - lo) / n_bins
    return [lo + width * i for i in range(n_bins + 1)]


def _histogram(samples: Sequence[float], edges: list[float]) -> list[float]:
    n_bins = len(edges) - 1
    lo, width = edges[0], (edges[-1] - edges[0]) / n_bins
    counts = [0.0] * n_bins
    for value in samples:
        idx = int((value - lo) / width) if width > 0 else 0
        counts[min(max(idx, 0), n_bins - 1)] += 1.0
    return counts


def _normalize(counts: list[float], epsilon: float = SMOOTHING_EPSILON) -> list[float]:
    smoothed = [c + epsilon for c in counts]           # never let a bin hit zero
    total = math.fsum(smoothed)
    return [c / total for c in smoothed]


def _kl_divergence(p: list[float], q: list[float]) -> float:
    return math.fsum(pi * math.log2(pi / qi) for pi, qi in zip(p, q) if pi > 0.0)


def jensen_shannon(p: list[float], q: list[float]) -> float:
    m = [(pi + qi) / 2.0 for pi, qi in zip(p, q)]
    # Base-2 JS divergence is bounded in [0.0, 1.0].
    return 0.5 * _kl_divergence(p, m) + 0.5 * _kl_divergence(q, m)


def population_stability_index(p: list[float], q: list[float]) -> float:
    # p = candidate, q = baseline; PSI convention uses the natural log.
    return math.fsum((pi - qi) * math.log(pi / qi) for pi, qi in zip(p, q))


class PlanDriftScorer:
    """Scores a candidate distribution against the anchored baseline."""

    def __init__(self, pool: asyncpg.Pool) -> None:
        self._pool = pool

    async def _baseline_samples(self, query_hash: str, metric: Metric) -> list[float]:
        rows = await self._pool.fetch(
            """
            SELECT sample_value
            FROM baseline_distribution
            WHERE query_hash = $1 AND metric = $2 AND is_anchored
            """,
            query_hash, metric,
        )
        return [float(r["sample_value"]) for r in rows]

    def _flag(self, js: float, psi: float) -> Literal["PASS", "WARN", "BLOCK"]:
        if js > JS_BLOCK_THRESHOLD or psi > PSI_BLOCK_THRESHOLD:
            return "BLOCK"
        if js > JS_WARN_THRESHOLD or psi >= PSI_WARN_THRESHOLD:
            return "WARN"
        return "PASS"

    async def score(self, candidate: DistributionSample) -> DriftScore:
        with tracer.start_as_current_span("plan_drift.score") as span:
            span.set_attribute("query_hash", candidate.query_hash)
            span.set_attribute("metric", candidate.metric)

            baseline = await self._baseline_samples(candidate.query_hash, candidate.metric)
            cand = list(candidate.candidate_samples)

            # Degenerate-support guard: never score against too-thin history.
            if len(baseline) < MIN_SAMPLES or len(cand) < MIN_SAMPLES:
                DEGENERATE_TOTAL.add(1, {"metric": candidate.metric})
                span.set_attribute("degenerate", True)
                log.warning(
                    "plan_drift_thin_support",
                    query_hash=candidate.query_hash,
                    baseline_n=len(baseline), candidate_n=len(cand),
                )
                return DriftScore(
                    candidate.query_hash, candidate.metric, 0.0, 0.0, 0.0,
                    "WARN", {"reason": "insufficient_samples"},
                )

            # Shared grid over the union support so both vectors are index-aligned.
            lo = min(min(baseline), min(cand))
            hi = max(max(baseline), max(cand))
            edges = _bin_edges(lo, hi)
            q = _normalize(_histogram(baseline, edges))   # baseline reference
            p = _normalize(_histogram(cand, edges))        # candidate

            js = jensen_shannon(p, q)
            psi = population_stability_index(p, q)
            flag = self._flag(js, psi)

            DRIFT_SCORE.record(js, {"metric": candidate.metric})
            DRIFT_VERDICT_TOTAL.add(1, {"verdict": flag, "metric": candidate.metric})
            span.set_attribute("js_divergence", js)
            span.set_attribute("psi", psi)
            span.set_attribute("routing_flag", flag)
            log.info(
                "plan_drift_scored",
                query_hash=candidate.query_hash, metric=candidate.metric,
                js=round(js, 4), psi=round(psi, 4), flag=flag,
            )
            return DriftScore(
                candidate.query_hash, candidate.metric, js, psi, js, flag,
                {"bins": N_BINS, "support_lo": lo, "support_hi": hi},
            )


async def build_pool() -> asyncpg.Pool:
    return await asyncpg.create_pool(
        dsn=os.environ["DRIFT_DSN"],
        min_size=2,
        max_size=int(os.environ.get("DRIFT_POOL_MAX", "8")),
        command_timeout=5.0,
    )

Two properties make this safe under load. First, a thin candidate or baseline window never emits a silent PASS; too few samples always surface as WARN so a genuinely regressed but rarely-run query cannot slip through unscored. Second, the additive smoothing floor of 1e-9 guarantees the KL terms inside the JS computation are always finite even when a baseline bin is empty — the divergence that would otherwise blow up to infinity stays bounded in [0.0, 1.0].

Threshold Reference #

The gating bands are fixed and identical across metrics; the operational SLOs below describe when the scorer itself is unhealthy, independent of the drift it reports.

A sustained BLOCK share above 5 % almost always means the anchored baseline is stale relative to the fleet — usually a fleet-wide ANALYZE shifted every cost distribution at once — not that the fleet simultaneously regressed. Encode that guard as a Prometheus alert:

groups:
  - name: plan-drift-scorer
    rules:
      - alert: PlanDriftBlockStorm
        expr: |
          (
            sum(rate(plan_drift_verdict_total{verdict="BLOCK"}[15m]))
            /
            sum(rate(plan_drift_verdict_total[15m]))
          ) > 0.05
        for: 10m
        labels:
          severity: page
        annotations:
          summary: "BLOCK drift share above 5% for 10m — baseline likely stale, not the fleet"
          runbook: "Re-anchor baseline_distribution after the last fleet ANALYZE before re-gating."

      - alert: PlanDriftDegenerateSupports
        expr: |
          sum(rate(plan_drift_degenerate_total[10m]))
          /
          sum(rate(plan_drift_verdict_total[10m])) > 0.1
        for: 10m
        labels:
          severity: ticket
        annotations:
          summary: "Over 10% of fingerprints scored on <200 samples — widen the capture window"

The static bands here feed the adaptive machinery in Defining Regression Thresholds for Query Plans; the scalar noise filter that consumes the flag downstream is Tuning Thresholds for False Positive Reduction.

Failure Scenarios and Root Cause Analysis #

1. Block storm after a statistics refresh. Symptom: BLOCK share jumps from ~1 % to ~40 % minutes after an ANALYZE sweep. Root cause: the planner re-costs every plan at once, shifting all cost distributions while the anchored baseline still reflects the old estimates. Diagnose by correlating the spike against stats timestamps:

SELECT relname, last_analyze
FROM pg_stat_user_tables
WHERE last_analyze > now() - interval '30 minutes'
ORDER BY last_analyze DESC;

Mitigation: gate a re-anchor of baseline_distribution behind the stats event so the reference is re-seeded from the new plans rather than fighting them.

2. Degenerate thin support. Symptom: plan_drift_degenerate_total climbs and a batch job’s regression scores WARN forever, never BLOCK. Root cause: the candidate window carries fewer than MIN_SAMPLES (200) executions, so PSI is dominated by smoothing noise. Diagnose the window depth:

SELECT query_hash, count(*) AS n
FROM plan_cost_sample
WHERE captured_at > now() - interval '1 hour' AND NOT is_anchored
GROUP BY query_hash
HAVING count(*) < 200
ORDER BY n ASC;

Mitigation: extend the capture window for low-frequency fingerprints, or accept the forced WARN as correct behavior for cold paths.

3. Mismatched supports collapse the histogram. Symptom: an obvious latency blow-out scores a low JS divergence near 0.03. Root cause: a single candidate outlier stretches the union support so wide that almost all mass lands in one bin, flattening both densities. Diagnose the range spread:

SELECT min(total_cost), max(total_cost),
       percentile_cont(0.99) WITHIN GROUP (ORDER BY total_cost) AS p99
FROM plan_cost_sample
WHERE query_hash = $1 AND NOT is_anchored;

Mitigation: winsorize the candidate at the 99th percentile before binning so one runaway sample cannot dominate the grid.

4. Empty-bin PSI explosion. Symptom: PSI reports an implausibly large value like 18.7 while JS stays modest. Root cause: a baseline bin is genuinely empty and the natural-log ratio detonates when smoothing is misconfigured to zero. Diagnose the smoothing constant in effect:

grep -E 'SMOOTHING_EPSILON|smoothing_epsilon' /etc/queryplan/drift_scoring.yaml

Mitigation: keep smoothing_epsilon at 1e-9; treat PSI above 10.0 as a data-quality signal, not a regression, and fall back to the JS flag.

5. Cross-shard baseline divergence. Symptom: the same fingerprint yields different scores on retry. Root cause: SHARD_COUNT changed without draining, so windows now hash to a worker holding a different anchored snapshot. Mitigation: treat SHARD_COUNT as immutable at runtime and re-shard only through a full drain-and-reseed of baseline_distribution.

Configuration Reference #

Widen js_block_threshold and psi_block_threshold together only when a re-anchor has already ruled out a stale baseline; tighten n_bins and min_samples when chasing missed tail regressions. Move one axis at a time so the calibration stays reproducible against replayed sample windows.

Related #

• Scoring Plan Drift with Jensen-Shannon Divergence — the focused runbook on binning, zero-bin smoothing, and base-2 versus base-e for the cost distribution.
• Tracking Cost Deltas Across Baseline Versions — the scalar comparator this distribution scorer runs alongside.
• Tuning Thresholds for False Positive Reduction — the downstream noise filter that consumes the routing flag.
• Graph-Diff Scoring for Execution-Plan Trees — the structural detector cross-referenced when drift coincides with an operator-tree change.
• Defining Regression Thresholds for Query Plans — the threshold model the static drift bands feed.

← Back to Regression Detection & Rule Engines