Skip to content

Evaluate text-to-SQL on Databricks

This guide shows how to connect, authenticate, and run SQL evals in a Databricks SQL warehouse.

Prerequisites

uv add "evaldata[databricks]"

What this guide covers

  • Type resolution. The typed case checks amount: DECIMAL(10, 2). evaldata gets the precision and scale from the warehouse; the driver reports only DECIMAL.
  • Warehouse checks. The ExpectationSuite (row_count / not_null / unique) and result-set equivalence run as SQL in Databricks.
  • Authentication. The platform stores the workspace host and HTTP path. The Databricks SDK authenticates, for example through databricks auth login.

The fixture creates a session-scoped TEMPORARY VIEW. It needs only query permissions and leaves no catalog objects.

Write the eval

Create test_databricks.py:

"""Databricks example evals: fixed SQL run against a live Databricks SQL Warehouse.

Precise column types are resolved from the warehouse, and expectation and equivalence checks
are pushed down into SQL.

Point it at a warehouse with `DATABRICKS_SERVER_HOSTNAME` and `DATABRICKS_HTTP_PATH`.
Credentials are not part of the platform reference — authenticate with the Databricks CLI
(https://github.com/databricks/cli).
"""

import os
from decimal import Decimal

import pytest

from evaldata import (
    CallableSolver,
    EvalCase,
    ExpectationSuiteScorer,
    ResultSetEquivalence,
    assert_eval,
    eval_case,
)
from evaldata.platforms import databricks_platform

pytestmark = [pytest.mark.e2e, pytest.mark.cloud, pytest.mark.databricks]

_ORDERS = """(
    SELECT CAST(id AS INT) AS id, CAST(customer AS STRING) AS customer,
           CAST(amount AS DECIMAL(10, 2)) AS amount
    FROM VALUES (1, 'Ada', 10.00), (2, 'Bo', 5.50), (3, 'Cy', 20.00)
    AS orders(id, customer, amount)
) AS evaldata_ex04_orders"""
_PLATFORM = databricks_platform(
    name="examples-databricks",
    server_hostname=os.environ.get("DATABRICKS_SERVER_HOSTNAME", "your-workspace.cloud.databricks.com"),
    http_path=os.environ.get("DATABRICKS_HTTP_PATH", "/sql/1.0/warehouses/your-warehouse-id"),
)


# Precise column types. The `DECIMAL(10, 2)` assertion holds only because evaldata resolves
# the column types from the warehouse — the driver's own description reports a scaleless `DECIMAL`.
@eval_case(
    input="List each order's customer and amount, ordered by id.",
    expected={
        "rows": [
            {"customer": "Ada", "amount": Decimal("10.00")},
            {"customer": "Bo", "amount": Decimal("5.50")},
            {"customer": "Cy", "amount": Decimal("20.00")},
        ],
        "schema": [
            {"name": "customer", "type": "STRING"},
            {"name": "amount", "type": "DECIMAL(10, 2)"},
        ],
    },
    platform=_PLATFORM,
)
def test_precise_types_resolved(case: EvalCase) -> None:
    """Assert exact rows plus precise column types recovered from the warehouse."""
    solver = CallableSolver(lambda c: f"SELECT customer, amount FROM {_ORDERS} ORDER BY id")
    assert_eval(case, solver, scorers=[ResultSetEquivalence()])


# Untyped result set: values only, no column-type assertion.
@eval_case(
    input="What is the total order amount?",
    expected={"rows": [{"total": Decimal("35.50")}]},
    platform=_PLATFORM,
)
def test_untyped_total(case: EvalCase) -> None:
    """Compare a warehouse-computed aggregate by value only."""
    solver = CallableSolver(lambda c: f"SELECT sum(amount) AS total FROM {_ORDERS}")
    assert_eval(case, solver, scorers=[ResultSetEquivalence()])


# Gold query: the reference query's executed RESULT is the expected answer (execution
# accuracy). The comparison is on the executed result, not the SQL text, so any query that
# returns the same rows passes.
@eval_case(
    input="What is the total order amount per customer?",
    expected={
        "kind": "gold_query",
        "sql": f"SELECT customer, sum(amount) AS total FROM {_ORDERS} GROUP BY customer",
    },
    platform=_PLATFORM,
)
def test_gold_query(case: EvalCase) -> None:
    """Score against a reference query's executed result (execution accuracy)."""
    solver = CallableSolver(
        lambda c: f"SELECT customer, sum(amount) AS total FROM {_ORDERS} GROUP BY 1 ORDER BY customer DESC"
    )
    assert_eval(case, solver, scorers=[ResultSetEquivalence()])


# Expectation suite: structural assertions (`row_count` / `not_null` / `unique`) pushed into
# the warehouse and evaluated as server-side SQL.
@eval_case(
    input="List every order's id and customer.",
    expected={
        "kind": "expectation_suite",
        "expectations": [
            {"kind": "row_count", "exact": 3},
            {"kind": "not_null", "column": "id"},
            {"kind": "unique", "column": "id"},
        ],
    },
    platform=_PLATFORM,
)
def test_expectation_suite_pushdown(case: EvalCase) -> None:
    """Assert structural properties of the result, evaluated server-side."""
    solver = CallableSolver(lambda c: f"SELECT id, customer FROM {_ORDERS}")
    assert_eval(case, solver, scorers=[ExpectationSuiteScorer()])

The example reads its warehouse connection from the environment, so set these before running:

  • DATABRICKS_SERVER_HOSTNAME, DATABRICKS_HTTP_PATH: your warehouse's host and HTTP path. These are arguments to databricks_platform(); pass them as literals if you prefer.
  • DATABRICKS_TOKEN: read by the Databricks SDK to authenticate. Or use another method it supports, e.g. databricks auth login for OAuth.

Run it

uv run pytest test_databricks.py -q

Run it from a clone

This is the bundled examples/04_databricks/ example. If you've cloned the repo, run it with uv run pytest examples/04_databricks.

Next steps