fix: error a unit test when its tested output query fails (#1208)

* fix: error a unit test when its tested output query fails

Storing a unit test's actual output once and comparing against it fell
back to the inline comparison on any storage error. On a warehouse that
prunes unread columns, the inline comparison of the expected columns can
pass when computing the full output fails, for example when a fixture value
cannot be converted in a column the expected rows leave out, so the test
reported PASS over a warehouse error. When storing fails, the output query
now runs on its own; if it fails too, the test errors with the warehouse's
message, and only a failure to store a computable output compares inline.

* fix: check a unit test output that cannot be stored on its own, inline

When a later chain step's output could not be stored, the probe ran that
step's query reading the earlier steps' stored outputs, which had just been
dropped, so it reported a missing table instead of comparing inline. The
probe now renders the failing step fully inline, runs only when storing
itself failed rather than rendering, and keeps the storage error in the
debug log when the step's query fails on its own.

* fix: compare a unit test inline when DuckDB fails to store it internally

DuckDB 1.5.0 to 1.5.4 raise an internal binder error for two identical
window functions over a GROUP BY of a constant input, which is the shape
of a unit test's fixture CTE. Storing the output and running it on its own
both hit it, so the output check reported the engine bug as the model's
error while the inline comparison computes every column correctly. An
internal engine error while storing an output now compares inline.

Author: kvlonge

src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/testing.md

@@ -919,7 +919,12 @@ SQLBuild drops after the test, including after a failure or timeout. PostgreSQL
 one transaction with an `ON COMMIT DROP` temporary table that the transaction's end removes, so the
 reads stay on one server session behind a transaction-mode connection pooler. Snowflake reads the
 first query's persisted result through `RESULT_SCAN`. If a stored output can no longer be read,
-the test falls back to the inline comparison. Other adapters compare the inline query as before. `target/run/tests/` keeps the self-contained inline comparison query for each test.
+the test falls back to the inline comparison. If storing an output fails, SQLBuild runs the output
+query on its own: when that query fails too, for example because a fixture value cannot be
+converted in a column the expected rows leave out, the test errors with the warehouse's message
+instead of passing on a comparison of the other columns; otherwise it compares inline. On DuckDB and
+MotherDuck, an internal engine error while storing an output is a DuckDB defect rather than a
+problem with the model, so the test compares inline. Other adapters compare the inline query as before. `target/run/tests/` keeps the self-contained inline comparison query for each test.
 
 ## Skipping SQL analysis
 

src/sqlbuild/adapter/contract/classes/duckdb_backed_adapter.py

@@ -40,6 +40,7 @@
     CANCELLABLE_DURATION_LIMIT_SECONDS,
     DIFF_LEFT_SIDE,
     DIFF_RIGHT_SIDE,
+    DUCKDB_INTERNAL_ERROR_CLASS_NAME,
     DUCKDB_MISSING_OBJECT_MARKER,
     DUCKDB_NOT_FOUND_ERROR_CLASS_NAMES,
     QUALIFIED_NAME_SEPARATOR,
@@ -91,6 +92,8 @@
 from sqlbuild.adapter.type_system.main.normalize_numeric_family import normalize_numeric_family
 from sqlbuild.adapter.type_system.main.types_equal import types_equal
 from sqlbuild.compiler.compile.types import FunctionLanguage
+from sqlbuild.diagnostics.classes.diagnostic_record_redactor import DiagnosticRecordRedactor
+from sqlbuild.diagnostics.main.log_debug_event import log_debug_event
 from sqlbuild.diagnostics.main.log_sql import log_sql
 from sqlbuild.runtime.user_code.main.user_code_spec import user_code_spec
 from sqlbuild.spec.contracts.constants import DEFAULT_SEED_CSV_SETTINGS
@@ -118,9 +121,19 @@ def get_columns_for_relations(
     execution_duration_limit_seconds: ClassVar[int | None] = CANCELLABLE_DURATION_LIMIT_SECONDS
 
     def create_sql_test_output(self, *, connection: Any, sql: str, name: str) -> str | None:
-        """Store a unit test's actual output once in a session temporary table."""
-
-        _ = self.execute(connection=connection, sql=f"CREATE TEMPORARY TABLE {name} AS {sql}")
+        """Store a unit test's output in a temporary table; compare inline on an engine bug."""
+
+        try:
+            _ = self.execute(connection=connection, sql=f"CREATE TEMPORARY TABLE {name} AS {sql}")
+        except Exception as error:
+            if type(error).__name__ != DUCKDB_INTERNAL_ERROR_CLASS_NAME:
+                raise
+            log_debug_event(
+                logger=logging.getLogger("sqlbuild.adapter.duckdb"),
+                message="DuckDB could not store a SQL test output; comparing inline",
+                sqlbuild_error=DiagnosticRecordRedactor.text(str(error)),
+            )
+            return None
         return name
 
     def drop_sql_test_output(self, *, connection: Any, relation: str) -> None:

src/sqlbuild/adapter/contract/constants.py

@@ -17,4 +17,5 @@
     {"CatalogException", "BinderException"}
 )
 DUCKDB_MISSING_OBJECT_MARKER: str = "does not exist"
+DUCKDB_INTERNAL_ERROR_CLASS_NAME: str = "InternalException"
 CANCELLABLE_DURATION_LIMIT_SECONDS: int = 2_147_483

src/sqlbuild/executor/testing/classes/sql_test_actual_outputs.py

@@ -50,22 +50,19 @@ def materialize(self) -> dict[int, str]:
         """Store each compared step's output, or return no relations to compare inline."""
 
         token: str = uuid4().hex[:12]
+        storing: int | None = None
         try:
             for index in range(len(self._test_entry.chain)):
-                (sql,) = render_sql_test_comparison_sql(
-                    (self._test_entry,),
-                    set_difference_operator=self._adapter.render_set_difference_operator(),
-                    sql_analysis_dialect=self._adapter.sql_analysis_dialect(),
-                    output_step_index=index,
-                    actual_relations=dict(self.relations),
-                )
+                sql: str = self._output_sql(index=index, relations=dict(self.relations))
                 if not sql:
                     continue
+                storing = index
                 relation: str | None = self._adapter.create_sql_test_output(
                     connection=self._connection,
                     sql=sql,
                     name=f"{_OUTPUT_PREFIX}_{token}_{index}",
                 )
+                storing = None
                 if relation is None:
                     self.release()
                     return {}
@@ -75,16 +72,50 @@ def materialize(self) -> dict[int, str]:
         except (ItemTimeoutError, ExecutionDurationLimitError):
             raise
         except Exception as error:
+            self.release()
+            self._evaluate_output(index=storing, storage_error=error)
             log_debug_event(
                 logger=_LOGGER,
                 message="SQL test output could not be stored; comparing inline",
                 test_name=self._test_entry.name,
                 sqlbuild_error=DiagnosticRecordRedactor.text(str(error)),
             )
-            self.release()
             return {}
         return dict(self.relations)
 
+    def _output_sql(self, *, index: int, relations: dict[int, str]) -> str:
+        """The SQL of step ``index``'s output, reading the given stored outputs of earlier steps."""
+
+        (sql,) = render_sql_test_comparison_sql(
+            (self._test_entry,),
+            set_difference_operator=self._adapter.render_set_difference_operator(),
+            sql_analysis_dialect=self._adapter.sql_analysis_dialect(),
+            output_step_index=index,
+            actual_relations=relations,
+        )
+        return sql
+
+    def _evaluate_output(self, *, index: int | None, storage_error: Exception) -> None:
+        """Run a step that could not be stored fully inline, raising the query's own error."""
+
+        if index is None:
+            return
+        try:
+            cursor: Any = self._adapter.execute(
+                connection=self._connection, sql=self._output_sql(index=index, relations={})
+            )
+            _ = cursor.fetchall()
+        except (ItemTimeoutError, ExecutionDurationLimitError):
+            raise
+        except Exception:
+            log_debug_event(
+                logger=_LOGGER,
+                message="SQL test output could not be stored, and its query fails on its own",
+                test_name=self._test_entry.name,
+                sqlbuild_error=DiagnosticRecordRedactor.text(str(storage_error)),
+            )
+            raise
+
     def release(self) -> None:
         """Drop every stored output; later comparisons run inline."""
 

tests/e2e/src/sqlbuild/cli/commands/main/test/_test_types.py

@@ -470,3 +470,16 @@ class TemporaryFunctionConnectionE2ETestCase:
     description: str
     expected_exit_code: int
     expected_stderr_fragment: str
+
+
+@dataclass(frozen=True)
+class OutputEvaluationErrorE2ETestCase:
+    """How a unit test ends when storing its tested model's output fails for a given reason."""
+
+    description: str
+    adapter_source: str
+    extra_files: dict[str, str]
+    expected_exit_code: int
+    expected_fragments: tuple[str, ...]
+    expected_json_fragments: tuple[str, ...]
+    unexpected_fragments: tuple[str, ...]

tests/e2e/src/sqlbuild/cli/commands/main/test/helpers.py

@@ -2096,3 +2096,20 @@ def slow_promotion_files(
             f"SELECT {customer_id} AS customer_id\n"
         ),
     }
+
+
+def order_totals_test_files(total_sql: str) -> dict[str, str]:
+    """Return an order_totals model computing ``total_sql`` and a test expecting both models."""
+
+    return {
+        "models/order_totals.sql": (
+            "MODEL (description 'Test model order_totals.');\n\n"
+            f'SELECT order_id, {total_sql} AS total FROM __ref("orders")\n'
+        ),
+        "tests/unit/test_orders.sql": (
+            'TEST (name "orders_ids");\n\nWITH\n'
+            "__source__raw_orders AS (\n  SELECT 1 AS order_id, '7' AS code\n),\n"
+            "__expected__orders AS (\n  SELECT 1 AS order_id\n),\n"
+            "__expected__order_totals AS (\n  SELECT 1 AS order_id\n)\nSELECT 1\n"
+        ),
+    }

tests/e2e/src/sqlbuild/cli/commands/main/test/test_output_evaluation_errors.py

@@ -0,0 +1,174 @@
+"""E2E coverage: a warehouse error in a tested model's output fails the unit test."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from tests.e2e.src.sqlbuild.cli.commands.main.test._test_types import (
+    OutputEvaluationErrorE2ETestCase,
+)
+from tests.e2e.src.sqlbuild.cli.commands.main.test.helpers import order_totals_test_files
+from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import (
+    execute_duckdb,
+    prepare_inline_project,
+    run_sqb,
+)
+
+_CONVERSION_ERROR: str = "Numeric value 'cluster::r1' is not recognized"
+# Like a warehouse that prunes unread columns: the stored output computes every column and
+# fails on the fixture, while an inline comparison of the expected columns alone succeeds.
+_PRUNING_WAREHOUSE: str = (
+    "from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter\n\n\n"
+    "class PruningDuckDbAdapter(DuckDbAdapter):\n"
+    '    adapter_name = "pruning_duckdb"\n'
+    "    failing_outputs = set()\n\n"
+    "    def create_sql_test_output(self, *, connection, sql, name):\n"
+    "        self.failing_outputs.add(sql)\n"
+    f'        raise RuntimeError("{_CONVERSION_ERROR}")\n\n'
+    "    def execute(self, *, connection, sql):\n"
+    "        if sql in self.failing_outputs:\n"
+    f'            raise RuntimeError("{_CONVERSION_ERROR}")\n'
+    "        return super().execute(connection=connection, sql=sql)\n"
+)
+# A session that cannot keep test outputs at all, while every query runs.
+_NO_TEMPORARY_TABLES: str = (
+    "from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter\n\n\n"
+    "class PruningDuckDbAdapter(DuckDbAdapter):\n"
+    '    adapter_name = "pruning_duckdb"\n\n'
+    "    def create_sql_test_output(self, *, connection, sql, name):\n"
+    '        raise PermissionError("temporary tables are disabled on this session")\n'
+)
+# A session that keeps one test output and refuses every later one.
+_ONE_TEMPORARY_TABLE: str = (
+    "from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter\n\n\n"
+    "class PruningDuckDbAdapter(DuckDbAdapter):\n"
+    '    adapter_name = "pruning_duckdb"\n'
+    "    stored = []\n\n"
+    "    def create_sql_test_output(self, *, connection, sql, name):\n"
+    "        self.stored.append(name)\n"
+    "        if len(self.stored) > 1:\n"
+    '            raise PermissionError("no more temporary tables in this session")\n'
+    "        return super().create_sql_test_output(connection=connection, sql=sql, name=name)\n"
+)
+
+
+_PROJECT_FILES: dict[str, str] = {
+    "sqlbuild_project.toml": (
+        'name = "orders"\nadapter = "pruning_duckdb"\n\n'
+        '[connection]\ndatabase = "orders.duckdb"\n\n'
+        '[defaults]\nmaterialized = "table"\n'
+    ),
+    "sources/raw_orders.yml": (
+        "sources:\n"
+        "  - name: raw_orders\n    description: Test source raw_orders.\n"
+        "    schema: main\n"
+        "    table: raw_orders\n"
+        "    columns:\n"
+        "      - name: order_id\n        type: INTEGER\n"
+        "      - name: code\n        type: VARCHAR\n"
+    ),
+    "models/orders.sql": (
+        "MODEL (description 'Test model orders.');\n\n"
+        'SELECT order_id, TRY_CAST(code AS INTEGER) AS code_number FROM __source("raw_orders")\n'
+    ),
+    "tests/unit/test_orders.sql": (
+        'TEST (name "orders_ids");\n\nWITH\n'
+        "__source__raw_orders AS (\n  SELECT 1 AS order_id, 'cluster::r1' AS code\n),\n"
+        "__expected__orders AS (\n  SELECT 1 AS order_id\n)\nSELECT 1\n"
+    ),
+}
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        OutputEvaluationErrorE2ETestCase(
+            description="an output the warehouse cannot compute errors the test",
+            adapter_source=_PRUNING_WAREHOUSE,
+            extra_files={},
+            expected_exit_code=1,
+            expected_fragments=(
+                "test 'orders_ids' encountered an execution error while running 'orders'",
+                _CONVERSION_ERROR,
+                "PASS=0  FAIL=1",
+            ),
+            expected_json_fragments=('"error"', _CONVERSION_ERROR),
+            unexpected_fragments=("PASS=1  FAIL=0",),
+        ),
+        OutputEvaluationErrorE2ETestCase(
+            description="a session that cannot store outputs still compares inline",
+            adapter_source=_NO_TEMPORARY_TABLES,
+            extra_files={},
+            expected_exit_code=0,
+            expected_fragments=("PASS=1  FAIL=0",),
+            expected_json_fragments=('"pass"',),
+            unexpected_fragments=("encountered an execution error",),
+        ),
+        OutputEvaluationErrorE2ETestCase(
+            description="a computable later output that cannot be stored compares inline",
+            adapter_source=_ONE_TEMPORARY_TABLE,
+            extra_files=order_totals_test_files("code_number + 1"),
+            expected_exit_code=0,
+            expected_fragments=("PASS=1  FAIL=0",),
+            expected_json_fragments=('"pass"',),
+            unexpected_fragments=("encountered an execution error", "does not exist"),
+        ),
+        OutputEvaluationErrorE2ETestCase(
+            description="a later output that fails on its own reports the warehouse error",
+            adapter_source=_ONE_TEMPORARY_TABLE,
+            extra_files=order_totals_test_files("CAST('cluster::r1' AS INTEGER)"),
+            expected_exit_code=1,
+            expected_fragments=("Could not convert string 'cluster::r1'", "PASS=0  FAIL=1"),
+            expected_json_fragments=('"error"', "Could not convert string"),
+            unexpected_fragments=("does not exist",),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_output_store_failure_when_testing_then_only_storage_problems_fall_back(
+    test_case: OutputEvaluationErrorE2ETestCase, tmp_path: Path
+) -> None:
+    """An inline comparison of some columns must not hide an error computing the others."""
+
+    project_dir: Path = prepare_inline_project(
+        tmp_path=tmp_path,
+        project_name="orders",
+        repo_files={
+            **_PROJECT_FILES,
+            "adapters/pruning_duckdb.py": test_case.adapter_source,
+            **test_case.extra_files,
+        },
+    )
+    config_path: Path = project_dir / "sqlbuild_project.toml"
+    _ = config_path.write_text(
+        config_path.read_text(encoding="utf-8").replace(
+            '"orders.duckdb"', f'"{(project_dir / "orders.duckdb").as_posix()}"'
+        ),
+        encoding="utf-8",
+    )
+    execute_duckdb(
+        db_path=project_dir / "orders.duckdb",
+        sql="CREATE TABLE raw_orders AS SELECT 1 AS order_id, '1' AS code",
+    )
+
+    tested: subprocess.CompletedProcess[str] = run_sqb(
+        command=("--no-color", "test"), project_dir=project_dir
+    )
+    as_json: subprocess.CompletedProcess[str] = run_sqb(
+        command=("--no-color", "test", "--json"), project_dir=project_dir
+    )
+    payload: dict[str, Any] = json.loads(as_json.stdout)
+
+    output: str = tested.stdout + tested.stderr
+    assert tested.returncode == test_case.expected_exit_code, output
+    assert as_json.returncode == test_case.expected_exit_code, as_json.stdout
+    assert all(fragment in output for fragment in test_case.expected_fragments), output
+    assert not any(fragment in output for fragment in test_case.unexpected_fragments), output
+    assert all(fragment in json.dumps(payload) for fragment in test_case.expected_json_fragments), (
+        as_json.stdout
+    )

tests/integration/src/sqlbuild/executor/testing/_test_types.py

@@ -64,3 +64,36 @@ class SqlTestBuildParityTestCase:
     model_name: str
     expected_whole_body_rendered: bool
     expected_verbatim_fragments: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True)
+class OutputStorageTestCase:
+    """A tested output that cannot be stored, and whether the test still runs inline."""
+
+    description: str
+    chain_steps: tuple[tuple[str, str, str | None], ...]
+    storage_error: str
+    output_fails: bool
+    expected_outcome: SqlTestOutcome
+    expected_error_fragment: str
+
+
+@dataclass(frozen=True)
+class LaterStepStorageTestCase:
+    """A chain whose first output is stored and whose second cannot be, and how the test ends."""
+
+    description: str
+    chain_steps: tuple[tuple[str, str, str | None], ...]
+    expected_outcome: SqlTestOutcome
+    expected_error_fragment: str
+    unexpected_error_fragment: str
+
+
+@dataclass(frozen=True)
+class EngineInternalErrorTestCase:
+    """A tested output DuckDB fails to store with an engine assertion error."""
+
+    description: str
+    chain_steps: tuple[tuple[str, str, str | None], ...]
+    raise_internal_error: bool
+    expected_outcome: SqlTestOutcome