Skip to content

Evaluate SQL with Pydantic Evals

Already use Pydantic Evals? SqlEquivalence adds execution-based SQL scoring to its datasets and reports. It runs generated SQL on your data platform and checks the result against a gold query or expected rows, so equivalent queries pass even when their SQL text differs.

Prerequisites

uv add "evaldata[pydantic-evals]"

Write the eval

A Pydantic Evals task receives each case's inputs; its return value becomes ctx.output. SqlEquivalence expects that output to be a non-empty SQL string. The case's expected_output can be a gold SQL string or an evaldata GoldQuery, UntypedResultSet, or TypedResultSet.

Create test_pydantic_evals.py:

"""Pydantic Evals integration with execution-based SQL scoring on DuckDB."""

from collections.abc import Iterator

import pytest

from evaldata.platforms import duckdb_platform, resolve
from evaldata.pydantic_evals import SqlEquivalence, close_all
from evaldata.types import ExecutionSuccess
from pydantic_evals import Case, Dataset

_PLATFORM = duckdb_platform(name="examples-pydantic-evals")
_GENERATED_SQL = {
    "What is the total order amount by region?": (
        "WITH totals AS ("
        "SELECT region, SUM(amount) AS total FROM orders GROUP BY region"
        ") SELECT region, total FROM totals"
    ),
    "What is the average order amount by region?": (
        "SELECT region, COUNT(amount) AS average_amount FROM orders GROUP BY region"
    ),
}


@pytest.fixture(scope="module", autouse=True)
def _database() -> Iterator[None]:
    result = resolve(_PLATFORM).execute(
        "CREATE TABLE orders (id INTEGER, region VARCHAR, amount INTEGER); "
        "INSERT INTO orders VALUES (1, 'east', 50), (2, 'west', 120), (3, 'east', 20)"
    )
    assert isinstance(result, ExecutionSuccess)
    try:
        yield
    finally:
        close_all()


def generate_sql(question: str) -> str:
    """Return the fixed generated SQL for a question."""
    return _GENERATED_SQL[question]


def test_sql_equivalence() -> None:
    """Score correct and incorrect generated SQL through Pydantic Evals."""
    dataset = Dataset(
        name="text-to-sql",
        cases=[
            Case(
                name="totals-by-region",
                inputs="What is the total order amount by region?",
                expected_output="SELECT region, SUM(amount) AS total FROM orders GROUP BY region",
            ),
            Case(
                name="average-by-region",
                inputs="What is the average order amount by region?",
                expected_output=("SELECT region, AVG(amount) AS average_amount FROM orders GROUP BY region"),
            ),
        ],
        evaluators=[SqlEquivalence(platform=_PLATFORM)],
    )

    report = dataset.evaluate_sync(generate_sql)
    verdicts = {case.name: case.assertions["SqlEquivalence"].value for case in report.cases}

    assert verdicts == {"totals-by-region": True, "average-by-region": False}

The example uses fixed task responses so it runs without a model or network call. The first response rewrites the gold query with a CTE but returns the same rows, so it passes. The second uses COUNT where the question requires AVG, so it fails.

Run it

uv run pytest test_pydantic_evals.py -q

Run it from a clone

This is the bundled examples/11_pydantic_evals/ example. If you've cloned the repository, run it with uv run --extra pydantic-evals pytest examples/11_pydantic_evals -q.

How scoring works

For a gold query, SqlEquivalence first checks whether the generated and gold SQL normalize to the same structure. If that check cannot confirm equivalence, it runs the gold query and compares the two result sets. An expected result set is compared directly with the generated query's result.

Different results fail with a reason summarizing any row-count, column, type, or value mismatches. SQL execution errors also produce a failed evaluation rather than escaping from the evaluator. Invalid case contracts, such as an empty task output or unsupported expected_output, raise ValueError.

Read the report

evaluate_sync returns a Pydantic Evals EvaluationReport. Each case stores the evaluator's boolean verdict and explanation under assertions:

for case in report.cases:
    result = case.assertions["SqlEquivalence"]
    print(case.name, result.value, result.reason)

Run cases concurrently

Pass max_concurrency to let Pydantic Evals score cases in parallel:

report = dataset.evaluate_sync(generate_sql, max_concurrency=8)

Each case checks out a session from the platform's connection pool. The pool caps the number of simultaneous sessions for one platform name; additional cases wait for a session to become available.

Engine Default pool size
DuckDB 8
Postgres, Snowflake, BigQuery, Databricks 4
SQLite 1

Pooled sessions are reused without resetting session-local state. Keep evaluated SQL side-effect-free: temporary tables, session parameters, roles, and schema changes may be visible to a later case that receives the same session. Call close_all() when the dataset is finished; the example's fixture does this during teardown.

Next steps