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

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