Skip to content

5. Text-to-SQL

Intermediate · 14 min read

"How much revenue did Pune customers bring in last month?" — text-to-SQL lets people ask that in plain language: an LLM writes the query, your code runs it, and the result (or an LLM summary of it) comes back. It is one of the most requested enterprise GenAI features. It is also a feature where an LLM writes code that runs against your database, so safety is designed in from the start.

flowchart LR
    Q[Question] --> P[Prompt: schema + rules + examples]
    P --> L[LLM writes SQL]
    L --> V{Validate}
    V -->|rejected| L
    V -->|ok| R[(Read-only DB<br/>with timeout + LIMIT)]
    R -->|error| L
    R --> A[Rows → table or LLM summary]

5.1 Sample database (as a file, so it can be opened read-only)

import re
import sqlite3

setup = sqlite3.connect("shop.db")
setup.executescript("""
DROP TABLE IF EXISTS customers; DROP TABLE IF EXISTS orders;
CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, city TEXT, plan TEXT, email TEXT);
CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER REFERENCES customers(id),
                     amount REAL, status TEXT, created_at TEXT);
INSERT INTO customers VALUES
    (1,'Priya','Bengaluru','pro','priya@example.com'), (2,'Rahul','Pune','free','rahul@example.com'),
    (3,'Ananya','Pune','pro','ananya@example.com'),    (4,'Arjun','Delhi','free','arjun@example.com');
INSERT INTO orders VALUES
    (101,1,2499,'delivered','2026-09-02'), (102,2,799,'delivered','2026-09-05'),
    (103,1,1299,'shipped','2026-09-20'),   (104,3,5600,'delivered','2026-09-21'),
    (105,4,450,'cancelled','2026-09-22'),  (106,3,999,'shipped','2026-10-01');
""")
setup.commit()
setup.close()

5.2 Describe the schema to the model

The LLM can only write correct SQL for tables and columns it knows about. Generate the description from the database itself, and add what the column names don't say — allowed values, units, date formats:

NOTES = {
    "orders.amount": "order value in INR",
    "orders.status": "one of 'delivered', 'shipped', 'cancelled'",
    "orders.created_at": "ISO date text 'YYYY-MM-DD'",
    "customers.plan": "'free' or 'pro'",
}
HIDDEN_COLUMNS = {"customers.email"}            # never show personal data columns to the model

def describe_schema(con) -> str:
    lines = []
    for (table,) in con.execute("SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name"):
        cols = [(name, ctype) for _, name, ctype, *_ in con.execute(f"PRAGMA table_info({table})")
                if f"{table}.{name}" not in HIDDEN_COLUMNS]
        body = []
        for i, (name, ctype) in enumerate(cols):
            comma = "," if i < len(cols) - 1 else ""           # comma BEFORE the comment, not inside it
            note = NOTES.get(f"{table}.{name}")
            body.append(f"  {name} {ctype}{comma}" + (f"  -- {note}" if note else ""))
        lines.append(f"TABLE {table} (\n" + "\n".join(body) + "\n)")
    return "\n".join(lines)

ro = sqlite3.connect("file:shop.db?mode=ro", uri=True)       # READ-ONLY connection
SCHEMA = describe_schema(ro)
print(SCHEMA)
Output
TABLE customers (
  id INTEGER,
  name TEXT,
  city TEXT,
  plan TEXT  -- 'free' or 'pro'
)
TABLE orders (
  id INTEGER,
  customer_id INTEGER,
  amount REAL,  -- order value in INR
  status TEXT,  -- one of 'delivered', 'shipped', 'cancelled'
  created_at TEXT  -- ISO date text 'YYYY-MM-DD'
)

For large databases (hundreds of tables), don't send everything: retrieve the relevant tables for each question (embed table descriptions and search them), or let an agent explore — see Text-to-SQL agents.

5.3 The prompt

FEW_SHOT = [
    ("How many pro customers are there?", "SELECT COUNT(*) AS pro_customers FROM customers WHERE plan = 'pro';"),
    ("Revenue per city from delivered orders",
     "SELECT c.city, SUM(o.amount) AS revenue FROM orders o JOIN customers c ON c.id = o.customer_id "
     "WHERE o.status = 'delivered' GROUP BY c.city ORDER BY revenue DESC;"),
]

def build_prompt(question: str, today: str = "2026-10-07") -> list[dict]:
    system = f"""You translate questions into a single SQLite SELECT query.
Rules:
- Use only the tables and columns in the schema. If the question can't be answered from them, reply exactly: CANNOT_ANSWER
- Read-only: a single SELECT (WITH ... SELECT is fine). Never modify data.
- Exclude cancelled orders from revenue unless asked.
- Today's date is {today}.
- Reply with the SQL only, no explanation, no code fences.

Schema:
{SCHEMA}"""
    msgs = [{"role": "system", "content": system}]
    for q, sql in FEW_SHOT:
        msgs += [{"role": "user", "content": q}, {"role": "assistant", "content": sql}]
    return msgs + [{"role": "user", "content": question}]

print(len(build_prompt("Which city has the most orders?")), "messages")
Output
6 messages

Few-shot examples teach your conventions — the join path between tables, how "revenue" is defined, how to alias columns — which matter more than SQL syntax.

5.4 Validate before you run

Never execute generated SQL directly. Check it first:

ALLOWED_TABLES = {"customers", "orders"}
FORBIDDEN = re.compile(r"\b(insert|update|delete|drop|alter|create|replace|attach|pragma|vacuum)\b", re.I)

def validate_sql(sql: str, max_rows: int = 200) -> str:
    sql = sql.strip().removeprefix("```sql").removesuffix("```").strip().rstrip(";")
    if sql.upper() == "CANNOT_ANSWER":
        raise ValueError("The model says the question can't be answered from this database.")
    if ";" in sql:
        raise ValueError("Only one statement is allowed.")
    if not re.match(r"^\s*(select|with)\b", sql, re.I):
        raise ValueError("Only SELECT queries are allowed.")
    if FORBIDDEN.search(sql):
        raise ValueError("Query contains a forbidden keyword.")
    tables = {t.lower() for t in re.findall(r"\b(?:from|join)\s+([a-zA-Z_]\w*)", sql, re.I)}
    cte_names = {n.lower() for n in re.findall(r"\b(\w+)\s+as\s*\(", sql, re.I)}
    unknown = tables - ALLOWED_TABLES - cte_names
    if unknown:
        raise ValueError(f"Unknown or forbidden table(s): {', '.join(sorted(unknown))}")
    if "email" in sql.lower():
        raise ValueError("Query touches a restricted column.")
    return f"SELECT * FROM ({sql}) LIMIT {max_rows}"           # cap the result size

for candidate in ["SELECT city, COUNT(*) FROM customers GROUP BY city;",
                  "DELETE FROM orders",
                  "SELECT * FROM orders; DROP TABLE orders",
                  "SELECT name, email FROM customers",
                  "SELECT * FROM salaries"]:
    try:
        print("OK     ", validate_sql(candidate))
    except ValueError as e:
        print("REJECT ", e)
Output
OK      SELECT * FROM (SELECT city, COUNT(*) FROM customers GROUP BY city) LIMIT 200
REJECT  Only SELECT queries are allowed.
REJECT  Only one statement is allowed.
REJECT  Query touches a restricted column.
REJECT  Unknown or forbidden table(s): salaries

Regex checks are a first filter, not a security boundary. The real protections are in the database:

  • A read-only connection or database user (mode=ro above; in PostgreSQL a role with only SELECT grants).
  • Access only to allowed tables or views — create views that hide sensitive columns and filter to the current tenant/user, and grant the role access to those views only.
  • A statement timeout (PostgreSQL SET statement_timeout = '5s') and a row LIMIT.
  • For production-grade parsing, a SQL parser such as sqlglot instead of regex.

5.5 Run safely, with a timeout

import time

def run_readonly(sql: str, timeout_s: float = 2.0):
    deadline = time.monotonic() + timeout_s
    ro.set_progress_handler(lambda: 1 if time.monotonic() > deadline else 0, 10_000)   # abort long queries
    try:
        cur = ro.execute(sql)
        return [d[0] for d in cur.description], cur.fetchall()
    finally:
        ro.set_progress_handler(None, 0)

print(run_readonly(validate_sql("SELECT city, COUNT(*) AS n FROM customers GROUP BY city ORDER BY n DESC")))
try:
    ro.execute("DELETE FROM orders")                          # even if validation were bypassed…
except sqlite3.OperationalError as e:
    print("database refused:", e)
Output
(['city', 'n'], [('Pune', 2), ('Delhi', 1), ('Bengaluru', 1)])
database refused: attempt to write a readonly database

5.6 The full loop, with self-correction

When the query fails (bad column, syntax error), send the error back to the model and let it fix the SQL. A scripted fake LLM shows the flow:

class ScriptedLLM:
    def __init__(self, replies): self.replies = iter(replies)
    def __call__(self, messages): return next(self.replies)

def ask(question: str, llm, max_attempts: int = 3):
    messages = build_prompt(question)
    for attempt in range(1, max_attempts + 1):
        sql = llm(messages)
        try:
            cols, rows = run_readonly(validate_sql(sql))
            return sql, cols, rows
        except (ValueError, sqlite3.Error) as e:
            print(f"attempt {attempt}: {e}")
            messages += [{"role": "assistant", "content": sql},
                         {"role": "user", "content": f"That query failed: {e}. Fix it and reply with SQL only."}]
    raise RuntimeError("Could not produce a working query.")

llm = ScriptedLLM([
    "SELECT c.city, SUM(o.total) AS revenue FROM orders o JOIN customers c ON c.id = o.customer_id GROUP BY c.city",
    "SELECT c.city, SUM(o.amount) AS revenue FROM orders o JOIN customers c ON c.id = o.customer_id "
    "WHERE o.status != 'cancelled' GROUP BY c.city ORDER BY revenue DESC",
])
sql, cols, rows = ask("Revenue by city?", llm)
print(cols, rows)
Output
attempt 1: no such column: o.total
['city', 'revenue'] [('Pune', 7398.0), ('Bengaluru', 3798.0)]

Then either show the table directly, or ask the LLM to summarise the rows in a sentence — passing the rows as data, and showing the SQL to the user (in a "show query" expander) so they can check it.

5.7 What goes wrong — and how to catch it

A query that runs can still be wrong. The most common silent errors:

Error Example Mitigation
Wrong business definition revenue includes cancelled orders define metrics in the prompt or a semantic layer; few-shot examples
Wrong join / double counting joining orders and tickets inflates totals document join paths; examples; review the SQL
Wrong filter values city = 'Bangalore' when data says 'Bengaluru' give allowed values (or a sample of distinct values) in the schema
Date logic "last month" computed from the model's training date put today's date in the prompt
Ambiguous question "top customers" — by revenue? orders? ask a clarifying question, or state the assumption with the answer

5.8 Evaluate with execution accuracy

Build a set of questions with gold (correct) SQL. Run both the generated and gold queries and compare the results, not the SQL text — different SQL can be equally correct:

EVAL = [
    ("How many customers are on the pro plan?", "SELECT COUNT(*) FROM customers WHERE plan = 'pro'"),
    ("Total revenue from delivered orders", "SELECT SUM(amount) FROM orders WHERE status = 'delivered'"),
    ("Which customers are from Pune?", "SELECT name FROM customers WHERE city = 'Pune'"),
]
generated = [   # what the model produced for each question
    "SELECT COUNT(id) AS n FROM customers WHERE plan = 'pro'",
    "SELECT SUM(amount) FROM orders",                                   # forgot the status filter
    "SELECT name FROM customers WHERE city = 'Pune' ORDER BY name",
]

def same_result(sql_a: str, sql_b: str) -> bool:
    a = ro.execute(sql_a).fetchall()
    b = ro.execute(sql_b).fetchall()
    return sorted(map(tuple, a)) == sorted(map(tuple, b))              # ignore row order and column names

results = [same_result(g, gold) for g, (_, gold) in zip(generated, EVAL)]
for (q, _), ok in zip(EVAL, results):
    print(f"{'PASS' if ok else 'FAIL'}  {q}")
print(f"execution accuracy: {sum(results)}/{len(results)}")
Output
PASS  How many customers are on the pro plan?
FAIL  Total revenue from delivered orders
PASS  Which customers are from Pune?
execution accuracy: 2/3

The failed case is the classic one: valid SQL, plausible number, wrong answer. Build the eval set from real user questions, and re-run it whenever you change the prompt, the model or the schema.

Interview questions

How would you build a safe text-to-SQL feature?

Give the model the relevant schema with column descriptions, allowed values and business definitions, plus few-shot examples. Validate the output (single SELECT, allow-listed tables, no sensitive columns), and run it with a read-only role on views scoped to the user/tenant, with a timeout and row limit. Feed errors back for self-correction, show the SQL to the user, and measure execution accuracy on an eval set.

Why is text-to-SQL harder than it looks?

The SQL can be valid and still wrong: business definitions (what counts as revenue), join paths, filter values, date logic and ambiguous questions. Large schemas don't fit in the prompt. And the model is generating code that runs on real data, so security has to be enforced by the database, not by the prompt.

How do you evaluate text-to-SQL?

Execution accuracy: run generated and gold SQL and compare result sets (order-insensitive). Track it on a set of real questions with gold queries, by category, and add every production failure as a new case.

Practice

  • Add a third few-shot example that defines "average order value" and test a question that needs it.
  • Replace the regex table check with sqlglot.parse_one(sql).find_all(sqlglot.exp.Table).

Next: Text-to-SQL agents — let the model explore the database itself.