Skip to content

Evaluate text-to-SQL with dbt

Run text-to-SQL evals against a dbt project. evaldata reads the compiled artifacts (manifest.json and optional catalog.json), uses the warehouse connection from the project's dbt profile, and checks each answer against a gold query you write.

Prerequisites

uv add "evaldata[dbt,litellm]"

The litellm extra backs the solver when you run dbt-bench with a model id.

You also need a built dbt project. Compile it and generate its catalog so target/ holds both artifacts:

dbt build          # or: dbt compile
dbt docs generate  # writes catalog.json with resolved column types

catalog.json is optional. It gives evaldata the column types resolved by the warehouse. Without it, evaldata uses the types declared in your model YAML.

Write the cases file

A cases file pairs each question with the SQL whose result is the correct answer:

# cases.yml
- question: How many customers placed an order in 2024?
  gold_sql: |
    select count(distinct customer_id) as n
    from customers
    where first_order >= '2024-01-01'
  select: [customers]   # optional: limit the schema shown to the model

- question: What is total revenue by month?
  gold_sql: select date_trunc('month', ordered_at) as month, sum(amount) as revenue from orders group by 1

Each entry needs a question and a gold_sql. select limits the schema to named tables, and id names the case; both are optional.

Run it

evaldata dbt-bench path/to/dbt_project --model openai/gpt-4o-mini --cases cases.yml

evaldata reads the warehouse connection from the project's dbt profile, sends the project's schema to the model, runs the model's SQL for each question, and compares the result to the gold query. It reports execution accuracy: the fraction of questions whose result matches.

EX (dbt): 72.0% (18/25)

--model is any litellm model id. Other options:

  • --mode model: skip the cases file; instead take every documented model, asking its description as the question and using its compiled SQL as the gold answer.
  • --mode tests: check each documented model's result against its not_null and unique data tests instead of a gold query.
  • --target-dir DIR: where the artifacts live, if not <project>/target.
  • --profiles-dir DIR / --target NAME: find and select the dbt profile target.
  • --limit N: run only the first N questions.
  • --json PATH: also write the scores and every result to a JSON file.

Check the connection

See whether evaldata can reach the project's warehouse:

evaldata doctor --dbt-project path/to/dbt_project

Run it in pytest

Run dbt evals as pytest tests with your own prompt, fine-tune, agent, or scorer. Load the cases yourself:

import pytest

from evaldata import ExecutionAccuracy, assert_eval
from evaldata.dbt import DbtError, load_dbt, platform_from_profile
from evaldata.solvers import SCHEMA_PROMPT_TEMPLATE, PromptSolver

platform = platform_from_profile("path/to/dbt_project")
if isinstance(platform, DbtError):
    raise RuntimeError(platform.message)

cases = load_dbt("path/to/dbt_project/target", platform=platform, cases="cases.yml")
if isinstance(cases, DbtError):
    raise RuntimeError(cases.message)


@pytest.mark.parametrize("case", cases, ids=lambda case: case.id)
def test_dbt_question(case):
    solver = PromptSolver("openai/gpt-4o-mini", prompt_template=SCHEMA_PROMPT_TEMPLATE)
    assert_eval(case, solver, scorers=[ExecutionAccuracy(row_order="ignore", multiplicity="set")])

load_dbt and platform_from_profile return a DbtError when the project can't be read. The cases are ordinary EvalCase objects, so any scorer works.

Bundled example

The bundled example includes a small jaffle-shop dbt project, a DuckDB database, and fixed model responses.

"""Text-to-SQL evals against a dbt project (jaffle shop), with the model stubbed.

`platform_from_profile` reads the warehouse connection from the project's dbt profile and
`load_dbt` builds one `EvalCase` per question from a cases file. Each case's gold answer is a SQL
query; `ExecutionAccuracy` runs the candidate and the gold and compares their result rows, so a
query written differently from the gold that returns the same rows passes. The project ships a
small DuckDB warehouse, so the file runs offline with no model or network — `StubLlm` supplies the
AI SQL in place of a live model.
"""

import shutil
from collections.abc import Iterator
from pathlib import Path

import pytest

from evaldata import ExecutionAccuracy, PromptSolver, assert_eval
from evaldata.dbt import DbtError, load_dbt, platform_from_profile
from evaldata.llm import StubLlm
from evaldata.platforms.registry import close_all
from evaldata.solvers import SCHEMA_PROMPT_TEMPLATE
from evaldata.types import EvalCase

_PROJECT = Path(__file__).parent / "jaffle"
_CASES = Path(__file__).parent / "cases.yml"
_SCORER = ExecutionAccuracy(row_order="ignore", multiplicity="set")


@pytest.fixture(autouse=True)
def _close_connections() -> Iterator[None]:
    """Close warehouse connections after each test so a copied DuckDB is never reused across cases."""
    yield
    close_all()


def _cases(tmp_path: Path) -> dict[str, EvalCase]:
    """Copy the project to a temp dir and load its cases, keyed by id.

    Args:
        tmp_path: A temp directory the project is copied into so its DuckDB is never mutated.

    Returns:
        The loaded eval cases, keyed by case id.
    """
    project = tmp_path / "jaffle"
    shutil.copytree(_PROJECT, project)
    platform = platform_from_profile(project)
    assert not isinstance(platform, DbtError)
    cases = load_dbt(project / "artifacts", platform=platform, cases=_CASES)
    assert not isinstance(cases, DbtError)
    return {case.id: case for case in cases}


def test_reworded_sql_passes(tmp_path: Path) -> None:
    """A query written differently from the gold but returning the same rows passes: results are compared, not SQL text."""
    case = _cases(tmp_path)["customers-count"]
    solver = PromptSolver(
        model=StubLlm("SELECT COUNT(customer_id) AS total FROM stg_customers"),
        prompt_template=SCHEMA_PROMPT_TEMPLATE,
    )
    assert_eval(case, solver, scorers=[_SCORER])


def test_wrong_sql_fails(tmp_path: Path) -> None:
    """A query that returns the wrong rows fails the eval."""
    case = _cases(tmp_path)["customers-count"]
    solver = PromptSolver(model=StubLlm("SELECT 1 AS n"), prompt_template=SCHEMA_PROMPT_TEMPLATE)
    with pytest.raises(AssertionError):
        assert_eval(case, solver, scorers=[_SCORER])


def test_revenue_by_region_passes(tmp_path: Path) -> None:
    """Total revenue by customer region — a join across orders and customers, scored on its result rows."""
    case = _cases(tmp_path)["revenue-by-region"]
    sql = (
        "SELECT c.region, SUM(o.amount) AS revenue "
        "FROM stg_orders o JOIN stg_customers c ON o.customer_id = c.customer_id "
        "GROUP BY c.region"
    )
    solver = PromptSolver(model=StubLlm(sql), prompt_template=SCHEMA_PROMPT_TEMPLATE)
    assert_eval(case, solver, scorers=[_SCORER])

How it works

  • The warehouse comes from the project's dbt profile. duckdb and postgres targets are supported; a duckdb path is resolved relative to the project.
  • The schema given to the model is the project's sources and models as CREATE TABLE statements, with column types from catalog.json and descriptions from your model YAML.
  • ExecutionAccuracy compares results as a set, ignoring row order and duplicate rows: a question passes when the model's SQL and the gold query return the same rows.

Next steps