Prompt-to-Spatial-SQL Generation

Generated SQL is the most powerful tool a spatial agent has and the one with the largest blast radius. A query that is syntactically valid, semantically wrong and unbounded in cost will run happily for four minutes and return an answer nobody can check. This topic is about constraining generation so that everything reaching the database is expressible, safe, index-aware and cheap enough to run.

It belongs to geospatial prompt engineering and tool routing and assumes the positions in a question have already been resolved by geocoding and place-name resolution — a generated query containing a model-recalled coordinate is a fast way to produce a confident answer about the wrong place.

Foundational Principles

Generate a query shape, not free text. The model should emit the parts of a query — predicate, filters, ordering, limit — into a template the application assembles, rather than a string the application executes. That single change removes injection, makes the allow-list enforceable, and turns validation from parsing into checking fields.

Every value is bound, never interpolated. Geometry, extents, dates and identifiers arrive as parameters. A query built by string formatting is unsafe even when the model is well-behaved, because the values frequently come from documents rather than from the model.

Index-awareness is part of correctness. A spatial predicate written without a bounding-box operator may or may not use the index depending on the planner’s mood, and the difference is between forty milliseconds and forty seconds. Emit the index-aware form always; check the plan before running.

Step-by-Step Implementation Pipeline

1. Define the query shapes the agent may produce

A small set of shapes covers the overwhelming majority of spatial questions, and constraining generation to them is what makes everything downstream tractable.

import logging
from dataclasses import dataclass
from typing import Literal, Optional, Sequence

log = logging.getLogger("spatial_sql")

Shape = Literal["features_in_region", "nearest_to", "aggregate_in_region", "attribute_lookup"]

ALLOWED_PREDICATES = {"intersects", "within", "contains", "dwithin"}
ALLOWED_AGGREGATES = {"count", "sum_area", "min", "max", "median"}


@dataclass(frozen=True)
class QuerySpec:
    shape: Shape
    table: str
    predicate: Optional[str] = None
    columns: tuple[str, ...] = ()
    filters: tuple[tuple[str, str, object], ...] = ()   # (column, op, value)
    aggregate: Optional[str] = None
    order_by_distance: bool = False
    limit: int = 100

2. Validate the specification against the schema

Validation happens against the real schema rather than against a list, so a renamed column produces a rejection rather than a query that fails at execution — or worse, one that silently matches nothing.

class SpecRejected(ValueError):
    """The generated specification cannot be turned into a safe query."""


def validate(spec: QuerySpec, schema: dict[str, set[str]],
             max_limit: int = 500) -> QuerySpec:
    """Check every field against the schema and the allow-lists. Never repairs."""
    if spec.table not in schema:
        raise SpecRejected(f"unknown table {spec.table!r}")
    columns = schema[spec.table]
    for col in spec.columns:
        if col not in columns:
            raise SpecRejected(f"unknown column {col!r} on {spec.table}")
    for col, op, _value in spec.filters:
        if col not in columns:
            raise SpecRejected(f"unknown filter column {col!r}")
        if op not in {"=", "<", "<=", ">", ">=", "in", "ilike"}:
            raise SpecRejected(f"unsupported operator {op!r}")
    if spec.predicate is not None and spec.predicate not in ALLOWED_PREDICATES:
        raise SpecRejected(f"predicate {spec.predicate!r} is not permitted")
    if spec.aggregate is not None and spec.aggregate not in ALLOWED_AGGREGATES:
        raise SpecRejected(f"aggregate {spec.aggregate!r} is not permitted")
    if not 1 <= spec.limit <= max_limit:
        raise SpecRejected(f"limit {spec.limit} outside 1..{max_limit}")
    return spec

Refusing to repair is the discipline that makes this reviewable. A validator that silently clamps a limit of ten thousand to five hundred has produced a different query from the one the model proposed, and the agent will never learn that its plan was unreasonable. The allow-list approach is developed in constraining generated SQL to an allow-listed function set.

3. Assemble the query from the specification

The application owns the SQL text. The model never sees it, never writes it, and cannot influence it beyond the fields the specification exposes.

-- features_in_region: index-aware by construction.
SELECT {columns}
FROM   {table}
WHERE  geom && ST_GeomFromEWKB(:region)          -- bounding box first, uses the index
  AND  ST_Intersects(geom, ST_GeomFromEWKB(:region))
  {extra_filters}
LIMIT  :limit;

PREDICATE_SQL = {
    "intersects": "ST_Intersects(geom, ST_GeomFromEWKB(:region))",
    "within":     "ST_Within(geom, ST_GeomFromEWKB(:region))",
    "contains":   "ST_Contains(geom, ST_GeomFromEWKB(:region))",
    "dwithin":    "ST_DWithin(geom::geography, ST_GeomFromEWKB(:region)::geography, :radius_m)",
}


def build(spec: QuerySpec) -> tuple[str, dict]:
    """Return SQL text and bound parameters. Values never enter the string."""
    cols = ", ".join(spec.columns) if spec.columns else "*"
    clauses, params = ["geom && ST_GeomFromEWKB(:region)"], {"limit": spec.limit}
    if spec.predicate:
        clauses.append(PREDICATE_SQL[spec.predicate])
    for i, (col, op, value) in enumerate(spec.filters):
        key = f"f{i}"
        clauses.append(f"{col} {op} :{key}")
        params[key] = value
    sql = (f"SELECT {cols} FROM {spec.table} "
           f"WHERE {' AND '.join(clauses)} LIMIT :limit")
    return sql, params

Putting the bounding-box operator first, unconditionally, is what makes every generated query index-aware regardless of which predicate the model chose. The composition and its failure modes are covered in generating valid PostGIS queries from natural language.

4. Check the plan before running

A validated, index-aware query can still be ruinous on a large table. Asking the planner for an estimate before execution is cheap and turns an unbounded risk into a decision.

def plan_is_acceptable(conn, sql: str, params: dict,
                       max_cost: float = 500_000.0) -> tuple[bool, str]:
    """Ask the planner what this will cost. Reject rather than discover at runtime."""
    try:
        with conn.cursor() as cur:
            cur.execute(f"EXPLAIN (FORMAT JSON) {sql}", params)
            plan = cur.fetchone()[0][0]["Plan"]
    except Exception as exc:                       # a plan failure is a rejection, not a crash
        return False, f"could not plan the query: {exc}"
    cost = float(plan.get("Total Cost", 0.0))
    node = plan.get("Node Type", "")
    if "Seq Scan" in str(plan) and cost > max_cost / 10:
        return False, f"plan uses a sequential scan at cost {cost:.0f}"
    if cost > max_cost:
        return False, f"estimated cost {cost:.0f} exceeds the limit {max_cost:.0f}"
    return True, f"{node}, estimated cost {cost:.0f}"

5. Bound the execution as well as the plan

Estimates are estimates. A statement timeout is the backstop that turns a mis-estimated query into a bounded failure rather than a saturated database.

-- Per-transaction, not per-session: it must not leak to other work.
SET LOCAL statement_timeout = '5s';
SET LOCAL lock_timeout = '1s';

6. Return results the agent can reason about

Rows are not an answer. Returning the row count, the truncation flag and the query shape alongside the data is what lets an agent say “the first hundred of four thousand” instead of implying completeness.

def to_result(rows: Sequence[dict], spec: QuerySpec, truncated: bool) -> dict:
    return {
        "rows": rows,
        "count": len(rows),
        "truncated": truncated,
        "shape": spec.shape,
        "note": ("results were limited; ask a narrower question for a complete set"
                 if truncated else ""),
    }

7. Keep the schema the model sees small and current

A model given a hundred-table schema will produce specifications referencing tables that exist and are irrelevant. Expose the handful of tables and columns that answer real questions, with one-line descriptions, and regenerate that description from the real schema so it cannot drift.

def schema_for_prompt(schema: dict[str, dict], exposed: set[str]) -> str:
    """A compact description of only the tables an agent may query."""
    lines = []
    for table in sorted(exposed & set(schema)):
        cols = ", ".join(sorted(schema[table]["columns"]))
        lines.append(f"{table}({cols}) — {schema[table]['description']}")
    missing = exposed - set(schema)
    if missing:
        log.warning("exposed tables missing from the schema: %s", sorted(missing))
    return "\n".join(lines)

8. Log the specification, not the SQL

The specification is what the model produced and what a reviewer needs; the SQL is a deterministic function of it. Logging the spec makes a disputed answer reproducible and makes it possible to replay a question against a changed schema.

def audit(spec: QuerySpec, verdict: str, cost_note: str, rows: int) -> dict:
    return {"shape": spec.shape, "table": spec.table, "predicate": spec.predicate,
            "filters": [(c, o) for c, o, _ in spec.filters], "limit": spec.limit,
            "verdict": verdict, "plan": cost_note, "rows": rows}

Note that the filter values are omitted from the audit record. They frequently contain user content, and the column and operator are what a reviewer needs to understand the query’s shape.

9. Handle the aggregate shapes separately

Aggregates are where a generated query is most likely to be both expensive and misleading. A count over a region is cheap when the region is small and ruinous when it is not, and the same specification produces both depending on a parameter the model chose.

AGGREGATE_SQL = {
    "count":     "count(*)",
    "sum_area":  "sum(ST_Area(geom::geography))",
    "min":       "min({col})",
    "max":       "max({col})",
    "median":    "percentile_cont(0.5) WITHIN GROUP (ORDER BY {col})",
}


def build_aggregate(spec: QuerySpec, region_area_km2: float,
                    max_region_km2: float = 5_000.0) -> tuple[str, dict]:
    """Aggregates require a bounded region — an unbounded count is not a query."""
    if spec.aggregate is None:
        raise SpecRejected("aggregate shape without an aggregate function")
    if region_area_km2 > max_region_km2:
        raise SpecRejected(
            f"region of {region_area_km2:.0f} km² exceeds the {max_region_km2:.0f} km² "
            "limit for aggregates; narrow the area or ask for a pre-computed statistic")
    body = AGGREGATE_SQL[spec.aggregate]
    if "{col}" in body:
        if not spec.columns:
            raise SpecRejected(f"aggregate {spec.aggregate!r} needs a column")
        body = body.format(col=spec.columns[0])
    sql = (f"SELECT {body} AS value FROM {spec.table} "
           "WHERE geom && ST_GeomFromEWKB(:region) "
           "AND ST_Intersects(geom, ST_GeomFromEWKB(:region))")
    return sql, {}

The region-area limit is a policy rather than a technical constraint, and stating it in the rejection is what makes it actionable. “Narrow the area or ask for a pre-computed statistic” tells the agent exactly which two moves are available, and the second of them is usually the right one for a genuinely national question.

Aggregates also deserve their own truncation semantics, which is to say none: an aggregate is either computed over the whole region or it is refused. A count limited to five hundred rows is not a count, and returning one with a truncation flag invites exactly the misreading the flag was meant to prevent.

10. Give the model examples of accepted specifications, not of SQL

Few-shot examples shape what a model produces more than instructions do, and examples of SQL teach it to write SQL. A handful of accepted specifications — one per shape, with realistic tables and filters — is both smaller in the context and better aligned with what the layer will accept.

def example_specs() -> list[dict]:
    """Compact, schema-accurate examples: one per shape, regenerated from the schema."""
    return [
        {"shape": "features_in_region", "table": "parcels",
         "predicate": "intersects", "columns": ["parcel_ref", "use_class"], "limit": 50},
        {"shape": "nearest_to", "table": "stations",
         "predicate": "dwithin", "order_by_distance": True, "limit": 5},
        {"shape": "aggregate_in_region", "table": "buildings",
         "aggregate": "count", "predicate": "within"},
    ]

Regenerating the examples from the live schema matters for the same reason the description does. An example naming a column that no longer exists teaches the model to produce specifications that will be rejected, and the rejection rate rises for reasons nobody connects to a migration weeks earlier.

Operating This Stage Over Time

Schemas change, and a generated-query layer is unusually sensitive to it. A renamed column turns every specification referencing it into a rejection, which is the correct behaviour and presents to users as the agent suddenly being unable to answer a question it handled yesterday. Regenerating the prompt’s schema description from the live schema on every deploy is what keeps the two in step; a hand-maintained description drifts within weeks.

The allow-lists drift in the other direction. Each new capability adds a predicate or an aggregate, and after a year the list admits most of the function library — at which point it is documentation rather than a constraint. Review it periodically against what is actually used: a permitted function that has never appeared in an accepted specification is a permission with no beneficiary.

Cost limits need the same treatment for the opposite reason. A limit set against a small table will reject reasonable queries once the table is large, and the symptom is a rising rejection rate rather than an error. Track rejections by reason — unknown column, forbidden predicate, plan too expensive — because those three point at completely different fixes.

Finally, watch the truncation rate. A question that consistently returns truncated results is a question the shape vocabulary does not serve well, usually because it wanted an aggregate and got a sample. That is a signal to add a shape rather than to raise a limit.

Failure Modes & Root Causes

The full-table scan. A correct query reads everything because a function wrapped the indexed column. Root cause: transforming the column rather than the parameter. Mitigation: emit the index-aware form from the template; check the plan.

The silently narrowed answer. A limit truncates results and the answer implies completeness. Root cause: rows returned without a truncation flag. Mitigation: return the flag and the count, and let the agent say so.

The query that matches nothing. A filter references a column that was renamed, or a value in the wrong units, and the result is an empty set that reads as “there is nothing there”. Root cause: validation against a stale schema description. Mitigation: validate against the live schema; regenerate the prompt description from it.

The expensive aggregate. A count over a national table with no region filter runs for minutes. Root cause: a shape that permits an unbounded aggregate. Mitigation: require a region on every shape, and reject a specification without one.

Production Validation Protocols

1. No-interpolation assertion. Assert every generated query is executed with bound parameters and that the SQL text contains no value from the specification.
2. Index-plan test. For each shape, assert the plan uses a spatial index against a representative table; a plan regression is silent and costly.
3. Allow-list enforcement test. Assert a specification naming a forbidden predicate is rejected, and that the rejection names the predicate.
4. Truncation reporting test. Assert that a result at the limit carries the truncation flag and the note.
5. Schema-drift check. Assert every exposed table and column in the prompt description exists in the live schema, on every deploy.
6. Rejection-reason breakdown. Publish rejections by reason and alert on a step change in any one; the three reasons have three different fixes.

Frequently Asked Questions

Related

• Up to the section overview: Geospatial Prompt Engineering and Tool Routing
• Technique: Generating Valid PostGIS Queries from Natural Language
• Technique: Preventing SQL Injection in LLM-Generated Spatial Queries
• Technique: Constraining Generated SQL to an Allow-Listed Function Set
• Peer topic: Spatial Function-Calling Schemas
• Peer topic: GeoPandas and PostGIS Tool Routing