#!/usr/bin/env python3 """Score an NL2SQL tool against the messy schema, by execution result. Reproducible companion to https://infinisynapse.com/use-cases/nl2sql How this scores: not by string-comparing SQL (many correct queries look different) but by running both the tool's SQL and a reference query and comparing the result sets. This is the same "execution accuracy" metric the Spider and BIRD benchmarks use. Workflow: 1. python make_messy_schema.py 2. python score_nl2sql.py --init-answers mytool.sql (writes a template with the 10 questions as comments) 3. Ask each question to the tool EXACTLY as written. Paste the SQL it produces under the matching -- Q01 marker. Do not fix its SQL. 4. python score_nl2sql.py --answers mytool.sql Scoring, per question: 1.0 result set matches the reference exactly 0.0 wrong result, SQL error, or no answer given There is no partial credit, because a revenue number that is 4% wrong is not 40% right - it is wrong, and it will be wrong in a board deck. If you want a softer signal, read the per-question diff the script prints. Standard library only. Released under CC BY 4.0. """ from __future__ import annotations import argparse import csv import os import re import sqlite3 import sys # Each entry: id, question as asked, the trap it targets, reference SQL. QUESTIONS = [ dict( id="Q01", q="How many active customers do we have?", trap="Soft deletes (is_deleted) plus a deprecated customers table that still holds rows.", gold="SELECT COUNT(*) FROM customers_v2 WHERE is_deleted = 0", ), dict( id="Q02", q="What was total revenue in Q1 2025?", trap="Three date columns (created_at / booked_at / closed_at); revenue recognises on " "booked_at. Voided and non-complete orders must be excluded.", gold="""SELECT ROUND(SUM(total_amount), 2) FROM orders WHERE void_flag = 0 AND LOWER(status) IN ('complete','completed') AND booked_at >= '2025-01-01' AND booked_at < '2025-04-01'""", ), dict( id="Q03", q="How many customers placed an order in 2025?", trap="Grain: counting orders instead of distinct customers. A few heavy accounts make " "the two numbers very different.", gold="""SELECT COUNT(DISTINCT o.customer_id) FROM orders o JOIN customers_v2 c ON c.customer_id = o.customer_id WHERE c.is_deleted = 0 AND o.void_flag = 0 AND o.booked_at >= '2025-01-01' AND o.booked_at < '2026-01-01'""", ), dict( id="Q04", q="What is the average order value for completed orders?", trap="Status enum drift: 'complete', 'completed' and 'COMPLETE' all occur and all mean " "the same thing. Matching only one undercounts.", gold="""SELECT ROUND(AVG(total_amount), 2) FROM orders WHERE void_flag = 0 AND LOWER(status) IN ('complete','completed')""", ), dict( id="Q05", q="What is total revenue by plan tier?", trap="Ambiguous join keys: orders has customer_id, user_id and external_ref. Only " "customer_id links to the account; external_ref is not unique.", gold="""SELECT c.plan_tier, ROUND(SUM(o.total_amount), 2) AS revenue FROM orders o JOIN customers_v2 c ON c.customer_id = o.customer_id WHERE o.void_flag = 0 AND LOWER(o.status) IN ('complete','completed') AND c.is_deleted = 0 GROUP BY c.plan_tier ORDER BY c.plan_tier""", ), dict( id="Q06", q="What is the total value of all order items?", trap="Unit mismatch: order_items.unit_price_cents is in CENTS while orders.total_amount " "is in DOLLARS. Summing without dividing by 100 is off by 100x.", gold="SELECT ROUND(SUM(qty * unit_price_cents) / 100.0, 2) FROM order_items", ), dict( id="Q07", q="Which acquisition channel has the highest revenue per customer?", trap="Requires the correct join, soft-delete filter, and dividing by DISTINCT customers " "rather than by order count.", gold="""SELECT c.acquisition_channel, ROUND(SUM(o.total_amount) / COUNT(DISTINCT c.customer_id), 2) AS rev_per_cust FROM orders o JOIN customers_v2 c ON c.customer_id = o.customer_id WHERE o.void_flag = 0 AND LOWER(o.status) IN ('complete','completed') AND c.is_deleted = 0 GROUP BY c.acquisition_channel ORDER BY rev_per_cust DESC LIMIT 1""", ), dict( id="Q08", q="How many orders have not been closed yet?", trap="closed_at is NULL for unclosed orders, but also NULL for many older rows. The " "honest reading is 'closed_at IS NULL', excluding voided orders.", gold="SELECT COUNT(*) FROM orders WHERE closed_at IS NULL AND void_flag = 0", ), dict( id="Q09", q="What is the actual lifetime revenue of customer 1, based on their orders?", trap="customers_v2.ltv_estimate is a stale snapshot column. The question asks for the " "value derived from orders, not the stored estimate.", gold="""SELECT ROUND(COALESCE(SUM(total_amount), 0), 2) FROM orders WHERE customer_id = 1 AND void_flag = 0 AND LOWER(status) IN ('complete','completed')""", ), dict( id="Q10", q="How many enterprise customers signed up in 2025?", trap="Soft deletes plus the deprecated customers table, which has no plan_tier at all " "and will silently produce a different answer if joined.", gold="""SELECT COUNT(*) FROM customers_v2 WHERE plan_tier = 'enterprise' AND is_deleted = 0 AND signup_date >= '2025-01-01' AND signup_date < '2026-01-01'""", ), ] def run(con: sqlite3.Connection, sql: str): cur = con.execute(sql) return cur.fetchall() def normalise(rows): """Round floats so 1.0 and 1.0000001 compare equal; order-insensitive.""" out = [] for r in rows: out.append(tuple(round(v, 2) if isinstance(v, float) else v for v in r)) return sorted(out, key=lambda t: [(x is None, str(x)) for x in t]) def init_answers(path: str) -> None: if os.path.exists(path): sys.exit(f"refusing to overwrite existing {path}") with open(path, "w", encoding="utf-8") as f: f.write("-- NL2SQL evaluation answers.\n") f.write("-- Ask each question EXACTLY as written, paste the tool's SQL below the marker.\n") f.write("-- Do not correct the SQL by hand - that is what you are measuring.\n\n") for item in QUESTIONS: f.write(f"-- {item['id']}: {item['q']}\n") f.write(f"-- watch for: {' '.join(item['trap'].split())}\n\n\n") print(f"wrote template {path} - fill it in, then re-run with --answers {path}") def parse_answers(path: str) -> dict[str, str]: text = open(path, encoding="utf-8").read() blocks: dict[str, str] = {} parts = re.split(r"^--\s*(Q\d\d)\s*:", text, flags=re.M) for i in range(1, len(parts), 2): qid, chunk = parts[i], parts[i + 1] # The rest of the marker line is the question text, not SQL. chunk = chunk.split("\n", 1)[1] if "\n" in chunk else "" sql = "\n".join(l for l in chunk.splitlines() if not l.strip().startswith("--")) blocks[qid] = sql.strip().rstrip(";") return blocks def main() -> None: ap = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--db", default="nl2sql-eval.db", help="database (default: %(default)s)") ap.add_argument("--answers", help="your filled-in .sql file") ap.add_argument("--init-answers", metavar="PATH", help="write a blank answers template") ap.add_argument("--show-reference", action="store_true", help="print reference SQL and its result for every question") ap.add_argument("--csv", metavar="PATH", help="also write per-question results to CSV") args = ap.parse_args() if args.init_answers: init_answers(args.init_answers) return if not os.path.exists(args.db): sys.exit(f"{args.db} not found - run make_messy_schema.py first") con = sqlite3.connect(args.db) if args.show_reference: for item in QUESTIONS: print(f"\n{item['id']} {item['q']}") print(f" trap: {' '.join(item['trap'].split())}") print(" sql :", " ".join(item["gold"].split())) print(" -> ", run(con, item["gold"])) return if not args.answers: sys.exit("pass --answers FILE, or --init-answers FILE to create a template, " "or --show-reference to inspect the reference queries") given = parse_answers(args.answers) total = 0.0 rows_out = [] print(f"scoring {args.answers} against {args.db}\n") for item in QUESTIONS: qid = item["id"] gold = normalise(run(con, item["gold"])) cand_sql = given.get(qid, "") if not cand_sql: verdict, detail = "SKIP", "no SQL provided" else: try: got = normalise(run(con, cand_sql)) if got == gold: verdict, detail = "PASS", "" total += 1 else: verdict = "FAIL" detail = f"got {got[:3]}{'...' if len(got) > 3 else ''} want {gold[:3]}{'...' if len(gold) > 3 else ''}" except Exception as e: verdict, detail = "ERROR", str(e).split("\n")[0] print(f" {verdict:<5} {qid} {item['q']}") if detail: print(f" {detail}") rows_out.append([qid, item["q"], verdict, detail]) print(f"\nexecution accuracy: {total:.0f}/{len(QUESTIONS)} ({total/len(QUESTIONS)*100:.0f}%)") print("Record the tool name, version/build string, and today's date next to this number.") if args.csv: with open(args.csv, "w", newline="", encoding="utf-8") as f: w = csv.writer(f) w.writerow(["question_id", "question", "verdict", "detail"]) w.writerows(rows_out) w.writerow([]) w.writerow(["execution_accuracy", f"{total:.0f}/{len(QUESTIONS)}"]) print(f"wrote {args.csv}") if __name__ == "__main__": main()