Skip to content

Evaluate text-to-SQL on Snowflake

This guide shows how to connect, authenticate, and run SQL evals in Snowflake.

Prerequisites

uv add "evaldata[snowflake]"

What this guide covers

  • Data checks. row_count, not_null, and unique expectations run in Snowflake.
  • Authentication. snowflake_platform(...) stores the account, warehouse, and role. Environment variables provide the identity.
  • Test data. The fixture creates a temporary table for the eval, then removes it.

Configure a connection

Describe your Snowflake connection with snowflake_platform:

from evaldata.platforms import snowflake_platform

platform = snowflake_platform(
    name="snowflake",
    account="myorg-myaccount",
    user="EVALDATA_SVC",
    warehouse="COMPUTE_WH",
    role="EVALDATA_ROLE",
    database="EVALDATA",
    schema="PUBLIC",
)

account is the Snowflake account identifier, for example myorg-myaccount. Omit the .snowflakecomputing.com suffix.

warehouse, role, database, and schema set the session defaults. Include database and schema when your SQL uses unqualified table names.

When an eval runs, resolve(platform) opens a connection with the configured credentials.

Authentication

  • Password. Set SNOWFLAKE_PASSWORD, and pass user to snowflake_platform(...).
  • Key pair. Set SNOWFLAKE_PRIVATE_KEY_FILE to a PEM-encoded PKCS#8 private key. If the key is encrypted, also set SNOWFLAKE_PRIVATE_KEY_FILE_PWD.
  • Workload identity (OIDC). Pass authenticator="WORKLOAD_IDENTITY" and workload_identity_provider="OIDC" to snowflake_platform(...), then set SNOWFLAKE_TOKEN to the token issued by your CI provider.

Write the eval

Create test_snowflake.py:

"""Snowflake example evals: fixed SQL run against a live Snowflake warehouse.

Expectation checks are pushed down into SQL, precise result-column types are resolved from the
warehouse, and execution accuracy compares against a gold query.

Set `SNOWFLAKE_ACCOUNT` (and `SNOWFLAKE_WAREHOUSE`/`SNOWFLAKE_ROLE`) and configure authentication
through the environment (see the Snowflake guide). Seeding targets `SNOWFLAKE_DATABASE` and
`SNOWFLAKE_SCHEMA` when set, defaulting to `EVALDATA_EXAMPLES`/`PUBLIC`.
"""

import os
from collections.abc import Iterator

import pytest

from evaldata import (
    CallableSolver,
    EvalCase,
    ExecutionAccuracy,
    ExpectationSuiteScorer,
    assert_eval,
    eval_case,
)
from evaldata.platforms import resolve, snowflake_platform
from evaldata.types import ExecutionFailure

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

_DATABASE = os.environ.get("SNOWFLAKE_DATABASE", "EVALDATA_EXAMPLES")
_SCHEMA = os.environ.get("SNOWFLAKE_SCHEMA", "PUBLIC")
_TABLE = f"{_DATABASE}.{_SCHEMA}.ORDERS_EX07"
_PLATFORM = snowflake_platform(
    name="examples-snowflake",
    account=os.environ.get("SNOWFLAKE_ACCOUNT", "your-account"),
    user=os.environ.get("SNOWFLAKE_USER"),
    warehouse=os.environ.get("SNOWFLAKE_WAREHOUSE"),
    role=os.environ.get("SNOWFLAKE_ROLE"),
    authenticator=os.environ.get("SNOWFLAKE_AUTHENTICATOR"),
    workload_identity_provider=os.environ.get("SNOWFLAKE_WORKLOAD_IDENTITY_PROVIDER"),
)


@pytest.fixture(scope="module", autouse=True)
def _seed_warehouse() -> Iterator[None]:
    adapter = resolve(_PLATFORM)
    statements = []
    if "SNOWFLAKE_DATABASE" not in os.environ:
        statements.append(f"CREATE DATABASE IF NOT EXISTS {_DATABASE}")
    if "SNOWFLAKE_SCHEMA" not in os.environ:
        statements.append(f"CREATE SCHEMA IF NOT EXISTS {_DATABASE}.{_SCHEMA}")
    statements += [
        f"CREATE OR REPLACE TABLE {_TABLE} (ID INT, CUSTOMER STRING, AMOUNT DECIMAL(10, 2))",
        f"INSERT INTO {_TABLE} VALUES (1, 'Ada', 10.00), (2, 'Bo', 5.50), (3, 'Cy', 20.00)",
    ]
    for sql in statements:
        result = adapter.execute(sql)
        if isinstance(result, ExecutionFailure):  # pragma: no cover
            msg = f"failed to seed Snowflake table {_TABLE!r}: {result.error.message}"
            raise RuntimeError(msg)
    yield


@eval_case(
    input="List each order's customer and amount, ordered by id.",
    expected={
        "kind": "gold_query",
        "sql": f"SELECT CUSTOMER, AMOUNT FROM {_TABLE} ORDER BY ID",
    },
    platform=_PLATFORM,
)
def test_ordered_list(case: EvalCase) -> None:
    """Score against a reference query's executed rows (execution accuracy)."""
    solver = CallableSolver(lambda c: f"SELECT CUSTOMER, AMOUNT FROM {_TABLE} ORDER BY ID")
    assert_eval(case, solver, scorers=[ExecutionAccuracy()])


@eval_case(
    input="What is the total order amount per customer?",
    expected={
        "kind": "gold_query",
        "sql": f"SELECT CUSTOMER, SUM(AMOUNT) AS TOTAL FROM {_TABLE} GROUP BY CUSTOMER",
    },
    platform=_PLATFORM,
)
def test_gold_query(case: EvalCase) -> None:
    """Score an unordered aggregate against a gold query, comparing rows by value."""
    solver = CallableSolver(
        lambda c: f"SELECT CUSTOMER, SUM(AMOUNT) AS TOTAL FROM {_TABLE} GROUP BY 1 ORDER BY CUSTOMER DESC"
    )
    assert_eval(case, solver, scorers=[ExecutionAccuracy(row_order="ignore")])


# `column_type` resolves the column's precise type from the warehouse: Snowflake reports the
# `AMOUNT` column as `NUMBER(10, 2)`, which matches the authored `DECIMAL(10, 2)`.
@eval_case(
    input="List every order's id, customer, and amount.",
    expected={
        "kind": "expectation_suite",
        "expectations": [
            {"kind": "row_count", "exact": 3},
            {"kind": "not_null", "column": "ID"},
            {"kind": "unique", "column": "ID"},
            {"kind": "column_type", "column": "AMOUNT", "expected_type": "DECIMAL(10, 2)"},
        ],
    },
    platform=_PLATFORM,
)
def test_expectation_suite_pushdown(case: EvalCase) -> None:
    """Assert structural properties and a precise column type, evaluated server-side."""
    solver = CallableSolver(lambda c: f"SELECT ID, CUSTOMER, AMOUNT FROM {_TABLE}")
    assert_eval(case, solver, scorers=[ExpectationSuiteScorer()])

resolve reuses one connection per platform name, so the fixture and the evals share a session.

Snowflake folds unquoted identifiers to uppercase, so the result columns are ID, CUSTOMER, and AMOUNT; the expectations and gold queries name them that way.

Run it

uv run pytest test_snowflake.py -q

Check the connection

evaldata doctor \
  --snowflake-account myorg-myaccount \
  --snowflake-user EVALDATA_SVC \
  --snowflake-warehouse COMPUTE_WH

The Snowflake doctor flags also read from environment variables:

  • SNOWFLAKE_ACCOUNT
  • SNOWFLAKE_USER
  • SNOWFLAKE_WAREHOUSE
  • SNOWFLAKE_ROLE

With these set, evaldata doctor checks the connection without extra flags.

Next steps