Building Async Ingestion Pipelines for High-Throughput Queries

In modern database observability stacks, the async ingestion stage is the shock absorber between volatile query execution and deterministic baseline tracking: it receives raw EXPLAIN payloads, validates their structure, applies deterministic routing, and guarantees at-least-once delivery to durable storage queues. Building async ingestion pipelines for high-throughput queries requires strict stage isolation, deterministic partition routing, and resilient backpressure management. This stage operates exclusively between the initial telemetry extraction performed by distributed capture agents and the downstream normalization subsystems. It does not parse execution trees, compute cost estimates, or trigger regression alerts — those responsibilities belong to stages defined elsewhere in Automated EXPLAIN Capture & Storage Workflows.

Architectural Boundaries and Stage Isolation #

The architectural boundary is defined by two operational contracts. Upstream, the ingestion gateway consumes structured telemetry emitted by capture agents. Downstream, it emits validated, schema-compliant messages onto a broker topic that the normalization engine described in Normalizing Query Plans for Cross-Engine Comparison drains. Any deviation from this contract introduces latency variance and breaks the deterministic guarantees required for accurate performance regression tracking.

Strict isolation prevents cross-contamination between ingestion throughput and downstream analytical complexity. The ingestion worker pool must remain stateless with respect to query semantics: it treats every payload as an opaque byte stream until schema validation completes. This design ensures that CPU-bound normalization tasks never block I/O-bound ingestion sockets. Intermediate caching, plan diffing, and alert evaluation are explicitly prohibited at this layer. When the ingestion stage begins to perform semantic analysis, queue depths spike, tail latency degrades, and baseline tracking accuracy collapses. Maintaining this boundary is non-negotiable for teams managing multi-tenant database fleets.

The pipeline enforces a unidirectional data flow: $Capture Agent \to Ingestion Gateway \to Validation Router \to Broker Queue \to Normalization Worker$. Malformed payloads branch off to a dead-letter queue rather than propagating downstream.

Deterministic Routing and Schema Enforcement #

High-throughput environments generate heterogeneous plan formats across PostgreSQL, MySQL, and distributed SQL engines. Every incoming payload undergoes synchronous structural validation against a versioned JSON Schema registry before it is allowed onto the broker. This is a lightweight structural gate, distinct from the deeper contract checks performed by Schema Validation for Baseline Metadata; the ingestion layer only verifies that mandatory envelope fields are present and well-typed so that poison pills never reach persistent storage.

The mandatory envelope contract is intentionally small. Every message must carry query_hash (hex-encoded, produced upstream by the plan hashing algorithm), execution_timestamp (ISO-8601), plan_version, and raw_explain_output. The registered schema for v1 envelopes is:

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "https://queryplan.org/schema/explain-envelope/v1.json",
  "type": "object",
  "additionalProperties": true,
  "required": ["query_hash", "execution_timestamp", "plan_version", "raw_explain_output"],
  "properties": {
    "query_hash": { "type": "string", "pattern": "^[0-9a-f]{16,64}$" },
    "execution_timestamp": { "type": "string", "format": "date-time" },
    "plan_version": { "type": "string", "enum": ["v1", "v2"] },
    "engine": { "type": "string", "enum": ["postgresql", "mysql", "cockroachdb"] },
    "raw_explain_output": { "type": "object" }
  }
}

Routing relies on deterministic partition keys derived from query_hash. The formula partition_id = int(query_hash, 16) % NUM_PARTITIONS guarantees that all plan variants for a given query fingerprint land on the same partition and therefore the same processing worker. This preserves temporal ordering per fingerprint and simplifies the baseline delta calculations that feed regression thresholds. The routing table remains immutable during pipeline execution and refreshes only during controlled deployment windows, because changing NUM_PARTITIONS remaps every fingerprint and temporarily breaks ordering guarantees.

Production-Ready Async Implementation #

Platform teams typically implement this ingestion layer with Python’s asyncio ecosystem, using non-blocking I/O to service thousands of concurrent producer connections without thread contention. The core worker loop follows a predictable validate → accumulate → dispatch pattern, with batching on both a size and a time trigger so that low-traffic fingerprints are never stranded in a partial batch.

import asyncio
import json
import time
import structlog
from typing import List, Dict, Any
from jsonschema import validate, ValidationError
from opentelemetry import metrics, trace

# Configuration thresholds
MAX_BATCH_BYTES = 5_242_880      # 5 MB
FLUSH_INTERVAL_MS = 200
QUEUE_WATERMARK_HIGH = 0.85
PARTITION_COUNT = 12

logger = structlog.get_logger()
tracer = trace.get_tracer("ingestion_pipeline")
meter = metrics.get_meter("ingestion_pipeline")

# Metrics
batch_size_hist = meter.create_histogram("pipeline.batch.size.bytes")
validation_success = meter.create_counter("pipeline.validation.success")
validation_failure = meter.create_counter("pipeline.validation.failure")
queue_depth_gauge = meter.create_up_down_counter("pipeline.queue.depth")

class IngestionRouter:
    def __init__(self, schema_registry: Dict[str, Any], broker_client: Any):
        self.schema_registry = schema_registry
        self.broker = broker_client
        self.buffer: List[Dict[str, Any]] = []
        self.buffer_bytes = 0
        self.last_flush = time.monotonic()
        self.queue = asyncio.Queue(maxsize=10_000)

    async def _validate_payload(self, payload: Dict[str, Any]) -> bool:
        try:
            version = payload.get("plan_version", "v1")
            validate(instance=payload, schema=self.schema_registry[version])
            validation_success.add(1)
            return True
        except (ValidationError, KeyError) as e:
            validation_failure.add(1)
            logger.warning("schema_validation_failed", error=str(e),
                           payload_hash=payload.get("query_hash"))
            return False

    def _route_to_partition(self, query_hash: str) -> int:
        return int(query_hash, 16) % PARTITION_COUNT

    async def _flush_buffer(self):
        if not self.buffer:
            return

        with tracer.start_as_current_span("ingest_flush"):
            batch_size_hist.record(self.buffer_bytes)
            tasks = []
            for payload in self.buffer:
                partition = self._route_to_partition(payload["query_hash"])
                tasks.append(self.broker.produce(
                    topic="explain_plans_raw",
                    partition=partition,
                    value=json.dumps(payload).encode("utf-8"),
                    key=payload["query_hash"].encode("utf-8")
                ))

            results = await asyncio.gather(*tasks, return_exceptions=True)
            for i, res in enumerate(results):
                if isinstance(res, Exception):
                    logger.error("broker_dispatch_failed", error=str(res),
                                 payload=self.buffer[i]["query_hash"])
                    await self._send_to_dlq(self.buffer[i])

            self.buffer.clear()
            self.buffer_bytes = 0
            self.last_flush = time.monotonic()

    async def _send_to_dlq(self, payload: Dict[str, Any]):
        await self.broker.produce(
            topic="explain_plans_dlq",
            value=json.dumps(payload).encode("utf-8"),
            key=payload.get("query_hash", "unknown").encode("utf-8")
        )

    async def ingest(self, payload: Dict[str, Any]):
        payload_size = len(json.dumps(payload).encode("utf-8"))
        self.buffer.append(payload)
        self.buffer_bytes += payload_size

        # Backpressure check
        if self.queue.qsize() / self.queue.maxsize > QUEUE_WATERMARK_HIGH:
            logger.warning("backpressure_threshold_exceeded",
                           queue_depth=self.queue.qsize())
            await asyncio.sleep(0.05)  # Yield to allow consumer drain

        # Flush conditions: size threshold OR time threshold
        should_flush = (
            self.buffer_bytes >= MAX_BATCH_BYTES or
            (time.monotonic() - self.last_flush) * 1000 >= FLUSH_INTERVAL_MS
        )

        if should_flush:
            await self._flush_buffer()

    async def run_consumer_loop(self):
        while True:
            try:
                payload = await self.queue.get()
                if await self._validate_payload(payload):
                    await self.ingest(payload)
                else:
                    await self._send_to_dlq(payload)
                self.queue.task_done()
            except asyncio.CancelledError:
                await self._flush_buffer()
                break

Tracing spans propagate the query_hash as a baggage attribute so that a single fingerprint can be correlated end-to-end from initial capture through normalization without coupling the ingestion layer to downstream business logic.

Threshold Table and Alerting SLOs #

Ingestion pipelines require granular telemetry to detect degradation before it reaches the normalization stage. The following SLOs are the actionable ones; each maps directly to a named metric emitted by the implementation above.

A sustained breach of the queue-depth or flush-latency band indicates consumer starvation or broker latency; a validation-failure or DLQ breach indicates upstream capture-agent drift or a schema registry mismatch. Wire these bands into Prometheus alerting rules so the block band pages an operator:

groups:
  - name: explain-ingestion-slo
    rules:
      - alert: IngestionQueueSaturation
        expr: pipeline_queue_depth / 10000 > 0.85
        for: 60s
        labels: { severity: page }
        annotations:
          summary: "Ingestion queue > 85% for 60s — consumer starvation likely"
      - alert: IngestionValidationFailureRate
        expr: |
          rate(pipeline_validation_failure_total[5m])
            / rate(pipeline_validation_success_total[5m]) > 0.05
        for: 5m
        labels: { severity: page }
        annotations:
          summary: "Schema validation failure rate > 5% — audit schema registry / capture agents"
      - alert: IngestionFlushLatencyP99
        expr: histogram_quantile(0.99, rate(ingest_flush_duration_seconds_bucket[5m])) > 0.15
        for: 5m
        labels: { severity: warn }
        annotations:
          summary: "p99 batch flush latency > 150ms — thread pool or network congestion"

Failure Scenarios and Root Cause Analysis #

1. Consumer starvation and queue saturation. Symptom: pipeline.queue.depth climbs past the 0.85 watermark and stays there while producer throughput is flat. Diagnose with kafka-consumer-groups.sh --describe --group explain-ingestors and compare LAG per partition against the pipeline.queue.depth gauge. The usual root cause is a normalization worker pool that is undersized relative to producer volume. Mitigation: scale the consumer pool horizontally and confirm the batch flush is not blocking on a single slow partition; the built-in asyncio.sleep(0.05) yield only buys headroom, it does not fix a structural throughput deficit.

2. Poison-pill propagation. Symptom: a specific query_hash repeatedly appears in explain_plans_dlq with schema_validation_failed warnings. Diagnose by tailing the DLQ topic and grouping by rejection_reason. Root cause is almost always a capture-agent version skew that emits a plan_version the registry does not recognize. Mitigation: malformed payloads are never retried inline — they are serialized to the DLQ with a rejection_reason tag and replayed by a rate-limited job at 500 msg/sec with exponential backoff (2^n * 1s) once the registry is updated.

3. Broker partition unavailability. Symptom: bursts of broker_dispatch_failed errors with ConnectionRefusedError or TimeoutError. Diagnose with kafka-topics.sh --describe --topic explain_plans_raw and look for partitions whose leader is -1 (no in-sync replica). Root cause is a broker outage or an under-replicated topic. Mitigation: the gateway opens a circuit breaker after three consecutive connection failures and routes all incoming payloads to a local disk-backed spool; once the broker health check passes for 30 seconds the breaker transitions to half-open and admits a controlled trickle before full restoration.

4. Partition skew from a hot fingerprint. Symptom: one partition shows lag orders of magnitude higher than its peers. Diagnose by aggregating produced volume per partition_id over a 5-minute window. Root cause is a dominant query fingerprint (for example a health-check query) monopolizing a single partition because int(query_hash, 16) % PARTITION_COUNT is deterministic. Mitigation: sample or pre-aggregate the hot fingerprint upstream so it does not overwhelm its assigned worker; never salt the key, as that would break the per-fingerprint ordering guarantee.

5. Spool exhaustion during extended broker outages. Symptom: the disk-backed spool fills and ingest begins raising write errors. Diagnose with df -h on the spool volume and check the spool backlog gauge. Mitigation: enforce a bounded spool with an oldest-first eviction policy, and encrypt the spool at rest in line with the baseline data storage security boundaries so buffered plans do not become an exfiltration surface.

Configuration Reference #

For teams selecting and tuning the broker itself, Using Kafka for Async Query Plan Ingestion at Scale provides partition-count sizing, retention policies, and consumer-group benchmarks that pair directly with these knobs. Where an audit trail must run alongside the broker, Routing EXPLAIN ANALYZE Output to Centralized Logs covers the dual-write patterns that preserve a durable record without compromising ingestion throughput.

By adhering to strict stage isolation, deterministic partitioning, explicit backpressure thresholds, and predictable degradation paths, platform teams can construct ingestion pipelines that absorb query-execution volatility while delivering deterministic, schema-compliant payloads to baseline tracking systems.

Related #

• Using Kafka for Async Query Plan Ingestion at Scale — broker sizing, retention, and consumer-group lag tuning for this stage.
• Normalizing Query Plans for Cross-Engine Comparison — the downstream consumer that drains the broker topic this stage produces.
• Schema Validation for Baseline Metadata — the deeper contract-validation gate that complements the lightweight envelope check here.
• Routing EXPLAIN ANALYZE Output to Centralized Logs — dual-write audit patterns alongside the message broker.
• Plan Hashing Algorithms for SQL Engines — how the query_hash partition key is produced upstream.

← Back to Automated EXPLAIN Capture & Storage Workflows