feat: improve Dagster checks and test fixtures

AGENTS.md

@@ -88,6 +88,13 @@
 - PR titles must follow Conventional Commits. Descriptions must be no longer than 2,000 characters and contain non-empty `## Why`, `## Changes`, and `## Verification` sections in that order.
 - Before creating or editing a PR, validate its metadata with `make check-pr-metadata PR_TITLE='type: summary' PR_BODY_FILE=/path/to/body.md`.
 - Monitor CI after every push and follow it through completion. Address failures before considering delivery complete.
+- When asked to babysit or watch a pull request, keep exactly one active background CI watcher for
+  the surviving pull request until its latest commit reaches a terminal state. Do not stop at the
+  initial push, PR creation, or an intermediate green run superseded by a later push. If pull
+  requests are consolidated, stop obsolete watchers and retain one watcher for the combined PR.
+- Babysitting includes inspecting failures, applying and pushing required fixes, restarting the
+  watcher after each push, and reporting the final green state. It does not authorize merging when
+  the user has asked for the pull request to remain unmerged.
 - Push follow-up commits only for CI failures or correctness findings that could not reasonably have been found before the first push.
 - For deployable changes, continue through auto-merge, release workflow completion, package publication, and published-version verification. Do not stop at PR creation unless the user explicitly asks.
 

README.md

@@ -94,6 +94,12 @@ model SQL or warehouse defaults. Required columns declared with `nullable false`
 unknown types must be supplied explicitly. When the relation's complete column set is authoritative,
 misspelled or unknown supplied fixture columns are rejected.
 
+A direct `NULL AS column_name` projection receives the authoritative relation type in compiled test
+SQL, including expected-output CTEs for contracted models. To represent a contracted upstream with
+no rows, use `SELECT * FROM __empty_fixture()` as the complete body of a `__ref__`, `__source__`, or
+`__seed__` fixture CTE, or a contracted `__expected__` CTE. SQLBuild expands it to the relation's
+full typed schema with a false filter.
+
 See the [documentation](https://docs.sqlbuild.com) for incremental models, scenarios, loaders, and more.
 
 ### Python project layout

src/sqlbuild/cli/commands/main/project/_format.py

@@ -6,13 +6,18 @@
 from dataclasses import replace
 from pathlib import Path
 
+from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter
 from sqlbuild.cli.commands._helpers.lint.runs import (
     prepare_lint_run,
     render_lint_result,
     render_lint_result_json,
 )
 from sqlbuild.cli.commands._helpers.lint.selection import resolve_lint_inputs
-from sqlbuild.compiler.compile.types import TypedSqlValueRenderer
+from sqlbuild.compiler.pipeline.main.graph import build_project_graph
+from sqlbuild.compiler.pipeline.models import ProjectGraph
+from sqlbuild.lint.main.has_fixture_typed_null_candidates import (
+    has_fixture_typed_null_candidates,
+)
 from sqlbuild.lint.main.run_format import run_format
 from sqlbuild.lint.models import LintConfig, LintRunResult
 from sqlbuild.presentation.main.supports_color import supports_color
@@ -34,24 +39,43 @@ def run_format_command(
     prepared: tuple[LintConfig, str | None] = prepare_lint_run(project_dir=base_dir)
     if prepared[1] is not None:
         print(f"WARN  {prepared[1]}")
-    value_renderer: TypedSqlValueRenderer | None = None
-    selected_paths: frozenset[Path] | None = None
     config: LintConfig = prepared[0]
-    if select or exclude:
-        value_renderer, selected_paths, _discovered_inputs = resolve_lint_inputs(
-            project_dir=base_dir,
-            select=select,
-            exclude=exclude,
-        )
-        config = replace(
-            config,
-            dialect=value_renderer.sql_analysis_dialect_name or "generic",
-        )
+    value_renderer: BaseAdapter | None = None
+    selected_paths: frozenset[Path] | None = None
+    graph: ProjectGraph | None = None
+    has_fixture_candidates: bool = has_fixture_typed_null_candidates(project_dir=base_dir)
+    if select or exclude or has_fixture_candidates:
+        try:
+            value_renderer, selected_paths, discovered_inputs = resolve_lint_inputs(
+                project_dir=base_dir,
+                select=select,
+                exclude=exclude,
+            )
+        except Exception:
+            if select or exclude:
+                raise
+        else:
+            config = replace(
+                config,
+                dialect=value_renderer.sql_analysis_dialect_name or "generic",
+            )
+            if has_fixture_candidates and selected_paths is None:
+                try:
+                    graph = build_project_graph(
+                        discovered_inputs=discovered_inputs,
+                        adapter=value_renderer,
+                    )
+                except Exception:
+                    graph = None
     result: LintRunResult = run_format(
         project_dir=base_dir,
         config=config,
         value_renderer=value_renderer,
         selected_paths=selected_paths,
+        compiled_project=(
+            graph.project if graph is not None and graph.project.settings.sql_analysis else None
+        ),
+        adapter=value_renderer,
         write=not (check or diff),
     )
     if diff:

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

@@ -13,6 +13,7 @@
 )
 from sqlbuild.compiler.compile._helpers.refs.references import extract_sql_references
 from sqlbuild.compiler.compile._helpers.render.macros import find_macro_call_names
+from sqlbuild.compiler.compile._helpers.sql_tests.empty_fixture import is_empty_fixture_query
 from sqlbuild.compiler.compile.constants import (
     ASSERT_TEST_CTE_PREFIX,
     DBT_REF_TEST_CTE_PREFIX,
@@ -489,7 +490,11 @@ def _classify_model_sql_test_ctes(
                     file_label=file_label,
                 )
             )
-            _validate_expected_cte_query(cte=cte, file_label=file_label)
+            _validate_expected_cte_query(
+                cte=cte,
+                file_label=file_label,
+                allow_empty_fixture=True,
+            )
             expected_ctes.append(cte)
             continue
         if cte.name.startswith(ASSERT_TEST_CTE_PREFIX):
@@ -736,8 +741,14 @@ def _read_sql_string_literal(*, sql: str, start: int) -> tuple[str, int]:
 
 
 def _validate_expected_cte_query(
-    *, cte: CompileSqlTestCte, file_label: str, label: str = "__expected__<model>"
+    *,
+    cte: CompileSqlTestCte,
+    file_label: str,
+    label: str = "__expected__<model>",
+    allow_empty_fixture: bool = False,
 ) -> None:
+    if allow_empty_fixture and is_empty_fixture_query(cte.sql_body):
+        return
     if _contains_select_star(cte.sql_body):
         raise CompileInputError(f"SQL test '{file_label}' must not use SELECT * in {label} CTEs")
     branch_column_names: tuple[tuple[str, ...], ...] = _extract_expected_branch_column_names(

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

@@ -0,0 +1,16 @@
+"""Recognition for the first-class empty relation fixture marker."""
+
+from __future__ import annotations
+
+import re
+
+_EMPTY_FIXTURE_PATTERN: re.Pattern[str] = re.compile(
+    r"^\s*SELECT\s+\*\s+FROM\s+__empty_fixture\s*\(\s*\)\s*$",
+    re.IGNORECASE,
+)
+
+
+def is_empty_fixture_query(sql: str) -> bool:
+    """Return whether SQL is exactly the empty relation fixture marker."""
+
+    return _EMPTY_FIXTURE_PATTERN.fullmatch(sql) is not None

src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py

@@ -25,8 +25,8 @@
 from sqlbuild.compiler.lineage.types import (
     ColumnLineageConfidence,
     ColumnTransformKind,
-    InferredNullability,
 )
+from sqlbuild.compiler.planner._helpers.fixtures.empty_fixture import is_empty_fixture_query
 from sqlbuild.compiler.planner.models import (
     FixtureColumnMetadata,
     FixtureRelationMetadata,
@@ -57,24 +57,24 @@ def build_relation_fixture_completion(
         project=project
     )
     model_map: dict[str, CompiledModel] = context.models_by_name
-    fixture_sql_by_key: dict[FixtureKey, str] = {}
-    inferred_by_fixture: dict[FixtureKey, tuple[InferredColumn, ...]] = {}
-    null_literal_names_by_fixture: dict[FixtureKey, frozenset[str]] = {}
-    quoted_names_by_fixture: dict[FixtureKey, frozenset[str]] = {}
-    for resource_type, fixtures in fixture_groups:
-        for name, sql in fixtures.items():
-            key: FixtureKey = (resource_type, name)
-            fixture_sql_by_key[key] = sql
-            inference: FixtureColumnInference | None = infer_fixture_column_facts(
-                query_sql=sql,
-                inference_profile=adapter.expression_inference_profile(),
-            )
-            if inference is not None:
-                inferred_by_fixture[key] = inference.columns
-                null_literal_names_by_fixture[key] = inference.null_literal_names
-                quoted_names_by_fixture[key] = inference.quoted_names
-
     relations: dict[FixtureKey, FixtureRelationMetadata] = context.relations
+    fixture_sql_by_key: dict[FixtureKey, str]
+    inferred_by_fixture: dict[FixtureKey, tuple[InferredColumn, ...]]
+    null_literal_names_by_fixture: dict[FixtureKey, frozenset[str]]
+    quoted_names_by_fixture: dict[FixtureKey, frozenset[str]]
+    diagnostics: list[RelationFixtureDiagnostic]
+    (
+        fixture_sql_by_key,
+        inferred_by_fixture,
+        null_literal_names_by_fixture,
+        quoted_names_by_fixture,
+        diagnostics,
+    ) = _prepare_fixture_inputs(
+        fixture_groups=fixture_groups,
+        relations=relations,
+        adapter=adapter,
+    )
+
     fixture_keys_requiring_analysis: frozenset[FixtureKey] = _fixture_keys_requiring_analysis(
         inferred_by_fixture=inferred_by_fixture,
         authoritative_columns=context.authoritative_columns,
@@ -98,10 +98,9 @@ def build_relation_fixture_completion(
         null_literal_names_by_fixture=null_literal_names_by_fixture,
         expected_types=expected_types,
     )
-    diagnostics: list[RelationFixtureDiagnostic] = list(
+    diagnostics.extend(
         _unknown_fixture_column_diagnostics(
-            inferred_by_fixture=inferred_by_fixture,
-            relations=relations,
+            inferred_by_fixture=inferred_by_fixture, relations=relations
         )
     )
 
@@ -212,6 +211,116 @@ def build_relation_fixture_completion(
     )
 
 
+def _prepare_fixture_inputs(
+    *,
+    fixture_groups: FixtureGroups,
+    relations: dict[FixtureKey, FixtureRelationMetadata],
+    adapter: BaseAdapter,
+) -> tuple[
+    dict[FixtureKey, str],
+    dict[FixtureKey, tuple[InferredColumn, ...]],
+    dict[FixtureKey, frozenset[str]],
+    dict[FixtureKey, frozenset[str]],
+    list[RelationFixtureDiagnostic],
+]:
+    fixture_sql_by_key: dict[FixtureKey, str] = {}
+    inferred_by_fixture: dict[FixtureKey, tuple[InferredColumn, ...]] = {}
+    null_literal_names_by_fixture: dict[FixtureKey, frozenset[str]] = {}
+    quoted_names_by_fixture: dict[FixtureKey, frozenset[str]] = {}
+    diagnostics: list[RelationFixtureDiagnostic] = []
+    for resource_type, fixtures in fixture_groups:
+        for name, sql in fixtures.items():
+            key: FixtureKey = (resource_type, name)
+            fixture_sql: str = sql
+            if is_empty_fixture_query(fixture_sql):
+                empty_sql, empty_error = complete_empty_fixture_sql(
+                    sql=fixture_sql,
+                    relation=relations.get(key),
+                    adapter=adapter,
+                )
+                if empty_error is not None:
+                    diagnostics.append(
+                        RelationFixtureDiagnostic(
+                            key=key,
+                            message=f"mock {key[0].value} '{key[1]}' {empty_error}",
+                        )
+                    )
+                    fixture_sql_by_key[key] = fixture_sql
+                    continue
+                fixture_sql = empty_sql
+            fixture_sql_by_key[key] = fixture_sql
+            inference: FixtureColumnInference | None = infer_fixture_column_facts(
+                query_sql=fixture_sql,
+                inference_profile=adapter.expression_inference_profile(),
+            )
+            if inference is not None:
+                inferred_by_fixture[key] = inference.columns
+                null_literal_names_by_fixture[key] = inference.null_literal_names
+                quoted_names_by_fixture[key] = inference.quoted_names
+    return (
+        fixture_sql_by_key,
+        inferred_by_fixture,
+        null_literal_names_by_fixture,
+        quoted_names_by_fixture,
+        diagnostics,
+    )
+
+
+def complete_typed_null_columns(
+    *,
+    sql: str,
+    inference: FixtureColumnInference,
+    expected_types: dict[str, str],
+    adapter: BaseAdapter,
+) -> str:
+    """Apply authoritative types to direct bare-NULL projections."""
+
+    typed_nulls: dict[str, str] = {
+        name: expected_types[name]
+        for name in inference.null_literal_names
+        if name in expected_types
+    }
+    if not typed_nulls:
+        return sql
+    return _completed_fixture_sql(
+        sql=sql,
+        inferred=inference.columns,
+        typed_nulls=typed_nulls,
+        quoted_names=inference.quoted_names,
+        completed=(),
+        adapter=adapter,
+    )
+
+
+def complete_empty_fixture_sql(
+    *,
+    sql: str,
+    relation: FixtureRelationMetadata | None,
+    adapter: BaseAdapter,
+) -> tuple[str, str | None]:
+    """Expand an empty-fixture marker from authoritative relation metadata."""
+
+    if not is_empty_fixture_query(sql):
+        return sql, None
+    if relation is None or not relation.authoritative_names:
+        return sql, "uses __empty_fixture() but its column schema is not authoritative"
+    unknown_types: tuple[str, ...] = tuple(
+        column.name for column in relation.columns if column.type is None
+    )
+    if unknown_types:
+        return (
+            sql,
+            f"uses __empty_fixture() but columns have unknown types: {', '.join(unknown_types)}",
+        )
+    if not relation.columns:
+        return sql, "uses __empty_fixture() but its schema has no columns"
+    projections: str = ",\n  ".join(
+        f"CAST(NULL AS {column.type}) AS {adapter.render_identifier(column.name)}"
+        for column in relation.columns
+    )
+    return f"SELECT\n  {projections}\nWHERE FALSE", None
+
+
 def _typed_null_fixture_types(
     *,
     null_literal_names_by_fixture: dict[FixtureKey, frozenset[str]],
@@ -520,7 +629,7 @@ def _model_fixture_metadata(*, model: CompiledModel) -> FixtureRelationMetadata:
         columns_by_name[key] = FixtureColumnMetadata(
             name=column.name,
             type=column.type,
-            nullable=_inferred_nullable(column.nullability),
+            nullable=None,
         )
     for column in model.schema_entry.columns if model.schema_entry is not None else ():
         key = column.name.casefold()
@@ -621,14 +730,6 @@ def _unknown_fixture_column_diagnostics(
     return tuple(diagnostics)
 
 
-def _inferred_nullable(value: InferredNullability) -> bool | None:
-    if value == InferredNullability.NULLABLE:
-        return True
-    if value == InferredNullability.NON_NULL:
-        return False
-    return None
-
-
 def _models_reading_columns(
     *, column_names: tuple[str, ...], required_columns: dict[str, set[str]]
 ) -> tuple[str, ...]:

src/sqlbuild/compiler/planner/_helpers/fixtures/empty_fixture.py

@@ -0,0 +1,14 @@
+"""Recognition for the planner's empty relation fixture marker."""
+
+import re
+
+_EMPTY_FIXTURE_PATTERN: re.Pattern[str] = re.compile(
+    r"^\s*SELECT\s+\*\s+FROM\s+__empty_fixture\s*\(\s*\)\s*$",
+    re.IGNORECASE,
+)
+
+
+def is_empty_fixture_query(sql: str) -> bool:
+    """Return whether SQL is exactly the empty relation fixture marker."""
+
+    return _EMPTY_FIXTURE_PATTERN.fullmatch(sql) is not None

src/sqlbuild/compiler/planner/_helpers/sql_tests/assembly.py

@@ -240,7 +240,7 @@ def plan_test(
             mock_ref_names=frozenset(mock_refs),
         )
     if sql_analysis_enabled and validate_fixtures:
-        mock_refs, mock_sources, mock_seeds = build_validated_test_fixtures(
+        mock_refs, mock_sources, mock_seeds, expected_map = build_validated_test_fixtures(
             test=test,
             project=project,
             adapter=adapter,