Part IV · Building It · Chapter 7

The Typed Query Plan

The model emits a plan, not a query string. The plan is type-checked, authorized, annotated, and only then compiled to SQL. Everything in this book becomes a pre-execution check.

This chapter is the capstone. We take the five annotations from Chapters 2–5, the semiring framework from Chapter 6, and wire them into a concrete system: a typed intermediate representation that sits between the LLM and execution. The IR is the trust boundary. Everything before it is uncertain; everything after it is validated.

7.1 Architecture: the three-stage pipeline

The system has three stages:

┌─────────────┐    ┌──────────────┐    ┌────────────┐
│ LLM emits   │───▶│ Validator    │───▶│ Executor   │
│ typed plan  │    │ (type-check, │    │ (compile   │
│ (JSON IR)   │    │  annotate,   │    │  to SQL,   │
│             │    │  authorize)  │    │  run,      │
└─────────────┘    └──────────────┘    │  narrate)  │
                          │             └────────────┘
                          │ rejection
                          ▼
                   ┌──────────────┐
                   │ Honest       │
                   │ refusal or   │
                   │ qualified    │
                   │ answer       │
                   └──────────────┘

The critical design decision: the LLM never emits SQL. It emits a typed plan in a constrained IR. The validator checks the plan against the metric registry, the grain catalog, and the authorization model. Only valid plans proceed to execution.

Why this matters:

• A plan can be checked before execution — no wasted compute on invalid queries.
• A plan can be annotated with completeness, confidence, and guarantee levels — the narrator knows its constraints before generating text.
• A plan can be logged, hashed, and replayed — audit and debugging are trivial.
• A plan cannot hide semantic errors inside SQL syntax — the structure is explicit.

7.2 The IR schema

{
  "plan": {
    "op": "aggregate",
    "function": "sum",
    "measure": "arr",
    "groups": ["account_id"],
    "input": {
      "op": "join",
      "type": "inner",
      "key": "account_id",
      "left": {
        "op": "scan",
        "source": "warehouse",
        "table": "monthly_account_arr",
        "filters": [
          {"field": "month", "op": "eq", "value": "2026-07"}
        ]
      },
      "right": {
        "op": "entity_join",
        "source": "vector_store",
        "query": "churn risk, non-renewal",
        "k": 50,
        "theta": 1.0,
        "link_config": {
          "methods": ["verified_domain"],
          "max_hops": 1
        }
      }
    }
  },
  "principal": "user:sarah@acme.com",
  "request_id": "req_abc123"
}

Every node in the plan has a type. The validator walks the tree and produces either a validated plan (with annotations) or a rejection.

7.3 The validator

from dataclasses import dataclass
from typing import Union

@dataclass
class ValidatedPlan:
    plan: dict
    annotations: "FullAnnotation"
    compiled_sql: str
    qualifiers: list[str]  # what the narrator must say

@dataclass
class Rejection:
    reason: str
    suggestion: str  # what the user could do instead

def validate(plan: dict, principal: str, registry: MetricRegistry,
             authz: AuthorizationModel) -> Union[ValidatedPlan, Rejection]:
    """Validate a query plan. Returns ValidatedPlan or Rejection."""

    # Phase 1: Structural validation
    error = check_structure(plan)
    if error:
        return Rejection(error, "Reformulate the question.")

    # Phase 2: Measure validation (additivity, grain)
    error = check_measures(plan, registry)
    if error:
        return Rejection(error.reason, error.suggestion)

    # Phase 3: Authorization (guarantee annotation)
    guarantee = compute_guarantee(plan, principal, authz)

    # Phase 4: Completeness annotation
    completeness = compute_completeness(plan)

    # Phase 5: Confidence annotation
    confidence = compute_confidence(plan)

    # Phase 6: Negation check
    if has_negation(plan):
        neg_inputs = get_negation_inputs(plan)
        for inp in neg_inputs:
            g = compute_guarantee(inp, principal, authz)
            if g.level != GuaranteeLevel.FULL:
                return Rejection(
                    f"Cannot compute this query: it requires determining "
                    f"what does NOT exist in a data source where your access "
                    f"is restricted.",
                    f"Request access to the underlying data, or reformulate "
                    f"without a NOT/anti-join."
                )

    # Phase 7: Aggregation gate
    agg_nodes = find_aggregation_nodes(plan)
    for node in agg_nodes:
        input_completeness = compute_completeness(node["input"])
        if not input_completeness.allows_aggregation(node["function"]):
            return Rejection(
                f"Cannot compute {node['function']} over this data: "
                f"the input is incomplete ({input_completeness.level.name}).",
                f"Use a bounded result (e.g., 'at least X') or request "
                f"an exhaustive retrieval."
            )

    # Phase 8: Compile and annotate
    annotations = FullAnnotation(
        completeness=completeness.level,
        confidence=confidence.value,
        guarantee=guarantee.level,
    )

    qualifiers = build_qualifiers(annotations, plan)
    sql = compile_to_sql(plan, registry)

    return ValidatedPlan(plan, annotations, sql, qualifiers)

7.4 Building qualifiers

The validator produces a list of qualifiers — statements the narrator must include in its response. These are not suggestions; they are structural constraints:

def build_qualifiers(annotations: FullAnnotation, plan: dict) -> list[str]:
    qualifiers = []

    if annotations.completeness < CompletenessLevel.EXACT:
        k = extract_k(plan)
        qualifiers.append(
            f"Result based on top {k} matching items. "
            f"Counts and sums are lower bounds."
        )

    if annotations.confidence < 1.0:
        qualifiers.append(
            f"Cross-system identity matching used (confidence: "
            f"{annotations.confidence:.0%}). Some attributions may be "
            f"incorrect."
        )

    if annotations.guarantee < GuaranteeLevel.FULL:
        qualifiers.append(
            "Your access to some data sources is restricted. "
            "Results reflect only data visible to you."
        )

    return qualifiers

7.5 Making unsound results unnarratable

The qualifiers are enforced in the harness — the layer between the validated result and the LLM’s response generation:

def generate_response(result: ValidatedPlan, llm) -> str:
    """Generate a response with enforced qualifiers."""

    # The LLM generates the analytical content
    analysis = llm.generate(
        system="""You are narrating a data query result.
        You MUST include the following qualifiers in your response.
        Do not report any number without its associated qualifier.
        Do not say 'exactly' or 'precisely' when a qualifier indicates
        a bound or approximation.""",
        context={
            "data": result.compiled_sql,  # or the actual result rows
            "qualifiers": result.qualifiers,
        }
    )

    # Post-processing: verify qualifiers are present
    for q in result.qualifiers:
        if not qualifier_present(analysis, q):
            # Force-append the qualifier
            analysis += f"\n\nNote: {q}"

    return analysis

The key principle: enforce this in the harness, not in the system prompt. A system prompt is a suggestion. The harness is a wall. The post-processing step guarantees that even if the LLM ignores the system prompt, the qualifier appears.

7.6 The metric registry as source of truth

The validator resolves every measure reference through the registry. This gives you:

• One definition of ARR. Version-controlled. With an owner.
• Compile-time additivity checking. The model cannot accidentally sum a stock.
• Automatic time intelligence. “ARR this quarter” resolves to the latest month without the model knowing the rule.
• Grain enforcement. Joins that would produce fan-out are rejected before execution.

The hard part is organizational, not technical:

# Registry ownership model
registry:
  ownership: federated  # each team owns their measures
  review_gate: true     # changes to shared measures require review
  shared_measures:      # measures used across teams
    - arr
    - revenue
    - active_customers
  review_required_by: data-platform-team

A registry owned by nobody drifts back into per-team SQL within two quarters. A registry owned by a central team becomes a ticket queue that people route around. Federated ownership with a review gate on shared measures is the shape that tends to survive.

7.7 Adaptive abstention

Chen, Chen, Koudas and Yu (SIGMOD 2025) formalize adaptive abstention — formal guarantees about when the system refuses rather than answers. Refusal is a first-class, guaranteed behavior.

In our system, abstention happens at three levels:

1. Plan-level: the validator rejects an invalid plan. The agent receives a structured rejection and can try a different approach.
2. Result-level: the plan is valid but the result carries qualifiers. The agent answers with caveats.
3. Confidence-level: the annotations indicate the result is so uncertain it is not useful. The agent refuses to commit to a number.

def should_abstain(annotations: FullAnnotation) -> bool:
    """Should the system refuse rather than answer?"""
    # Refuse if completeness is unknown
    if annotations.completeness == CompletenessLevel.UNKNOWN:
        return True
    # Refuse if confidence is below threshold
    if annotations.confidence < 0.5:
        return True
    # Refuse if guarantee is none
    if annotations.guarantee == GuaranteeLevel.NONE:
        return True
    return False

7.8 Binding patterns and answerability

Not every question is answerable given your available data sources and their access patterns. An API that only supports lookup-by-ID cannot answer “list all customers in APAC.” A vector store that requires a query embedding cannot enumerate its contents.

This is the binding pattern problem from the data integration literature (Florescu et al., SIGMOD 1999): given sources with limited access patterns, which queries are answerable?

data_sources:
  salesforce_api:
    access_patterns:
      - get_account_by_id: {input: [account_id], output: [name, region, arr]}
      - list_accounts_by_region: {input: [region], output: [account_id, name]}
      - search_accounts: {input: [query_string], output: [account_id, name]}
    # CANNOT: enumerate all accounts without a filter

  vector_store:
    access_patterns:
      - semantic_search: {input: [query, k], output: [chunk_id, score]}
    # CANNOT: enumerate all chunks, or filter without a query

  warehouse:
    access_patterns:
      - sql_query: {input: [sql], output: [rows]}
    # CAN: arbitrary queries (full access pattern)

Before planning a query, check answerability:

def is_answerable(question_plan: dict, sources: dict) -> tuple[bool, str]:
    """Check if the plan's data requirements can be satisfied by available sources."""
    for leaf in get_leaf_nodes(question_plan):
        source = sources[leaf["source"]]
        required_access = infer_access_pattern(leaf)
        if not source.supports(required_access):
            return False, (
                f"Cannot answer: {leaf['source']} does not support "
                f"{required_access}. Available patterns: "
                f"{source.access_patterns.keys()}"
            )
    return True, ""

Reject early rather than retry in loops. The current state of agent frameworks is to discover unanswerable questions through repeated failures and timeouts. Checking answerability at plan time is instant and definitive.

7.9 The honest answer

For reference: what the system produces for the original question when everything is working.

Less quotable. Considerably more useful. Nobody forwards it to the CRO without reading it first, which is the point.

---

Exercises

1. Implement the full three-stage pipeline for a simplified domain: a warehouse with 3 tables (accounts, monthly_arr, orders), a vector store for support conversations, and a metric registry with 5 measures. The LLM emits a plan (you can hardcode test plans rather than wiring a real LLM), the validator checks it, and valid plans compile to SQL. Test with:
   • A valid plan (simple SUM of revenue by account) → executes
   • A plan that sums ARR across months → rejected (additivity)
   • A plan that anti-joins with a table the user can’t fully see → rejected (guarantee)
   • A plan that aggregates over vector search results → qualified (completeness)
2. Implement the qualifier enforcement in the harness. Given a result with qualifiers, generate a mock response (template-based is fine) and verify that all qualifiers appear. Implement the force-append fallback for when a qualifier is missing from the generated text.

1. Design the IR for your organization’s data landscape. What operations does it need beyond what’s shown here? Consider: time-series windowing, percentile computation, cohort definition, funnel analysis. Each adds new nodes to the IR and potentially new validation rules. What is the minimal IR that covers 80% of your team’s analytical questions?
2. The validator currently rejects invalid plans entirely. Design a plan repair mechanism: when the validator rejects, it returns a structured suggestion that the LLM can use to emit a corrected plan. For example, “SUM across time rejected → suggest using time_rule=last” becomes a concrete edit the LLM can apply. How many rejection types can be automatically repaired vs. requiring the user to reformulate?

---