Schema Validation for Baseline Metadata

Schema validation for baseline metadata is the gatekeeping stage of the capture pipeline: it verifies that every incoming query plan artifact structurally agrees with a versioned schema snapshot before that artifact is allowed to become part of a stored baseline. This stage does not execute queries, capture execution traces, or transform plan trees. It performs deterministic structural checks against an immutable schema manifest, so that downstream regression scoring always operates on artifacts whose table, column, index, and constraint references are known-good. Within Automated EXPLAIN Capture & Storage Workflows it sits between plan normalization and durable baseline storage, and it is the single point where schema drift is turned into an explicit, routable decision instead of a silent corruption.

Architectural Boundaries #

The validator is intentionally the narrowest stage in the pipeline. It consumes exactly two inputs and emits exactly one of three outcomes, and it holds no mutable state of its own. This isolation is what makes it safe to run inline on the hot path of baseline promotion.

Upstream (consumes). The stage reads a normalized plan envelope produced by the canonicalization step described in Normalizing Query Plans for Cross-Engine Comparison. Each envelope carries a stable query fingerprint, a deterministic plan hash derived from the plan hashing algorithm, the declared schema_version, and the flattened set of table, column, and index references the plan touches. The second input is a schema manifest resolved from a version-controlled registry (a Git-backed object store or a schema-registry table), keyed by that same schema_version.

Downstream (emits). The validator publishes the envelope, annotated with a validation verdict, to one of three destinations: accepted records advance to durable baseline storage under the encryption and access rules covered in Security Boundaries for Baseline Data Storage; warning-tagged records go to a drift-monitoring queue for reconciliation; rejected records are quarantined to a dead-letter topic. Only accepted envelopes reach the centralized log sink described in Routing EXPLAIN ANALYZE Output to Centralized Logs, which keeps regression dashboards free of structurally invalid noise.

Isolation guarantees. The validator never opens a connection to a live production database, never issues DDL, never mutates baseline tables, and never performs statistical sampling. All inputs are treated as immutable JSON or Parquet artifacts. Because of this, a schema-registry outage or a malformed manifest can only ever stop promotion — it can never rewrite history that has already been committed.

Deterministic Routing & Schema Enforcement #

Validation is governed by an explicit field contract and a fixed set of numeric thresholds. Nothing about the decision is heuristic: given the same envelope and the same manifest, the verdict is always identical and idempotent.

Envelope field contract #

The incoming envelope is validated against a JSON Schema before any structural comparison runs. Malformed payloads are rejected at parse time so that later stages can assume a well-formed record.

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "PlanEnvelope",
  "type": "object",
  "required": ["query_hash", "schema_version", "plan_hash",
               "referenced_tables", "referenced_columns", "referenced_indexes"],
  "additionalProperties": false,
  "properties": {
    "query_hash":        { "type": "string", "pattern": "^[0-9a-f]{64}$" },
    "plan_hash":         { "type": "string", "pattern": "^[0-9a-f]{64}$" },
    "schema_version":    { "type": "string", "pattern": "^[0-9]+\\.[0-9]+\\.[0-9]+$" },
    "referenced_tables": { "type": "array", "items": { "type": "string" } },
    "referenced_columns":{ "type": "array",
      "items": { "type": "object",
        "required": ["table", "column"],
        "properties": { "table": {"type":"string"}, "column": {"type":"string"} } } },
    "referenced_indexes":{ "type": "array", "items": { "type": "string" } }
  }
}

Routing thresholds #

Each envelope must pass three deterministic checks. Routing is decided by hard numeric boundaries so that drift classification is never subjective.

The final verdict is the most severe of the three per-check outcomes: any single reject dominates, otherwise any single warn dominates, otherwise accept.

Partition key for the baseline store #

Accepted records are written to a partitioned store so that a schema epoch can be reasoned about — and reconciled — as a unit. The partition key is derived deterministically from the envelope, never from wall-clock capture time:

partition_key = f"{engine}/{schema_version}/{query_hash[:2]}"

Sharding on the first byte of query_hash yields 256 balanced partitions per schema_version, and grouping under schema_version means that invalidating an entire schema epoch after a migration is a single-prefix delete rather than a full table scan.

Production-Ready Implementation #

The validator is best deployed as a stateless async consumer. The implementation below pulls envelopes from a queue, resolves the matching manifest from an asyncpg-backed schema registry (with a TTL cache), applies the deterministic checks, emits structured logs and OpenTelemetry spans/metrics, and returns a routed outcome. It is the happy path — fallback behaviour is covered in the failure section.

import time
from typing import Literal

import asyncpg
import structlog
from jsonschema import Draft202012Validator
from pydantic import BaseModel, ValidationError
from opentelemetry import trace, metrics

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

verdicts = meter.create_counter(
    "validation_results_total", description="Validation outcomes by status"
)
duration = meter.create_histogram(
    "validation_duration_seconds", description="End-to-end validation latency"
)
manifest_age = meter.create_gauge(
    "schema_manifest_age_seconds", description="Age of the resolved manifest"
)


class PlanEnvelope(BaseModel):
    query_hash: str
    plan_hash: str
    schema_version: str
    referenced_tables: list[str]
    referenced_columns: list[dict[str, str]]
    referenced_indexes: list[str]


class ValidationOutcome(BaseModel):
    status: Literal["ACCEPT", "WARN", "REJECT"]
    query_hash: str
    schema_version: str
    drift_score: float
    details: list[str]


class ManifestCache:
    """TTL cache over an asyncpg-backed schema registry."""

    def __init__(self, pool: asyncpg.Pool, ttl_seconds: int = 300) -> None:
        self._pool = pool
        self._ttl = ttl_seconds
        self._entries: dict[str, tuple[float, dict]] = {}

    async def resolve(self, schema_version: str) -> tuple[dict, float]:
        now = time.monotonic()
        cached = self._entries.get(schema_version)
        if cached and now - cached[0] < self._ttl:
            return cached[1], now - cached[0]

        row = await self._pool.fetchrow(
            "SELECT tables, active_indexes, constraints, fetched_at "
            "FROM schema_manifest WHERE schema_version = $1",
            schema_version,
        )
        if row is None:
            raise LookupError(f"no manifest for schema_version={schema_version}")

        manifest = {
            "tables": set(row["tables"]),
            "active_indexes": set(row["active_indexes"]),
            "constraints": row["constraints"],
        }
        self._entries[schema_version] = (now, manifest)
        return manifest, 0.0


_ENVELOPE_VALIDATOR = Draft202012Validator(schema=PLAN_ENVELOPE_SCHEMA)


def _classify(envelope: PlanEnvelope, manifest: dict) -> ValidationOutcome:
    details: list[str] = []

    missing = [t for t in envelope.referenced_tables if t not in manifest["tables"]]
    stale = [i for i in envelope.referenced_indexes if i not in manifest["active_indexes"]]
    constraint_drops = manifest["constraints"].get("dropped_since_capture", 0)

    if missing:
        details.append(f"missing_refs={len(missing)} tables={missing}")
        return ValidationOutcome(status="REJECT", query_hash=envelope.query_hash,
                                 schema_version=envelope.schema_version,
                                 drift_score=1.0, details=details)
    if constraint_drops > 0:
        details.append(f"constraint_drops={constraint_drops}")
        return ValidationOutcome(status="REJECT", query_hash=envelope.query_hash,
                                 schema_version=envelope.schema_version,
                                 drift_score=0.95, details=details)
    if len(stale) > 3:
        details.append(f"stale_refs={len(stale)} indexes={stale}")
        return ValidationOutcome(status="REJECT", query_hash=envelope.query_hash,
                                 schema_version=envelope.schema_version,
                                 drift_score=0.9, details=details)
    if stale:
        details.append(f"stale_refs={len(stale)} (soft drift)")
        return ValidationOutcome(status="WARN", query_hash=envelope.query_hash,
                                 schema_version=envelope.schema_version,
                                 drift_score=0.4, details=details)

    return ValidationOutcome(status="ACCEPT", query_hash=envelope.query_hash,
                             schema_version=envelope.schema_version,
                             drift_score=0.0, details=[])


async def validate_envelope(raw: dict, cache: ManifestCache) -> ValidationOutcome:
    started = time.perf_counter()
    with tracer.start_as_current_span("validate_baseline_metadata") as span:
        errors = sorted(_ENVELOPE_VALIDATOR.iter_errors(raw), key=lambda e: e.path)
        if errors:
            span.set_attribute("validation.status", "REJECT")
            verdicts.add(1, {"status": "REJECT", "reason": "malformed"})
            log.error("malformed_envelope", errors=[e.message for e in errors])
            return ValidationOutcome(status="REJECT", query_hash=raw.get("query_hash", "unknown"),
                                     schema_version=raw.get("schema_version", "unknown"),
                                     drift_score=1.0, details=["INVALID_PAYLOAD"])

        try:
            envelope = PlanEnvelope(**raw)
            manifest, age = await cache.resolve(envelope.schema_version)
        except (ValidationError, LookupError) as exc:
            span.set_attribute("validation.status", "REJECT")
            verdicts.add(1, {"status": "REJECT", "reason": "manifest"})
            log.error("manifest_resolution_failed", error=str(exc))
            return ValidationOutcome(status="REJECT", query_hash=raw.get("query_hash", "unknown"),
                                     schema_version=raw.get("schema_version", "unknown"),
                                     drift_score=1.0, details=[str(exc)])

        outcome = _classify(envelope, manifest)
        span.set_attribute("validation.status", outcome.status)
        span.set_attribute("validation.drift_score", outcome.drift_score)
        manifest_age.set(age, {"schema_version": envelope.schema_version})
        verdicts.add(1, {"status": outcome.status})
        duration.record(time.perf_counter() - started, {"status": outcome.status})
        log.info("validated", status=outcome.status, query_hash=envelope.query_hash,
                 schema_version=envelope.schema_version, drift_score=outcome.drift_score)
        return outcome

Thresholds & Alerting #

The stage runs inline on baseline promotion, so its own latency and error budgets are tight. The following SLOs are the exact values to alert on; they are numeric, not advisory.

A spike in the reject rate almost always points upstream: a schema migration landed without a matching manifest publish, so validate before you enable the CI gate wired in Validating Schema Changes Against Baseline Metadata. Encode these boundaries as alert rules alongside the service:

groups:
  - name: schema-validator
    rules:
      - alert: ValidatorRejectRateHigh
        expr: |
          sum(rate(validation_results_total{status="REJECT"}[15m]))
            / sum(rate(validation_results_total[15m])) > 0.05
        for: 15m
        labels: { severity: page, priority: P3 }
        annotations:
          summary: "Baseline validator reject rate above 5% — check for an unpublished manifest"
      - alert: SchemaManifestStale
        expr: max(schema_manifest_age_seconds) > 3600
        for: 5m
        labels: { severity: page, priority: P2 }
        annotations:
          summary: "Schema manifest older than 3600s — registry sync likely broken"
      - alert: ValidatorLatencyP99
        expr: histogram_quantile(0.99, sum(rate(validation_duration_seconds_bucket[10m])) by (le)) > 0.12
        for: 10m
        labels: { severity: page, priority: P3 }
        annotations:
          summary: "Validator p99 latency above 120ms — inspect manifest cache hit ratio"

Failure Scenarios & Root Cause Analysis #

Four failure modes account for nearly every validator incident. Each has a distinct signature in the metrics above.

1. Unpublished manifest after a migration. Symptom: reject rate jumps to 20–100% for a single schema_version immediately after a deploy; logs show missing_refs on tables that clearly exist. Diagnosis: SELECT schema_version, fetched_at FROM schema_manifest ORDER BY fetched_at DESC LIMIT 5; — the new version is absent. Mitigation: re-run the manifest publisher for that version; the migration pipeline should block promotion until the manifest write is confirmed.

2. Stale manifest cache (registry partition). Symptom: schema_manifest_age_seconds climbs past the TTL while reject and warn rates hold steady, then drift alerts fire late. Diagnosis: check registry reachability from the pod (pg_isready -h $REGISTRY_HOST) and the cache hit ratio in traces. Mitigation: the TTL cache serves last-known-good until max_manifest_age_seconds; beyond that, envelopes route to quarantine rather than being validated against outdated structure.

3. Malformed envelope from the normalizer. Symptom: validation_results_total{status="REJECT",reason="malformed"} rises while structural rejects stay flat. Diagnosis: inspect the JSON Schema errors in the malformed_envelope log line; a changed field name or a non-hex plan_hash is typical. Mitigation: pin the envelope schema version between the normalizer and validator, and treat a schema bump as a coordinated deploy.

4. Silent constraint drop counted as accept. Symptom: a downstream join-type regression appears in Detecting Join-Type Shifts in Execution Plans even though validation accepted the plan. Diagnosis: the manifest’s constraints.dropped_since_capture was never populated by the migration diff. Mitigation: make constraint-diff computation a required field of the manifest contract so a missing value fails validation instead of defaulting to zero.

Configuration Reference #

All tuning knobs are externalized so thresholds and fallbacks can be adjusted without a redeploy.

validator:
  manifest_cache_ttl_seconds: 300
  routing_thresholds:
    max_missing_refs: 0
    max_stale_indexes_warn: 3
    max_stale_indexes_reject: 3
    max_type_promotions: 2
  fallback:
    enabled: true
    max_manifest_age_seconds: 3600
    circuit_breaker_failures: 5

Safe Fallback & Drift Quarantine #

Network partitions, registry outages, or delayed DDL propagation can starve the validator of fresh manifests. These protocols keep the pipeline moving without ever compromising baseline integrity:

1. Last-known-good cache. If the registry is unreachable, serve the most recently validated manifest up to max_manifest_age_seconds. Beyond that ceiling, route every payload to quarantine rather than risk validating against outdated structure.
2. Circuit breaker. Track consecutive fetch failures. After circuit_breaker_failures (default 5), open the circuit and route all traffic to the drift_quarantine topic tagged validation_mode="BYPASS" — no envelope is silently accepted while the breaker is open.
3. Quarantine & reconciliation. Rejected and bypassed payloads land in a dead-letter queue with full context. A background job periodically re-validates them against the latest manifest: records that now pass are promoted; the rest are archived with a schema_mismatch tag.
4. Automated drift resolution. When soft-drift warnings accumulate for a fingerprint, trigger a diff against the DDL migration log. The reconciliation procedure — and the exact CI thresholds that gate a migration — are documented in Validating Schema Changes Against Baseline Metadata.

By enforcing strict isolation, deterministic routing, and disciplined fallback, this stage keeps baseline metadata a trustworthy foundation for the regression scoring that follows in Regression Detection & Rule Engines.

Related #

• Validating Schema Changes Against Baseline Metadata — the CI-gated runbook for migrations
• Normalizing Query Plans for Cross-Engine Comparison — the upstream stage that produces the envelope this validator consumes
• Routing EXPLAIN ANALYZE Output to Centralized Logs — the downstream sink for accepted records
• Security Boundaries for Baseline Data Storage — how accepted metadata is protected at rest
• Plan Hashing Algorithms for SQL Engines — the fingerprinting the envelope contract depends on

← Back to Automated EXPLAIN Capture & Storage Workflows