feat: improve native test diagnostics

src/sqlbuild/.agents/skills/sqlbuild/SKILL.md

@@ -6592,6 +6592,17 @@ columns, and 120 characters per value, and pass through diagnostic redaction. Th
 as structured `unexpected_samples` and `missing_samples` in JSON output. Sampling failure never
 replaces the known test failure.
 
+When exactly one row exists in each difference direction and the bounded samples have the same
+columns, SQLBuild also aligns the rows and reports only changed columns with their redacted actual
+and expected values. It does not guess an alignment when several rows differ or sampling is
+incomplete.
+
+Before opening a warehouse connection, SQLBuild statically checks fixture shapes when SQL analysis
+is enabled. Missing columns are reported together with the test path, fixture resource, and models
+that read them. Statically provable collection-versus-scalar type conflicts identify the affected
+column and suggest explicit `CAST`, `ARRAY_CONSTRUCT`, or `PARSE_JSON` expressions. SQLBuild never
+invents missing values.
+
 The trailing `SELECT 1` is required as a ceremonial closing statement.
 
 ### CTE conventions
@@ -6634,6 +6645,17 @@ SELECT 1
 
 SQLBuild topologically sorts the expected models, resolves each intermediate model's real SQL with mocks substituted, and chains the outputs forward. Every model between the mocked sources and the expected model is computed automatically.
 
+Inspect that boundary without opening a warehouse connection:
+
+```bash
+sqb test --select fact_orders --inspect
+```
+
+The resolved plan lists mocked sources, refs, seeds, real models in execution order, expected
+models, and any unsatisfied leaf dependencies. A mocked ref is called out explicitly because its
+real model SQL will not execute. Inspection exits non-zero when the plan contains unresolved
+dependency errors.
+
 ### Mocking refs and seeds
 
 You can mock models directly with `__ref__<name>` and seeds with `__seed__<name>`, not just sources. This skips the model's real SQL (or the seed's real CSV data) and provides controlled data instead:
@@ -6965,8 +6987,15 @@ SELECT 1
 ```
 
 The cases report independently as `order status: maps source states [completed]`, `[cancelled]`,
-and `[pending]`. Authored order controls display order. Selection remains at the parent test's
-resolved model or resource, so selecting `stg_orders` selects every case; there is no case selector.
+and `[pending]`. Authored order controls display order. Selecting `stg_orders` selects every case by
+default. Add `--case` to run one named case without changing the test file:
+
+```bash
+sqb test --select stg_orders --case cancelled
+```
+
+If the case does not exist within the parent/model selection, SQLBuild reports the available case
+names.
 
 Supported scalar types are `string`, `integer`, `boolean`, `float`, and exact `decimal`. Decimal
 values are quoted, such as `tax_rate "0.2000"`, so they never pass through binary float. Boolean
@@ -7028,6 +7057,9 @@ SELECT 1
 This produces one aggregate test result. Use native cases when each row must pass or fail
 independently.
 
+Quoted strings in `VALUES` rows may contain commas, brackets, and parentheses. SQLBuild preserves
+these literals while extracting test CTEs; use normal SQL quote escaping for embedded quotes.
+
 ### Test placement
 
 Place unit test files under `tests/unit/` in your project directory. SQLBuild discovers all `.sql` files in this directory recursively.

src/sqlbuild/cli/commands/_helpers/entry/dispatch.py

@@ -247,6 +247,8 @@ def dispatch_cli_command(*, args: CliNamespace, handlers: CliEntrypointHandlers)
                 json_output=args.json,
                 json_output_path=args.json_output,
                 concurrency=args.concurrency,
+                case_name=args.case,
+                inspect=args.inspect,
             )
         )
     if args.command == CliCommand.CHECK:

src/sqlbuild/cli/commands/_helpers/entry/parser.py

@@ -276,6 +276,10 @@ def _add_quality_parsers(
     test_parser.add_argument("--json", action="store_true", default=False)
     test_parser.add_argument("--target", default=None)
     test_parser.add_argument("--concurrency", type=int, default=None)
+    test_parser.add_argument("--case", default=None, help="Run one named parameter case")
+    test_parser.add_argument(
+        "--inspect", action="store_true", default=False, help="Show the resolved test plan"
+    )
     _ = add_execution_json_output_arg(test_parser)
     _ = add_select_args(test_parser)
     _ = add_vars_args(test_parser)

src/sqlbuild/cli/commands/_helpers/entry/parsing.py

@@ -133,6 +133,8 @@ def parse_cli_invocation(
                 parser.error(str(error))
         if args.command == CliCommand.SCOPE:
             _validate_scope_args(args=args, parser=parser)
+        if args.command == CliCommand.TEST:
+            _validate_test_args(args=args, parser=parser)
     except SystemExit as error:
         exit_code: int = error.code if isinstance(error.code, int) else 1
         return ParsedCliInvocation(args=None, exit_code=exit_code)
@@ -181,3 +183,8 @@ def _validate_scope_args(*, args: CliNamespace, parser: argparse.ArgumentParser)
             parser.error("scope --explain requires a declaration-qualified identity")
         if not qualified_name:
             parser.error("scope --explain requires a declaration-qualified identity")
+
+
+def _validate_test_args(*, args: CliNamespace, parser: argparse.ArgumentParser) -> None:
+    if args.inspect and (args.json or args.json_output is not None):
+        parser.error("test --inspect cannot be combined with --json or --json-output")

src/sqlbuild/cli/commands/_helpers/test/inspection.py

@@ -0,0 +1,59 @@
+"""Resolved SQL-test plan inspection output."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import TextIO
+
+from sqlbuild.compiler.planner.models import PlanOutput, PlanWarning, SqlTestPlanEntry
+from sqlbuild.compiler.planner.types import WarningSeverity
+
+
+def write_test_plan_inspection(*, stream: TextIO, plan_output: PlanOutput) -> int:
+    """Write selected test fixture boundaries and real model chains without executing them."""
+
+    stream.write("\nResolved test plan\n\n")
+    for entry in plan_output.test_entries:
+        stream.write(f"{_entry_label(entry)}\n")
+        _write_names(stream=stream, label="mocked sources", names=entry.mock_source_names)
+        _write_names(stream=stream, label="mocked refs", names=entry.mock_ref_names)
+        _write_names(stream=stream, label="mocked seeds", names=entry.mock_seed_names)
+        _write_names(stream=stream, label="mocked dbt refs", names=entry.mock_dbt_ref_names)
+        _write_names(
+            stream=stream,
+            label="real models",
+            names=tuple(step.model_name for step in entry.chain),
+        )
+        _write_names(
+            stream=stream,
+            label="expected models",
+            names=tuple(
+                step.model_name for step in entry.chain if step.expected_cte_sql is not None
+            ),
+        )
+        for mock_name in entry.mock_ref_names:
+            stream.write(f"  boundary: {mock_name} is replaced by __ref__{mock_name}\n")
+        stream.write("\n")
+    error_warnings: tuple[PlanWarning, ...] = tuple(
+        warning for warning in plan_output.warnings if warning.severity == WarningSeverity.ERROR
+    )
+    for warning in plan_output.warnings:
+        stream.write(f"  {warning.severity.value}: {warning.message}\n")
+    outcome: str = "failed" if error_warnings else "complete"
+    stream.write(
+        f"Test plan inspection {outcome}: {len(plan_output.test_entries)} selected, "
+        f"{len(error_warnings)} errors.\n"
+    )
+    stream.flush()
+    return 1 if error_warnings else 0
+
+
+def _write_names(*, stream: TextIO, label: str, names: Iterable[str]) -> None:
+    values: tuple[str, ...] = tuple(names)
+    stream.write(f"  {label}: {', '.join(values) if values else 'none'}\n")
+
+
+def _entry_label(entry: SqlTestPlanEntry) -> str:
+    if entry.case_name is None:
+        return entry.name
+    return f"{entry.parent_name or entry.name} [{entry.case_name}]"

src/sqlbuild/cli/commands/_helpers/test/planning.py

@@ -13,7 +13,9 @@
     CompilePipelineResult,
     StaticCommandContext,
 )
+from sqlbuild.compiler.planner.exceptions import PlannerInputError
 from sqlbuild.compiler.planner.main.commands.sql_test import build_test_command_plan
+from sqlbuild.compiler.planner.models import PlanOutput
 
 
 def compile_test_plan(
@@ -42,12 +44,33 @@ def compile_test_plan(
         options=options,
         on_progress=invocation.planning_progress.on_progress,
     )
+    plan_output: PlanOutput = build_test_command_plan(
+        project=context.project,
+        adapter=invocation.adapter,
+        scope=context.scope,
+        relations=context.relations,
+        case_name=request.case_name,
+    )
+    if request.case_name is not None:
+        plan_output = _validate_parameter_case_selection(
+            plan_output=plan_output,
+            case_name=request.case_name,
+            available_cases=plan_output.available_test_case_names,
+        )
     return build_static_pipeline_result(
         context=context,
-        plan_output=build_test_command_plan(
-            project=context.project,
-            adapter=invocation.adapter,
-            scope=context.scope,
-            relations=context.relations,
-        ),
+        plan_output=plan_output,
+    )
+
+
+def _validate_parameter_case_selection(
+    *, plan_output: PlanOutput, case_name: str, available_cases: tuple[str, ...]
+) -> PlanOutput:
+    if plan_output.test_entries:
+        return plan_output
+    available_text: str = ", ".join(available_cases) or "none"
+    raise PlannerInputError(
+        f"SQL test case '{case_name}' did not match the selected tests; available cases: "
+        f"{available_text}",
+        code="S008",
     )

src/sqlbuild/cli/commands/classes/cli_namespace.py

@@ -25,6 +25,8 @@
     "defer_clone_from": None,
     "defer_sources_to": None,
     "target": None,
+    "case": None,
+    "inspect": False,
     "target_range": None,
     "from_target": None,
     "to_target": None,
@@ -175,6 +177,8 @@ class CliNamespace:
     defer_clone_from: str | None
     defer_sources_to: str | None
     target: str | None
+    case: str | None
+    inspect: bool
     target_range: str | None
     from_target: str | None
     to_target: str | None

src/sqlbuild/cli/commands/main/execution/_test.py

@@ -7,6 +7,7 @@
     prepare_test_execution,
     resolve_test_concurrency,
 )
+from sqlbuild.cli.commands._helpers.test.inspection import write_test_plan_inspection
 from sqlbuild.cli.commands._helpers.test.invocation import resolve_test_invocation
 from sqlbuild.cli.commands._helpers.test.outputs import (
     resolve_test_exit_code,
@@ -32,6 +33,11 @@ def run_test(request: TestCommandRequest) -> int:
         request=request,
         invocation=invocation,
     )
+    if request.inspect:
+        return write_test_plan_inspection(
+            stream=invocation.progress_stream,
+            plan_output=pipeline_result.plan_output,
+        )
     invocation.progress_stream.write("\n")
     write_execution_header(
         stream=invocation.progress_stream,

src/sqlbuild/cli/commands/models.py

@@ -1288,6 +1288,8 @@ class TestCommandRequest:
     json_output: bool = False
     json_output_path: Path | None = None
     concurrency: int | None = None
+    case_name: str | None = None
+    inspect: bool = False
 
 
 @dataclass(frozen=True)

src/sqlbuild/cli/output/_helpers/execution_result_document.py

@@ -958,6 +958,15 @@ def _format_sql_test_step(step: StepResult) -> dict[str, object]:
             "missing_row_count": step.missing_row_count,
             "unexpected_samples": _format_sql_test_difference_samples(step.unexpected_samples),
             "missing_samples": _format_sql_test_difference_samples(step.missing_samples),
+            "column_differences": [
+                {
+                    "name": difference.name,
+                    "actual": difference.actual,
+                    "expected": difference.expected,
+                }
+                for difference in step.column_differences
+            ]
+            or None,
             "error_code": step.error_code,
             "error_help": step.error_help,
             "error_message": step.error_message,

src/sqlbuild/cli/progress/main/_expectation_detail.py

@@ -34,6 +34,11 @@ def format_expectation_detail(step_result: StepResult) -> str:
         )
     parts.extend(_format_samples(label="unexpected sample", samples=step_result.unexpected_samples))
     parts.extend(_format_samples(label="missing sample", samples=step_result.missing_samples))
+    for difference in step_result.column_differences:
+        parts.append(
+            f"differing column {difference.name}: actual={difference.actual}, "
+            f"expected={difference.expected}"
+        )
     return f"  {'; '.join(parts)}"
 
 

src/sqlbuild/compiler/compile/_helpers/assembly/project.py

@@ -1364,6 +1364,7 @@ def _assemble_compiled_sql_test(
             mock_source_names=model_payload.mock_source_names,
             mock_seed_names=model_payload.mock_seed_names,
             mock_dbt_ref_names=model_payload.mock_dbt_ref_names,
+            expected_ctes=model_payload.expected_ctes,
             expected_model_names=model_payload.expected_model_names,
             assertion_ctes=model_payload.assertion_ctes,
             assertion_names=model_payload.assertion_names,

src/sqlbuild/compiler/compile/_helpers/attachment/sql_tests.py

@@ -370,6 +370,7 @@ def _build_test_input_payload(
                 mock_source_names=model_payload.mock_source_names,
                 mock_seed_names=model_payload.mock_seed_names,
                 mock_dbt_ref_names=model_payload.mock_dbt_ref_names,
+                expected_ctes=model_payload.expected_ctes,
                 expected_model_names=model_payload.expected_model_names,
                 assertion_ctes=model_payload.assertion_ctes,
                 assertion_names=model_payload.assertion_names,

src/sqlbuild/compiler/compile/_helpers/sql_tests/core.py

@@ -376,6 +376,7 @@ def _classify_model_sql_test_ctes(
     mock_source_names: list[str] = []
     mock_seed_names: list[str] = []
     mock_dbt_ref_names: list[str] = []
+    expected_ctes: list[CompileSqlTestCte] = []
     expected_model_names: list[str] = []
     assertion_ctes: list[CompileSqlTestCte] = []
     assertion_names: list[str] = []
@@ -460,6 +461,7 @@ def _classify_model_sql_test_ctes(
                 )
             )
             _validate_expected_cte_query(cte=cte, file_label=file_label)
+            expected_ctes.append(cte)
             continue
         if cte.name.startswith(ASSERT_TEST_CTE_PREFIX):
             assertion_names.append(
@@ -502,6 +504,7 @@ def _classify_model_sql_test_ctes(
         mock_source_names=tuple(mock_source_names),
         mock_seed_names=tuple(mock_seed_names),
         mock_dbt_ref_names=tuple(mock_dbt_ref_names),
+        expected_ctes=tuple(expected_ctes),
         expected_model_names=tuple(expected_model_names),
         assertion_ctes=tuple(assertion_ctes),
         assertion_names=tuple(assertion_names),

src/sqlbuild/compiler/compile/constants.py

@@ -82,6 +82,10 @@
 TEMPLATE_ESCAPE_TOKEN: str = "\\"
 TEMPLATE_OPEN_TOKEN: str = "${"
 TEMPLATE_FALSE_VALUES: frozenset[str] = frozenset({"", "0", "false"})
+POLYGLOT_LITERAL_KIND: str = "literal"
+POLYGLOT_ARRAY_KIND: str = "array_func"
+POLYGLOT_STRUCTURED_KINDS: frozenset[str] = frozenset({"parse_json", "struct"})
+POLYGLOT_FUNCTION_KIND: str = "function"
 MISSING_TEMPLATE_VALUE_MESSAGE_PARTS: frozenset[str] = frozenset(
     {
         "references missing ENV variable",