diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml
index 25b42435f0..45428b5cff 100644
--- a/.github/workflows/verify.yml
+++ b/.github/workflows/verify.yml
@@ -19,6 +19,7 @@ jobs:
       duckdb: ${{ steps.scope.outputs.duckdb }}
       postgres: ${{ steps.scope.outputs.postgres }}
       performance: ${{ steps.scope.outputs.performance }}
+      format_performance: ${{ steps.scope.outputs.format_performance }}
       rules_performance: ${{ steps.scope.outputs.rules_performance }}
     steps:
       - uses: actions/checkout@v4
@@ -36,18 +37,20 @@ jobs:
               echo "duckdb=$1"
               echo "postgres=$2"
               echo "performance=$3"
-              echo "rules_performance=$4"
+              echo "format_performance=$4"
+              echo "rules_performance=$5"
             } >> "${GITHUB_OUTPUT}"
           }
 
           if [[ "${EVENT_NAME}" != "pull_request" ]]; then
-            write_scope true true true true
+            write_scope true true true true true
             exit 0
           fi
 
           duckdb=false
           postgres=false
           performance=false
+          format_performance=false
           rules_performance=false
           broad=false
           changed=false
@@ -70,8 +73,13 @@ jobs:
                 ;;
               src/sqlbuild/rule_engine/*|src/sqlbuild/lint/*|scripts/rules_benchmark/*|scripts/benchmark_rules_ci.py|tests/performance/*)
                 broad=true
+                format_performance=true
                 rules_performance=true
                 ;;
+              tests/e2e/src/sqlbuild/cli/commands/main/format/*)
+                duckdb=true
+                format_performance=true
+                ;;
               tests/e2e/src/sqlbuild/cli/commands/main/postgres/*|tests/integration/src/sqlbuild/adapters/postgres/*|tests/integration/src/sqlbuild/virtual/state/classes/postgres/*|tests/integration/src/sqlbuild/runtime/execution_history/*)
                 postgres=true
                 ;;
@@ -82,6 +90,7 @@ jobs:
                 ;;
               .github/workflows/verify.yml|Makefile)
                 performance=true
+                format_performance=true
                 rules_performance=true
                 ;;
               .github/*|docs/*|*.md|.gitignore|LICENSE)
@@ -93,15 +102,21 @@ jobs:
           done < <(git diff --name-only -z "${BASE_SHA}" "${HEAD_SHA}")
 
           if [[ "${changed}" != "true" ]]; then
-            write_scope true true true true
+            write_scope true true true true true
             exit 0
           fi
 
           if [[ "${broad}" == "true" ]]; then
             duckdb=true
             postgres=true
+            format_performance=true
           fi
-          write_scope "${duckdb}" "${postgres}" "${performance}" "${rules_performance}"
+          write_scope \
+            "${duckdb}" \
+            "${postgres}" \
+            "${performance}" \
+            "${format_performance}" \
+            "${rules_performance}"
 
   lockfile:
     name: Lockfile consistency
@@ -207,6 +222,34 @@ jobs:
       - name: Run DuckDB E2E shard
         run: make ${{ matrix.target }}
 
+  format_performance:
+    name: Formatter performance guards
+    needs: [changes, lockfile]
+    if: needs.changes.outputs.format_performance == 'true'
+    runs-on: ubuntu-24.04
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v4
+      - uses: astral-sh/setup-uv@v5
+        with:
+          version: "0.12.3"
+          enable-cache: true
+          cache-dependency-glob: |
+            uv.lock
+            Cargo.lock
+      - uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: stable
+      - name: Install Python
+        run: uv python install 3.12
+      - name: Install dependencies
+        run: uv sync --locked --all-groups --all-extras --python 3.12
+      - name: Run formatter performance guards
+        run: >-
+          uv run pytest
+          tests/e2e/src/sqlbuild/cli/commands/main/format/test_format_performance.py
+          -n auto --dist loadfile -m performance
+
   performance:
     name: Performance guards
     needs: [changes, lockfile]
@@ -371,6 +414,7 @@ jobs:
         lockfile,
         checks,
         e2e_shards,
+        format_performance,
         performance,
         cold_compile_performance,
         rules_performance,
@@ -384,6 +428,7 @@ jobs:
           LOCKFILE_RESULT: ${{ needs.lockfile.result }}
           CHECKS_RESULT: ${{ needs.checks.result }}
           E2E_SHARDS_RESULT: ${{ needs.e2e_shards.result }}
+          FORMAT_PERFORMANCE_RESULT: ${{ needs.format_performance.result }}
           PERFORMANCE_RESULT: ${{ needs.performance.result }}
           COLD_COMPILE_PERFORMANCE_RESULT: ${{ needs.cold_compile_performance.result }}
           RULES_PERFORMANCE_RESULT: ${{ needs.rules_performance.result }}
@@ -397,6 +442,7 @@ jobs:
           fi
           for result in \
             "${E2E_SHARDS_RESULT}" \
+            "${FORMAT_PERFORMANCE_RESULT}" \
             "${PERFORMANCE_RESULT}" \
             "${COLD_COMPILE_PERFORMANCE_RESULT}" \
             "${RULES_PERFORMANCE_RESULT}" \
diff --git a/crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs b/crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs
index 63bddc0e49..e5761efd93 100644
--- a/crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs
+++ b/crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs
@@ -37,6 +37,12 @@ fn format_sql_json(py: Python<'_>, request_json: &str) -> PyResult<String> {
         .map_err(value_error)
 }
 
+#[pyfunction]
+fn format_sql_batch_json(py: Python<'_>, request_json: &str) -> PyResult<String> {
+    py.detach(|| crate::sql_lint::main::batch_formatter::format_batch_json(request_json))
+        .map_err(value_error)
+}
+
 #[pyfunction(name = "validate_sql_with_schema_json")]
 fn schema_validation_json(py: Python<'_>, request_json: &str) -> PyResult<String> {
     py.detach(|| crate::semantic_validation::main::validation_json(request_json))
@@ -255,6 +261,7 @@ pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
     module.add_function(wrap_pyfunction!(evaluate_json, module)?)?;
     module.add_function(wrap_pyfunction!(lint_sql_json, module)?)?;
     module.add_function(wrap_pyfunction!(format_sql_json, module)?)?;
+    module.add_function(wrap_pyfunction!(format_sql_batch_json, module)?)?;
     module.add_function(wrap_pyfunction!(schema_validation_json, module)?)?;
     module.add_function(wrap_pyfunction!(schema_validations_json, module)?)?;
     module.add_function(wrap_pyfunction!(analyze_sql_uses_json, module)?)?;
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs
index 0dd1bfbe44..bf432bf0a1 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs
@@ -2,6 +2,7 @@ use std::str::FromStr;
 
 use polyglot_sql::tokens::Token;
 use polyglot_sql::{Dialect, DialectType, format_by_name};
+use rayon::iter::{IntoParallelIterator, ParallelIterator};
 
 use crate::sql_lint::constants::{
     CAST_TYPE_SEPARATOR_KEYWORD, CLOSE_PARENTHESIS, LINT_API_VERSION, OPEN_PARENTHESIS,
@@ -19,6 +20,14 @@ pub(crate) fn format_json_impl(request_json: &str) -> Result<String, String> {
     serde_json::to_string(&response).map_err(|error| error.to_string())
 }
 
+pub(crate) fn format_batch_json_impl(request_json: &str) -> Result<String, String> {
+    let requests: Vec<FormatRequest> =
+        serde_json::from_str(request_json).map_err(|error| error.to_string())?;
+    let responses: Result<Vec<FormatResponse>, String> =
+        requests.into_par_iter().map(format_sql).collect();
+    serde_json::to_string(&responses?).map_err(|error| error.to_string())
+}
+
 fn format_sql(request: FormatRequest) -> Result<FormatResponse, String> {
     if request.version != LINT_API_VERSION {
         return Err(format!(
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/main/batch_formatter.rs b/crates/sqlbuild-rules-native/src/sql_lint/main/batch_formatter.rs
new file mode 100644
index 0000000000..d278cd430e
--- /dev/null
+++ b/crates/sqlbuild-rules-native/src/sql_lint/main/batch_formatter.rs
@@ -0,0 +1,5 @@
+use crate::sql_lint::_helpers::formatter::format_batch_json_impl;
+
+pub(crate) fn format_batch_json(request_json: &str) -> Result<String, String> {
+    format_batch_json_impl(request_json)
+}
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/main/mod.rs b/crates/sqlbuild-rules-native/src/sql_lint/main/mod.rs
index 62572fb63b..f4a89d1bea 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/main/mod.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/main/mod.rs
@@ -1,3 +1,4 @@
+pub(crate) mod batch_formatter;
 pub(crate) mod catalogue;
 pub(crate) mod engine;
 pub(crate) mod formatter;
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/tests/test_sql_lint.rs b/crates/sqlbuild-rules-native/src/sql_lint/tests/test_sql_lint.rs
index 5e4c0d0ed0..d8b4dddba5 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/tests/test_sql_lint.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/tests/test_sql_lint.rs
@@ -1,5 +1,6 @@
 use serde_json::{Value, json};
 
+use crate::sql_lint::main::batch_formatter::format_batch_json;
 use crate::sql_lint::main::engine::lint_json;
 use crate::sql_lint::main::formatter::format_json;
 use crate::sql_lint::tests::{helpers, test_types};
@@ -1913,6 +1914,49 @@ fn given_format_cases_when_formatting_then_output_matches() -> Result<(), String
     Ok(())
 }
 
+#[test]
+fn given_ordered_sql_batch_when_formatting_then_outputs_are_parallel_and_ordered()
+-> Result<(), String> {
+    let test_cases = [
+        test_types::FormatTestCase {
+            description: "first order query",
+            sql: "select order_id from orders where order_id=1",
+            expected_sql: "SELECT\n  order_id\nFROM orders\nWHERE\n  order_id = 1",
+            expected_changed: true,
+        },
+        test_types::FormatTestCase {
+            description: "second product query",
+            sql: "select product_id from products where product_id=2",
+            expected_sql: "SELECT\n  product_id\nFROM products\nWHERE\n  product_id = 2",
+            expected_changed: true,
+        },
+    ];
+    let requests: Vec<Value> = test_cases
+        .iter()
+        .map(|test_case| json!({"version": 1, "sql": test_case.sql, "dialect": "snowflake"}))
+        .collect();
+
+    let response =
+        format_batch_json(&serde_json::to_string(&requests).map_err(|error| error.to_string())?)?;
+    let payloads: Vec<Value> =
+        serde_json::from_str(&response).map_err(|error| error.to_string())?;
+
+    assert_eq!(payloads.len(), test_cases.len());
+    let mut payload_iter = payloads.iter();
+    for test_case in &test_cases {
+        let payload = payload_iter
+            .next()
+            .ok_or_else(|| format!("{}: batch response omitted an item", test_case.description))?;
+        assert_eq!(
+            payload["sql"], test_case.expected_sql,
+            "{}",
+            test_case.description
+        );
+        assert_eq!(payload["changed"], test_case.expected_changed);
+    }
+    Ok(())
+}
+
 #[test]
 fn given_unknown_rule_when_linting_then_request_is_rejected() -> Result<(), String> {
     let test_cases = [test_types::InvalidRuleTestCase {
diff --git a/src/sqlbuild/_native.pyi b/src/sqlbuild/_native.pyi
index 263bf2dcbe..09f29bf3c1 100644
--- a/src/sqlbuild/_native.pyi
+++ b/src/sqlbuild/_native.pyi
@@ -3,6 +3,7 @@
 def catalogue_json() -> str: ...
 def evaluate_json(request_json: str) -> str: ...
 def format_sql_json(request_json: str) -> str: ...
+def format_sql_batch_json(request_json: str) -> str: ...
 def lint_sql_json(request_json: str) -> str: ...
 def load_config_json(project_dir: str) -> str: ...
 def selected_codes_json(request_json: str) -> str: ...
diff --git a/src/sqlbuild/cli/commands/_helpers/entry/dispatch.py b/src/sqlbuild/cli/commands/_helpers/entry/dispatch.py
index bad5b857b9..51afc3d890 100644
--- a/src/sqlbuild/cli/commands/_helpers/entry/dispatch.py
+++ b/src/sqlbuild/cli/commands/_helpers/entry/dispatch.py
@@ -615,6 +615,7 @@ def _dispatch_lint_format_command(
         exclude=tuple(args.exclude),
         check=args.format_check,
         diff=args.format_diff,
+        fixtures_only=args.format_fixtures_only,
         json_output=args.json,
         no_color=args.no_color,
     )
diff --git a/src/sqlbuild/cli/commands/_helpers/entry/lazy_handlers.py b/src/sqlbuild/cli/commands/_helpers/entry/lazy_handlers.py
index 3c61b5e1bb..011924dc14 100644
--- a/src/sqlbuild/cli/commands/_helpers/entry/lazy_handlers.py
+++ b/src/sqlbuild/cli/commands/_helpers/entry/lazy_handlers.py
@@ -247,16 +247,17 @@ def build_lazy_cli_handlers() -> CliEntrypointHandlers:
             targets=targets,
             force=force,
         ),
-        run_format=lambda project_dir, select, exclude, check, diff, json_output, no_color: lazy[
-            "format"
-        ](
-            project_dir=project_dir,
-            select=select,
-            exclude=exclude,
-            check=check,
-            diff=diff,
-            json_output=json_output,
-            no_color=no_color,
+        run_format=lambda project_dir, select, exclude, check, diff, fixtures_only, json_output, no_color: (  # noqa: E501
+            lazy["format"](
+                project_dir=project_dir,
+                select=select,
+                exclude=exclude,
+                check=check,
+                diff=diff,
+                fixtures_only=fixtures_only,
+                json_output=json_output,
+                no_color=no_color,
+            )
         ),
         run_scenario=lazy["scenario"],
         run_scenario_capture=lazy["scenario_capture"],
diff --git a/src/sqlbuild/cli/commands/_helpers/entry/parser.py b/src/sqlbuild/cli/commands/_helpers/entry/parser.py
index 494781ac30..f46a51f1a4 100644
--- a/src/sqlbuild/cli/commands/_helpers/entry/parser.py
+++ b/src/sqlbuild/cli/commands/_helpers/entry/parser.py
@@ -309,6 +309,12 @@ def _add_quality_parsers(
     format_parser.add_argument("--json", action="store_true", default=False)
     format_parser.add_argument("--check", dest="format_check", action="store_true")
     format_parser.add_argument("--diff", dest="format_diff", action="store_true")
+    format_parser.add_argument(
+        "--fixtures-only",
+        dest="format_fixtures_only",
+        action="store_true",
+        help="only remove redundant typed-null SQL test fixture columns",
+    )
     _ = add_select_args(format_parser)
 
 
diff --git a/src/sqlbuild/cli/commands/_helpers/lint/selection.py b/src/sqlbuild/cli/commands/_helpers/lint/selection.py
index c08d04ce7a..8c57ba6923 100644
--- a/src/sqlbuild/cli/commands/_helpers/lint/selection.py
+++ b/src/sqlbuild/cli/commands/_helpers/lint/selection.py
@@ -8,12 +8,20 @@
 from sqlbuild.cli.commands._helpers.runtime.adapters import resolve_adapter
 from sqlbuild.compiler.compile.models import CompiledObjectKey
 from sqlbuild.compiler.compile.types import CompiledResourceType
+from sqlbuild.compiler.discovery.classes.selected_contract_input_discoverer import (
+    SelectedContractInputDiscoverer,
+)
 from sqlbuild.compiler.discovery.main.discover import discover_project_inputs
 from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs, DiscoveredSqlModelFile
 from sqlbuild.compiler.pipeline.main.graph import build_project_graph
 from sqlbuild.compiler.pipeline.models import ProjectGraph
 from sqlbuild.compiler.planner.exceptions import PlannerInputError
 from sqlbuild.compiler.planner.main.selection.selection import resolve_project_selectors
+from sqlbuild.lint.constants import LINT_DIRECTORY_NAMES, PARENT_PATH_SEGMENT, SQL_FILE_SUFFIX
+from sqlbuild.lint.main.scan_fixture_typed_null_candidates import (
+    scan_fixture_typed_null_candidates,
+)
+from sqlbuild.lint.models import FixtureNullCandidateScan
 from sqlbuild.spec.contracts.main.resolve_effective_adapter_name import (
     resolve_effective_adapter_name,
 )
@@ -26,17 +34,35 @@ def resolve_lint_inputs(
 ) -> tuple[BaseAdapter, frozenset[Path] | None, DiscoveredProjectInputs]:
     """Resolve the adapter and optional model-file scope through canonical selectors."""
 
-    discovered: DiscoveredProjectInputs = discover_project_inputs(
+    lint_paths: frozenset[Path] | None = _resolve_lint_paths(
+        project_dir=project_dir,
+        select=select,
+        exclude=exclude,
+    )
+    if lint_paths is not None:
+        _validate_selected_paths(paths=lint_paths)
+        candidates: FixtureNullCandidateScan = scan_fixture_typed_null_candidates(
+            project_dir=project_dir,
+            selected_paths=lint_paths,
+        )
+        discovered: DiscoveredProjectInputs = SelectedContractInputDiscoverer.discover(
+            project_dir=project_dir,
+            selected_test_paths=candidates.paths,
+            referenced_model_names=candidates.model_names,
+        )
+        adapter: BaseAdapter = _resolve_discovered_adapter(
+            project_dir=project_dir,
+            discovered=discovered,
+        )
+        return adapter, lint_paths, discovered
+    discovered = discover_project_inputs(
         project_dir=project_dir,
         sql_analysis_enabled_override=False,
         extract_output_column_locations=False,
     )
-    adapter: BaseAdapter = resolve_adapter(
-        adapter_name=resolve_effective_adapter_name(
-            project_config=discovered.project_config,
-            local_config=discovered.local_config,
-        ),
+    adapter = _resolve_discovered_adapter(
         project_dir=project_dir,
+        discovered=discovered,
     )
     if not select and not exclude:
         return adapter, None, discovered
@@ -71,6 +97,84 @@ def resolve_lint_inputs(
     return adapter, paths, discovered
 
 
+def _resolve_discovered_adapter(
+    *, project_dir: Path, discovered: DiscoveredProjectInputs
+) -> BaseAdapter:
+    return resolve_adapter(
+        adapter_name=resolve_effective_adapter_name(
+            project_config=discovered.project_config,
+            local_config=discovered.local_config,
+        ),
+        project_dir=project_dir,
+    )
+
+
+def _resolve_lint_paths(
+    *,
+    project_dir: Path,
+    select: tuple[str, ...],
+    exclude: tuple[str, ...],
+) -> frozenset[Path] | None:
+    """Resolve simple path selectors across every formatter-owned SQL root."""
+
+    selected_prefixes: tuple[str, ...] | None = _lint_path_prefixes(raw_selectors=select)
+    excluded_prefixes: tuple[str, ...] | None = _lint_path_prefixes(raw_selectors=exclude)
+    if selected_prefixes is None or excluded_prefixes is None:
+        return None
+    selected: tuple[Path, ...] = _paths_for_prefixes(
+        project_dir=project_dir,
+        prefixes=selected_prefixes,
+    )
+    return frozenset(
+        path
+        for path in selected
+        if not _matches_path_prefix(
+            relative_path=path.relative_to(project_dir.resolve()).as_posix(),
+            prefixes=excluded_prefixes,
+        )
+    )
+
+
+def _paths_for_prefixes(*, project_dir: Path, prefixes: tuple[str, ...]) -> tuple[Path, ...]:
+    roots: tuple[Path, ...] = (
+        tuple(project_dir / prefix for prefix in prefixes)
+        if prefixes
+        else tuple(project_dir / directory_name for directory_name in LINT_DIRECTORY_NAMES)
+    )
+    paths: set[Path] = set()
+    for root in roots:
+        if root.is_file() and root.suffix.casefold() == SQL_FILE_SUFFIX:
+            paths.add(root.resolve())
+        elif root.is_dir():
+            paths.update(file_path.resolve() for file_path in root.rglob("*.sql"))
+    return tuple(sorted(paths))
+
+
+def _lint_path_prefixes(*, raw_selectors: tuple[str, ...]) -> tuple[str, ...] | None:
+    prefixes: list[str] = []
+    for raw_selector in raw_selectors:
+        if not raw_selector.strip():
+            return None
+        for token in raw_selector.split():
+            normalized: str = token.replace("\\", "/")
+            if not normalized.startswith("path:"):
+                return None
+            prefix: str = normalized.removeprefix("path:").strip("/")
+            if any(marker in prefix for marker in ("+", "~", ",")):
+                return None
+            parts: tuple[str, ...] = tuple(part for part in prefix.split("/") if part)
+            if not parts or parts[0] not in LINT_DIRECTORY_NAMES or PARENT_PATH_SEGMENT in parts:
+                return None
+            prefixes.append("/".join(parts))
+    return tuple(prefixes)
+
+
+def _matches_path_prefix(*, relative_path: str, prefixes: tuple[str, ...]) -> bool:
+    return any(
+        relative_path == prefix or relative_path.startswith(f"{prefix}/") for prefix in prefixes
+    )
+
+
 def _resolve_exact_model_paths(
     *,
     project_dir: Path,
@@ -124,10 +228,10 @@ def _exact_names(*, raw_selectors: tuple[str, ...]) -> frozenset[str] | None:
 
 
 def _validate_selected_paths(*, paths: frozenset[Path]) -> None:
-    """Reject lint selections that contain no SQL model files."""
+    """Reject lint selections that contain no SQL files."""
 
     if not paths:
         raise PlannerInputError(
-            "lint selection matched no SQL model files",
+            "lint selection matched no SQL files",
             code="S007",
         )
diff --git a/src/sqlbuild/cli/commands/classes/cli_namespace.py b/src/sqlbuild/cli/commands/classes/cli_namespace.py
index cd2257f7f4..40a9b59ca3 100644
--- a/src/sqlbuild/cli/commands/classes/cli_namespace.py
+++ b/src/sqlbuild/cli/commands/classes/cli_namespace.py
@@ -152,6 +152,7 @@
     "rules_skills_check": False,
     "format_check": False,
     "format_diff": False,
+    "format_fixtures_only": False,
     "scope_target": None,
     "scope_at": None,
     "scope_as_path": None,
@@ -324,6 +325,7 @@ class CliNamespace:
     rules_skills_check: bool
     format_check: bool
     format_diff: bool
+    format_fixtures_only: bool
     scope_target: str | None
     scope_at: str | None
     scope_as_path: str | None
diff --git a/src/sqlbuild/cli/commands/main/project/_format.py b/src/sqlbuild/cli/commands/main/project/_format.py
index 15df1bb9a8..f42641ef3e 100644
--- a/src/sqlbuild/cli/commands/main/project/_format.py
+++ b/src/sqlbuild/cli/commands/main/project/_format.py
@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import difflib
+import sys
+import time
 from dataclasses import replace
 from pathlib import Path
 
@@ -13,13 +15,14 @@
     render_lint_result_json,
 )
 from sqlbuild.cli.commands._helpers.lint.selection import resolve_lint_inputs
-from sqlbuild.compiler.pipeline.main.graph import build_project_graph
-from sqlbuild.compiler.pipeline.models import ProjectGraph
+from sqlbuild.compiler.discovery.constants import SQL_ANALYSIS_SETTING_KEY
+from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
 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.classes.transient_status_reporter import TransientStatusReporter
 from sqlbuild.presentation.main.supports_color import supports_color
 
 
@@ -30,6 +33,7 @@ def run_format_command(
     exclude: tuple[str, ...] = (),
     check: bool = False,
     diff: bool = False,
+    fixtures_only: bool = False,
     json_output: bool = False,
     no_color: bool = False,
 ) -> int:
@@ -42,8 +46,10 @@ def run_format_command(
     config: LintConfig = prepared[0]
     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)
+    discovered_inputs: DiscoveredProjectInputs | None = None
+    has_fixture_candidates: bool = False
+    if not select and not exclude:
+        has_fixture_candidates = 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(
@@ -55,28 +61,46 @@ def run_format_command(
             if select or exclude:
                 raise
         else:
+            has_fixture_candidates = bool(discovered_inputs.test_files)
             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),
+    fixture_inputs: DiscoveredProjectInputs | None = None
+    if discovered_inputs is not None and has_fixture_candidates:
+        local_overrides: frozenset[str] = discovered_inputs.local_config.setting_overrides
+        sql_analysis_enabled: bool = (
+            discovered_inputs.local_config.settings.sql_analysis
+            if SQL_ANALYSIS_SETTING_KEY in local_overrides
+            else discovered_inputs.project_config.settings.sql_analysis
+        )
+        if sql_analysis_enabled:
+            fixture_inputs = discovered_inputs
+    format_started_at: float = time.monotonic()
+    format_status: TransientStatusReporter = TransientStatusReporter(
+        stream=sys.stderr,
+        use_color=not no_color and supports_color(),
+    )
+    format_status.start("Formatting SQL  START")
+    try:
+        result: LintRunResult = run_format(
+            project_dir=base_dir,
+            config=config,
+            value_renderer=value_renderer,
+            selected_paths=selected_paths,
+            discovered_inputs=fixture_inputs,
+            fixtures_only=fixtures_only,
+            write=not (check or diff),
+        )
+    except Exception:
+        format_status.error("Formatting SQL  ERROR")
+        raise
+    elapsed_seconds: float = time.monotonic() - format_started_at
+    format_status.complete(
+        message=(
+            f"Formatting SQL  OK  ({elapsed_seconds:.2f}s; "
+            f"{result.files_checked} files checked, {len(result.formatted_files)} changed)"
+        )
     )
     if diff:
         _ = _render_format_diff(result=result)
diff --git a/src/sqlbuild/cli/commands/types.py b/src/sqlbuild/cli/commands/types.py
index d6f416bc2b..0f2d1036e4 100644
--- a/src/sqlbuild/cli/commands/types.py
+++ b/src/sqlbuild/cli/commands/types.py
@@ -159,6 +159,7 @@ def __call__(
         exclude: tuple[str, ...],
         check: bool,
         diff: bool,
+        fixtures_only: bool,
         json_output: bool,
         no_color: bool,
     ) -> int: ...
diff --git a/src/sqlbuild/compiler/compile/classes/__init__.py b/src/sqlbuild/compiler/compile/classes/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/src/sqlbuild/compiler/compile/classes/sql_test_cte_extractor.py b/src/sqlbuild/compiler/compile/classes/sql_test_cte_extractor.py
new file mode 100644
index 0000000000..53edda5aa0
--- /dev/null
+++ b/src/sqlbuild/compiler/compile/classes/sql_test_cte_extractor.py
@@ -0,0 +1,18 @@
+"""SQL-test CTE extraction class."""
+
+from __future__ import annotations
+
+from sqlbuild.compiler.compile._helpers.sql_tests.core import (
+    extract_unclassified_sql_test_ctes,
+)
+from sqlbuild.compiler.compile.models import CompileSqlTestCte
+
+
+class SqlTestCteExtractor:
+    """Expose authored SQL-test CTE parsing to compiler consumers."""
+
+    @staticmethod
+    def extract(*, sql: str, file_label: str) -> tuple[CompileSqlTestCte, ...]:
+        """Extract top-level SQL-test CTEs without classifying their roles."""
+
+        return extract_unclassified_sql_test_ctes(sql=sql, file_label=file_label)
diff --git a/src/sqlbuild/compiler/discovery/_helpers/filesystem/core.py b/src/sqlbuild/compiler/discovery/_helpers/filesystem/core.py
index 3425a7edf6..afc40de786 100644
--- a/src/sqlbuild/compiler/discovery/_helpers/filesystem/core.py
+++ b/src/sqlbuild/compiler/discovery/_helpers/filesystem/core.py
@@ -349,6 +349,7 @@ def discover_model_files(
     project_dir: Path,
     extract_implicit_alias_columns: bool = True,
     extract_output_column_locations: bool = True,
+    selected_model_names: frozenset[str] | None = None,
     on_fault: Callable[[DiscoveryFileFault], None] | None = None,
 ) -> tuple[DiscoveredSqlModelFile, ...]:
     """Discover SQL model files under models/."""
@@ -360,6 +361,8 @@ def discover_model_files(
     loaded_model_files: list[tuple[Path, str | None, Exception | None]] = []
     file_path: Path
     for file_path in sorted(model_root.rglob("*.sql")):
+        if selected_model_names is not None and file_path.stem not in selected_model_names:
+            continue
         if _is_in_scoped_declaration_tree(file_path=file_path, project_dir=project_dir):
             continue
         try:
@@ -722,7 +725,10 @@ def discover_seed_files(*, project_dir: Path) -> tuple[DiscoveredSeedFile, ...]:
 
 
 def discover_test_files(
-    *, project_dir: Path, on_fault: Callable[[DiscoveryFileFault], None] | None = None
+    *,
+    project_dir: Path,
+    selected_paths: frozenset[Path] | None = None,
+    on_fault: Callable[[DiscoveryFileFault], None] | None = None,
 ) -> tuple[DiscoveredSqlTestFile, ...]:
     """Discover SQL-native unit test files under tests/unit/."""
 
@@ -733,6 +739,8 @@ def discover_test_files(
     discovered_test_files: list[DiscoveredSqlTestFile] = []
     file_path: Path
     for file_path in sorted(tests_root.rglob("*.sql")):
+        if selected_paths is not None and file_path.resolve() not in selected_paths:
+            continue
         if _is_in_scoped_declaration_tree(file_path=file_path, project_dir=project_dir):
             continue
         try:
diff --git a/src/sqlbuild/compiler/discovery/classes/__init__.py b/src/sqlbuild/compiler/discovery/classes/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/src/sqlbuild/compiler/discovery/classes/selected_contract_input_discoverer.py b/src/sqlbuild/compiler/discovery/classes/selected_contract_input_discoverer.py
new file mode 100644
index 0000000000..0f5878b792
--- /dev/null
+++ b/src/sqlbuild/compiler/discovery/classes/selected_contract_input_discoverer.py
@@ -0,0 +1,50 @@
+"""Bounded discovery for selected contract consumers."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from sqlbuild.compiler.discovery._helpers.filesystem.core import (
+    discover_model_files,
+    discover_model_schema_files,
+    discover_schema_files,
+    discover_source_files,
+    discover_test_files,
+)
+from sqlbuild.compiler.discovery._helpers.yml.project import (
+    load_local_config,
+    load_project_config,
+)
+from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
+
+
+class SelectedContractInputDiscoverer:
+    """Discover contracts and selected SQL tests without unrelated runtime resources."""
+
+    @staticmethod
+    def discover(
+        *,
+        project_dir: Path,
+        selected_test_paths: frozenset[Path],
+        referenced_model_names: frozenset[str],
+    ) -> DiscoveredProjectInputs:
+        """Build selected contract inputs without importing Python runtime resources."""
+
+        return DiscoveredProjectInputs(
+            project_config=load_project_config(project_dir=project_dir),
+            local_config=load_local_config(project_dir=project_dir),
+            project_dir=project_dir,
+            model_files=discover_model_files(
+                project_dir=project_dir,
+                extract_implicit_alias_columns=False,
+                extract_output_column_locations=False,
+                selected_model_names=referenced_model_names,
+            ),
+            model_schema_files=discover_model_schema_files(project_dir=project_dir),
+            schema_files=discover_schema_files(project_dir=project_dir),
+            source_files=discover_source_files(project_dir=project_dir),
+            test_files=discover_test_files(
+                project_dir=project_dir,
+                selected_paths=selected_test_paths,
+            ),
+        )
diff --git a/src/sqlbuild/compiler/planner/_helpers/sql_tests/fixture_formatting.py b/src/sqlbuild/compiler/planner/_helpers/sql_tests/fixture_formatting.py
index f9a9cbccd1..4c72ed67ae 100644
--- a/src/sqlbuild/compiler/planner/_helpers/sql_tests/fixture_formatting.py
+++ b/src/sqlbuild/compiler/planner/_helpers/sql_tests/fixture_formatting.py
@@ -5,36 +5,29 @@
 import re
 from pathlib import Path
 
-from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter
+from sqlbuild.compiler.compile.classes.sql_test_cte_extractor import SqlTestCteExtractor
 from sqlbuild.compiler.compile.constants import (
     REF_TEST_CTE_PREFIX,
     SEED_TEST_CTE_PREFIX,
     SOURCE_TEST_CTE_PREFIX,
 )
-from sqlbuild.compiler.compile.main._infer_fixture_columns import infer_fixture_column_facts
-from sqlbuild.compiler.compile.models import (
-    CompiledModelSqlTestPayload,
-    CompiledProject,
-    CompiledSqlTest,
-    CompileSqlTestCte,
-    FixtureColumnInference,
-)
+from sqlbuild.compiler.compile.exceptions import CompileInputError
+from sqlbuild.compiler.compile.models import CompileSqlTestCte
 from sqlbuild.compiler.compile.types import CompiledResourceType, SqlTestMode
-from sqlbuild.compiler.planner._helpers.fixtures.completion import (
-    build_relation_fixture_completion,
-    build_relation_fixture_context,
-)
-from sqlbuild.compiler.planner._helpers.sql_tests.assembly import (
-    build_sql_test_planning_context,
+from sqlbuild.compiler.discovery.main._model_schema_columns import parse_schema_columns
+from sqlbuild.compiler.discovery.models import (
+    DiscoveredProjectInputs,
+    DiscoveredSqlModelFile,
+    DiscoveredSqlTestBlock,
+    DiscoveredSqlTestFile,
 )
+from sqlbuild.compiler.path_defaults.main._select import select_path_default
 from sqlbuild.compiler.planner.models import (
     FixtureColumnMetadata,
     FixtureRelationMetadata,
-    RelationFixtureCompletion,
-    RelationFixturePlanningContext,
-    SqlTestPlanningContext,
 )
-from sqlbuild.compiler.planner.types import FixtureGroups, FixtureKey
+from sqlbuild.compiler.planner.types import ContractPolicy, FixtureKey
+from sqlbuild.spec.contracts.models import SchemaColumn, SchemaSeedEntry, SourceEntry
 
 _TYPED_NULL_LINE_PATTERN: re.Pattern[str] = re.compile(
     r"^(?P<indent>[ \t]*)(?:"
@@ -54,123 +47,169 @@
 _SET_OPERATION_PATTERN: re.Pattern[str] = re.compile(
     r"\b(?:UNION|INTERSECT|EXCEPT)\b", re.IGNORECASE
 )
+_MACRO_INVOCATION_MARKER: str = "@"
+_TYPED_NULL_CANDIDATE_PATTERN: re.Pattern[str] = re.compile(
+    r"(?:CAST\s*\(\s*NULL\s+AS\b|NULL\s*::)", re.IGNORECASE
+)
+_PROJECTION_ALIAS_PATTERN: re.Pattern[str] = re.compile(
+    r"\bAS\s+(?P<alias>\"[^\"]+\"|`[^`]+`|[A-Za-z_][A-Za-z0-9_]*)\s*,?\s*$",
+    re.IGNORECASE,
+)
+_SELECT_LINE_PATTERN: re.Pattern[str] = re.compile(r"^\s*SELECT\b(?P<body>.*)$", re.IGNORECASE)
 
 
 def format_redundant_fixture_nulls(
     *,
     files: dict[Path, str],
     project_dir: Path,
-    project: CompiledProject,
-    adapter: BaseAdapter,
+    discovered_inputs: DiscoveredProjectInputs,
 ) -> dict[Path, str]:
     """Remove typed-null fixture projections that planner completion can reproduce."""
 
-    planning_context: SqlTestPlanningContext = build_sql_test_planning_context(
-        project=project,
-        tests=project.sql_tests,
-    )
-    fixture_context: RelationFixturePlanningContext = build_relation_fixture_context(
-        project=project
+    relations: dict[FixtureKey, FixtureRelationMetadata] = _fixture_relations(
+        discovered_inputs=discovered_inputs
     )
     updated: dict[Path, str] = {}
-    processed_blocks: set[tuple[Path, int]] = set()
-    for test in project.sql_tests:
-        source_path: Path | None = test.source_path
-        block_index: int | None = test.block_index
-        if source_path is None or block_index is None:
-            continue
-        block_key: tuple[Path, int] = (source_path, block_index)
-        if block_key in processed_blocks or test.mode is not SqlTestMode.MODEL:
+    test_file: DiscoveredSqlTestFile
+    for test_file in discovered_inputs.test_files:
+        file_path: Path = (project_dir / test_file.relative_path).resolve()
+        if file_path not in files:
             continue
-        processed_blocks.add(block_key)
-        file_path: Path = (project_dir / source_path).resolve()
         contents: str | None = updated.get(file_path, files.get(file_path))
         if contents is None:
             continue
-        fixed_block: str = _format_test_block(
-            test=test,
-            project=project,
-            adapter=adapter,
-            planning_context=planning_context,
-            fixture_context=fixture_context,
-        )
-        if fixed_block == test.test_block.sql_body:
-            continue
-        rewritten: str = _replace_test_block(
-            contents=contents,
-            original=test.test_block.sql_body,
-            replacement=fixed_block,
-        )
-        if rewritten != contents:
-            updated[file_path] = rewritten
+        block: DiscoveredSqlTestBlock
+        for block in test_file.blocks:
+            fixed_block: str = _format_test_block(
+                block=block,
+                file_label=str(test_file.relative_path),
+                relations=relations,
+            )
+            if fixed_block == block.sql_body:
+                continue
+            contents = _replace_test_block(
+                contents=contents,
+                original=block.sql_body,
+                replacement=fixed_block,
+            )
+            updated[file_path] = contents
     return updated
 
 
 def _format_test_block(
     *,
-    test: CompiledSqlTest,
-    project: CompiledProject,
-    adapter: BaseAdapter,
-    planning_context: SqlTestPlanningContext,
-    fixture_context: RelationFixturePlanningContext,
+    block: DiscoveredSqlTestBlock,
+    file_label: str,
+    relations: dict[FixtureKey, FixtureRelationMetadata],
 ) -> str:
-    payload: object = test.payload
-    if not isinstance(payload, CompiledModelSqlTestPayload):
-        return test.test_block.sql_body
-    if test.sql_body != test.test_block.sql_body:
-        return test.test_block.sql_body
-    ordered_model_names: tuple[str, ...] | None = planning_context.chain_names_by_test_key.get(
-        test.key
-    )
-    if ordered_model_names is None:
-        return test.test_block.sql_body
-    fixture_ctes: tuple[CompileSqlTestCte, ...] = tuple(
-        cte for cte in payload.authored_ctes if _fixture_key(cte.name) is not None
-    )
-    fixture_sql: dict[tuple[CompiledResourceType, str], str] = {
-        key: cte.sql_body for cte in fixture_ctes if (key := _fixture_key(cte.name)) is not None
-    }
+    if (
+        block.mode is not SqlTestMode.MODEL
+        or _MACRO_INVOCATION_MARKER in block.sql_body
+        or _TYPED_NULL_CANDIDATE_PATTERN.search(block.sql_body) is None
+    ):
+        return block.sql_body
+    try:
+        authored_ctes: tuple[CompileSqlTestCte, ...] = SqlTestCteExtractor.extract(
+            sql=block.sql_body,
+            file_label=file_label,
+        )
+    except CompileInputError:
+        return block.sql_body
     changed_bodies: list[tuple[str, str]] = []
-    for cte in fixture_ctes:
+    for cte in authored_ctes:
         key: FixtureKey | None = _fixture_key(cte.name)
         if key is None:
             continue
-        relation: FixtureRelationMetadata | None = fixture_context.relations.get(key)
+        relation: FixtureRelationMetadata | None = relations.get(key)
         if relation is None or not relation.authoritative_names:
             continue
         candidate: str = _remove_redundant_typed_null_lines(
-            sql=fixture_sql[key],
+            sql=cte.sql_body,
             relation=relation,
-            adapter=adapter,
         )
-        if candidate == fixture_sql[key]:
+        if candidate == cte.sql_body:
             continue
-        candidate_fixture_sql: dict[tuple[CompiledResourceType, str], str] = {
-            **fixture_sql,
-            key: candidate,
-        }
-        completion: RelationFixtureCompletion = build_relation_fixture_completion(
-            project=project,
-            adapter=adapter,
-            ordered_model_names=ordered_model_names,
-            fixture_groups=_fixture_groups(candidate_fixture_sql),
-            planning_context=fixture_context,
-        )
-        if completion.diagnostics:
-            continue
-        fixture_sql[key] = candidate
         changed_bodies.append((cte.sql_body, candidate))
     if not changed_bodies:
-        return test.test_block.sql_body
+        return block.sql_body
     return _replace_cte_bodies(
-        sql=test.test_block.sql_body,
+        sql=block.sql_body,
         replacements=tuple(changed_bodies),
     )
 
 
-def _remove_redundant_typed_null_lines(
-    *, sql: str, relation: FixtureRelationMetadata, adapter: BaseAdapter
-) -> str:
+def _fixture_relations(
+    *, discovered_inputs: DiscoveredProjectInputs
+) -> dict[FixtureKey, FixtureRelationMetadata]:
+    relations: dict[FixtureKey, FixtureRelationMetadata] = {}
+    default_contract: str | None = discovered_inputs.project_config.defaults.contract
+    path_defaults: dict[str, dict[str, object]] = discovered_inputs.project_config.path_defaults
+    model_file: DiscoveredSqlModelFile
+    for model_file in discovered_inputs.model_files:
+        selected_default: str | None = select_path_default(
+            model_path=str(model_file.relative_path),
+            path_keys=tuple(path_defaults),
+        ).selected_key
+        contract: object = default_contract
+        if selected_default is not None:
+            contract = path_defaults[selected_default].get("contract", contract)
+        contract = model_file.header_values.get("contract", contract)
+        if contract != ContractPolicy.ENFORCED.value:
+            continue
+        columns: tuple[SchemaColumn, ...] = parse_schema_columns(
+            raw_columns=model_file.header_values.get("columns"),
+            file_path=model_file.file_path,
+            label="model",
+            error_class=CompileInputError,
+            column_locations=model_file.header_column_locations,
+        )
+        if not columns:
+            continue
+        relations[(CompiledResourceType.MODEL, model_file.file_path.stem)] = _relation_metadata(
+            columns=columns
+        )
+    for source_file in discovered_inputs.source_files:
+        source: SourceEntry
+        for source in source_file.source_entries:
+            source_contract: str | None = source.contract or default_contract
+            if source_contract != ContractPolicy.ENFORCED.value:
+                continue
+            relations[(CompiledResourceType.SOURCE, source.name)] = FixtureRelationMetadata(
+                columns=tuple(
+                    FixtureColumnMetadata(
+                        name=column.name,
+                        type=column.type,
+                        nullable=column.nullable,
+                    )
+                    for column in source.columns
+                ),
+                authoritative_names=bool(source.columns),
+            )
+    for schema_file in discovered_inputs.schema_files:
+        seed: SchemaSeedEntry
+        for seed in schema_file.seed_entries:
+            if seed.columns:
+                relations[(CompiledResourceType.SEED, seed.name)] = _relation_metadata(
+                    columns=seed.columns
+                )
+    return relations
+
+
+def _relation_metadata(*, columns: tuple[SchemaColumn, ...]) -> FixtureRelationMetadata:
+    return FixtureRelationMetadata(
+        columns=tuple(
+            FixtureColumnMetadata(
+                name=column.name,
+                type=column.type,
+                nullable=column.nullable,
+            )
+            for column in columns
+        ),
+        authoritative_names=True,
+    )
+
+
+def _remove_redundant_typed_null_lines(*, sql: str, relation: FixtureRelationMetadata) -> str:
     if _SET_OPERATION_PATTERN.search(sql) is not None:
         return sql
     metadata_by_name: dict[str, FixtureColumnMetadata] = {
@@ -200,18 +239,54 @@ def _remove_redundant_typed_null_lines(
         return "SELECT * FROM __empty_fixture()"
     if not remove_indexes:
         return sql
+    authored_names: tuple[str, ...] | None = _projection_aliases(lines=lines)
+    if authored_names is None or not _names_form_schema_prefix(
+        names=authored_names,
+        relation=relation,
+    ):
+        return sql
     retained: list[str] = [line for index, line in enumerate(lines) if index not in remove_indexes]
     retained = _without_trailing_projection_comma(lines=retained)
     candidate: str = "\n".join(retained)
-    inference: FixtureColumnInference | None = infer_fixture_column_facts(
-        query_sql=candidate,
-        inference_profile=adapter.expression_inference_profile(),
-    )
-    if inference is None or not inference.columns:
+    retained_names: tuple[str, ...] | None = _projection_aliases(lines=retained)
+    if retained_names is None or not retained_names:
+        return sql
+    if not _names_form_schema_prefix(names=retained_names, relation=relation):
         return sql
     return candidate
 
 
+def _projection_aliases(*, lines: list[str]) -> tuple[str, ...] | None:
+    aliases: list[str] = []
+    saw_select: bool = False
+    for line in lines:
+        candidate: str = line
+        if not saw_select:
+            select_match: re.Match[str] | None = _SELECT_LINE_PATTERN.match(line)
+            if select_match is None:
+                if line.strip():
+                    return None
+                continue
+            saw_select = True
+            candidate = select_match.group("body")
+        if _CLAUSE_LINE_PATTERN.match(candidate):
+            break
+        if not candidate.strip():
+            continue
+        alias_match: re.Match[str] | None = _PROJECTION_ALIAS_PATTERN.search(candidate)
+        if alias_match is None:
+            return None
+        aliases.append(alias_match.group("alias").strip('"`').casefold())
+    return tuple(aliases) if saw_select else None
+
+
+def _names_form_schema_prefix(*, names: tuple[str, ...], relation: FixtureRelationMetadata) -> bool:
+    schema_prefix: tuple[str, ...] = tuple(
+        column.name.casefold() for column in relation.columns[: len(names)]
+    )
+    return names == schema_prefix
+
+
 def _without_trailing_projection_comma(*, lines: list[str]) -> list[str]:
     updated: list[str] = list(lines)
     clause_index: int | None = next(
@@ -238,24 +313,6 @@ def _fixture_key(name: str) -> tuple[CompiledResourceType, str] | None:
     return None
 
 
-def _fixture_groups(
-    fixture_sql: dict[tuple[CompiledResourceType, str], str],
-) -> FixtureGroups:
-    groups: list[tuple[CompiledResourceType, dict[str, str]]] = []
-    resource_types: tuple[CompiledResourceType, ...] = (
-        CompiledResourceType.MODEL,
-        CompiledResourceType.SOURCE,
-        CompiledResourceType.SEED,
-    )
-    for resource_type in resource_types:
-        fixtures: dict[str, str] = {}
-        for (candidate_type, name), sql in fixture_sql.items():
-            if candidate_type == resource_type:
-                fixtures[name] = sql
-        groups.append((resource_type, fixtures))
-    return tuple(groups)
-
-
 def _replace_cte_bodies(*, sql: str, replacements: tuple[tuple[str, str], ...]) -> str:
     result: str = sql
     for original, replacement in replacements:
diff --git a/src/sqlbuild/compiler/planner/classes/fixture_null_autofix.py b/src/sqlbuild/compiler/planner/classes/fixture_null_autofix.py
index 380124d2a8..97cf4178ec 100644
--- a/src/sqlbuild/compiler/planner/classes/fixture_null_autofix.py
+++ b/src/sqlbuild/compiler/planner/classes/fixture_null_autofix.py
@@ -4,8 +4,7 @@
 
 from pathlib import Path
 
-from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter
-from sqlbuild.compiler.compile.models import CompiledProject
+from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
 from sqlbuild.compiler.planner._helpers.sql_tests.fixture_formatting import (
     format_redundant_fixture_nulls,
 )
@@ -19,14 +18,12 @@ def apply(
         *,
         files: dict[Path, str],
         project_dir: Path,
-        project: CompiledProject,
-        adapter: BaseAdapter,
+        discovered_inputs: DiscoveredProjectInputs,
     ) -> dict[Path, str]:
         """Return contract-safe fixture source rewrites."""
 
         return format_redundant_fixture_nulls(
             files=files,
             project_dir=project_dir,
-            project=project,
-            adapter=adapter,
+            discovered_inputs=discovered_inputs,
         )
diff --git a/src/sqlbuild/lint/_helpers/fixture_null_candidates.py b/src/sqlbuild/lint/_helpers/fixture_null_candidates.py
index c1e4266c26..35c6cd7f0f 100644
--- a/src/sqlbuild/lint/_helpers/fixture_null_candidates.py
+++ b/src/sqlbuild/lint/_helpers/fixture_null_candidates.py
@@ -5,19 +5,41 @@
 import re
 from pathlib import Path
 
+from sqlbuild.lint.models import FixtureNullCandidateScan
+
 _FIXTURE_TYPED_NULL_PATTERN: re.Pattern[str] = re.compile(
     r"(?:CAST\s*\(\s*NULL\s+AS\b|NULL\s*::)", re.IGNORECASE
 )
 _FIXTURE_CTE_MARKERS: tuple[str, ...] = ("__ref__", "__source__", "__seed__")
 _IGNORED_SEARCH_DIRECTORIES: frozenset[str] = frozenset({".git", ".venv", "target", "node_modules"})
+_MODEL_FIXTURE_PATTERN: re.Pattern[str] = re.compile(
+    r"\b__ref__(?P<name>[A-Za-z_][A-Za-z0-9_]*)\s+AS\s*\(",
+    re.IGNORECASE,
+)
 
 
 def has_fixture_typed_null_candidates(*, project_dir: Path) -> bool:
     """Return whether source files may contain safe fixture-null autofixes."""
 
-    for file_path in project_dir.rglob("*.sql"):
+    return bool(scan_fixture_typed_null_candidates(project_dir=project_dir).paths)
+
+
+def scan_fixture_typed_null_candidates(
+    *, project_dir: Path, selected_paths: frozenset[Path] | None = None
+) -> FixtureNullCandidateScan:
+    """Return selected SQL-test paths and model names with possible typed-null fixtures."""
+
+    paths: set[Path] = set()
+    model_names: set[str] = set()
+    tests_root: Path = project_dir / "tests" / "unit"
+    if not tests_root.is_dir():
+        return FixtureNullCandidateScan()
+    for file_path in tests_root.rglob("*.sql"):
         if any(part in _IGNORED_SEARCH_DIRECTORIES for part in file_path.parts):
             continue
+        resolved_path: Path = file_path.resolve()
+        if selected_paths is not None and resolved_path not in selected_paths:
+            continue
         try:
             contents: str = file_path.read_text(encoding="utf-8")
         except OSError:
@@ -26,5 +48,11 @@ def has_fixture_typed_null_candidates(*, project_dir: Path) -> bool:
         if not any(marker in folded for marker in _FIXTURE_CTE_MARKERS):
             continue
         if _FIXTURE_TYPED_NULL_PATTERN.search(contents) is not None:
-            return True
-    return False
+            paths.add(resolved_path)
+            model_names.update(
+                match.group("name") for match in _MODEL_FIXTURE_PATTERN.finditer(contents)
+            )
+    return FixtureNullCandidateScan(
+        paths=frozenset(paths),
+        model_names=frozenset(model_names),
+    )
diff --git a/src/sqlbuild/lint/_helpers/native.py b/src/sqlbuild/lint/_helpers/native.py
index d3b8925c37..ac9dcd97c9 100644
--- a/src/sqlbuild/lint/_helpers/native.py
+++ b/src/sqlbuild/lint/_helpers/native.py
@@ -10,6 +10,7 @@
 
 from sqlbuild.compiler.discovery._helpers.sql.model_files import (
     parse_header_values,  # noqa: FFL102 - the compiler owns the canonical native header grammar
+    prepare_model_header_tokens,
 )
 from sqlbuild.compiler.discovery.exceptions import ModelSqlParseError
 from sqlbuild.lint.constants import (
@@ -85,6 +86,20 @@ def lint_native_headers(
     return tuple(violations)
 
 
+def prepare_native_header_cache(*, files: dict[Path, str]) -> None:
+    """Batch parse formatter-owned headers before per-file formatting and linting."""
+
+    headers: list[str] = []
+    from sqlbuild.lint._helpers.headers import scan_headers
+
+    for contents in files.values():
+        headers.extend(
+            _inner_header_text(header_text=contents[header.start : header.end])
+            for header in scan_headers(contents=contents)
+        )
+    prepare_model_header_tokens(headers)
+
+
 def format_native_headers(
     *,
     contents: str,
diff --git a/src/sqlbuild/lint/_helpers/native_format.py b/src/sqlbuild/lint/_helpers/native_format.py
index 764e57dfeb..24b16dc631 100644
--- a/src/sqlbuild/lint/_helpers/native_format.py
+++ b/src/sqlbuild/lint/_helpers/native_format.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
@@ -11,11 +12,22 @@
 from sqlbuild.lint._helpers.sqlbuild_tokens import neutralize_interpolation, restore_interpolation
 from sqlbuild.lint.constants import CARRIAGE_RETURN_LINE_FEED, LINE_FEED
 from sqlbuild.lint.exceptions import NativeLintError
-from sqlbuild.lint.models import HeaderSpan, LintConfig
+from sqlbuild.lint.models import HeaderSpan, InterpolationSite, LintConfig
 
 _NATIVE_FORMAT_API_VERSION: int = 1
 
 
+@dataclass(frozen=True)
+class _PreparedBody:
+    """One authored SQL body prepared for batched native formatting."""
+
+    start: int
+    end: int
+    trailing: str
+    cache_key: tuple[str, str]
+    interpolation_sites: tuple[InterpolationSite, ...]
+
+
 def newline_style(*, contents: str) -> str:
     """Return the newline convention used by the authored contents."""
 
@@ -36,57 +48,95 @@ def format_native_sql_bodies(
 ) -> dict[Path, str]:
     """Format supported authored SQL bodies and preserve unsupported bodies unchanged."""
 
-    formatted_files: dict[Path, str] = {}
-    response_cache: dict[tuple[str, str], dict[str, Any]] = {}
+    prepared_by_path: dict[Path, tuple[_PreparedBody, ...]] = {}
+    requests_by_key: dict[tuple[str, str], dict[str, object]] = {}
     for file_path, contents in sorted(files.items()):
         headers: tuple[HeaderSpan, ...] = scan_headers(contents=contents)
-        updated: str = contents
         body_ranges: tuple[tuple[int, int], ...] = lint_body_ranges(
             contents=contents,
             headers=headers,
             file_path=file_path,
             project_dir=project_dir,
         )
-        for body_start, body_end in reversed(body_ranges):
+        prepared_bodies: list[_PreparedBody] = []
+        for body_start, body_end in body_ranges:
             body: str = contents[body_start:body_end]
             trailing: str = body[len(body.rstrip()) :]
             core: str = body[: len(body) - len(trailing)] if trailing else body
             neutralized, sites = neutralize_interpolation(body=core)
             cache_key: tuple[str, str] = (neutralized, config.dialect)
-            response: dict[str, Any] | None = response_cache.get(cache_key)
-            if response is None:
-                response = _format_response(
-                    sql=neutralized,
-                    dialect=config.dialect,
+            requests_by_key.setdefault(
+                cache_key,
+                {
+                    "version": _NATIVE_FORMAT_API_VERSION,
+                    "sql": neutralized,
+                    "dialect": config.dialect,
+                },
+            )
+            prepared_bodies.append(
+                _PreparedBody(
+                    start=body_start,
+                    end=body_end,
+                    trailing=trailing,
+                    cache_key=cache_key,
+                    interpolation_sites=sites,
                 )
-                response_cache[cache_key] = response
+            )
+        prepared_by_path[file_path] = tuple(prepared_bodies)
+    response_cache: dict[tuple[str, str], dict[str, Any]] = _format_responses(
+        requests_by_key=requests_by_key
+    )
+    formatted_files: dict[Path, str] = {}
+    for file_path, contents in sorted(files.items()):
+        updated: str = contents
+        for prepared in reversed(prepared_by_path[file_path]):
+            response: dict[str, Any] = response_cache[prepared.cache_key]
             raw_sql: object = response.get("sql")
             changed: object = response.get("changed")
             if not isinstance(raw_sql, str) or not isinstance(changed, bool):
                 raise NativeLintError("native formatter returned invalid SQL or changed state")
             if not changed:
                 continue
-            restored: str = restore_interpolation(fixed=raw_sql, sites=sites)
-            updated = f"{updated[:body_start]}{restored}{trailing}{updated[body_end:]}"
+            restored: str = restore_interpolation(
+                fixed=raw_sql,
+                sites=prepared.interpolation_sites,
+            )
+            updated = (
+                f"{updated[: prepared.start]}{restored}{prepared.trailing}{updated[prepared.end :]}"
+            )
         if updated != contents:
             formatted_files[file_path] = updated
     return formatted_files
 
 
-def _format_response(*, sql: str, dialect: str) -> dict[str, Any]:
-    payload: dict[str, object] = {
-        "version": _NATIVE_FORMAT_API_VERSION,
-        "sql": sql,
-        "dialect": dialect,
-    }
+def _format_responses(
+    *, requests_by_key: dict[tuple[str, str], dict[str, object]]
+) -> dict[tuple[str, str], dict[str, Any]]:
+    if not requests_by_key:
+        return {}
+    ordered_requests: tuple[tuple[tuple[str, str], dict[str, object]], ...] = tuple(
+        sorted(requests_by_key.items())
+    )
     try:
         decoded: object = json.loads(
-            _native.format_sql_json(json.dumps(payload, sort_keys=True, separators=(",", ":")))
+            _native.format_sql_batch_json(
+                json.dumps(
+                    [request for _, request in ordered_requests],
+                    sort_keys=True,
+                    separators=(",", ":"),
+                )
+            )
         )
     except (TypeError, ValueError) as error:
         raise NativeLintError(str(error)) from error
-    if not isinstance(decoded, dict):
-        raise NativeLintError("native formatter returned a non-object response")
-    if decoded.get("version") != _NATIVE_FORMAT_API_VERSION:
-        raise NativeLintError("native formatter returned an unsupported response version")
-    return {str(key): value for key, value in decoded.items()}
+    if not isinstance(decoded, list) or len(decoded) != len(ordered_requests):
+        raise NativeLintError("native formatter returned an invalid batch response")
+    responses: dict[tuple[str, str], dict[str, Any]] = {}
+    for (cache_key, _), raw_response in zip(ordered_requests, decoded, strict=True):
+        if not isinstance(raw_response, dict):
+            raise NativeLintError("native formatter returned a non-object response")
+        response: dict[str, Any] = {str(key): value for key, value in raw_response.items()}
+        if response.get("version") != _NATIVE_FORMAT_API_VERSION:
+            raise NativeLintError("native formatter returned an unsupported response version")
+        responses[cache_key] = response
+    return responses
diff --git a/src/sqlbuild/lint/constants.py b/src/sqlbuild/lint/constants.py
index 14a1fa9392..e39076ac2e 100644
--- a/src/sqlbuild/lint/constants.py
+++ b/src/sqlbuild/lint/constants.py
@@ -52,6 +52,8 @@
     "hooks",
     "schemas",
 )
+PARENT_PATH_SEGMENT: str = ".."
+SQL_FILE_SUFFIX: str = ".sql"
 
 RULE_DESCRIPTION_PRESENT: str = "description-present"
 RULE_DESCRIPTION_LENGTH: str = "description-length"
diff --git a/src/sqlbuild/lint/main/run_format.py b/src/sqlbuild/lint/main/run_format.py
index 5688d9941c..c1bca1d779 100644
--- a/src/sqlbuild/lint/main/run_format.py
+++ b/src/sqlbuild/lint/main/run_format.py
@@ -4,12 +4,15 @@
 
 from pathlib import Path
 
-from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter
-from sqlbuild.compiler.compile.models import CompiledProject
 from sqlbuild.compiler.compile.types import TypedSqlValueRenderer
+from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
 from sqlbuild.compiler.planner.classes.fixture_null_autofix import FixtureNullAutofix
 from sqlbuild.lint._helpers.headers import scan_headers
-from sqlbuild.lint._helpers.native import format_native_headers, lint_native_headers
+from sqlbuild.lint._helpers.native import (
+    format_native_headers,
+    lint_native_headers,
+    prepare_native_header_cache,
+)
 from sqlbuild.lint._helpers.native_format import (
     format_native_sql_bodies,
     newline_style,
@@ -32,8 +35,8 @@ def run_format(
     config: LintConfig,
     value_renderer: TypedSqlValueRenderer | None = None,
     selected_paths: frozenset[Path] | None = None,
-    compiled_project: CompiledProject | None = None,
-    adapter: BaseAdapter | None = None,
+    discovered_inputs: DiscoveredProjectInputs | None = None,
+    fixtures_only: bool = False,
     write: bool = True,
 ) -> LintRunResult:
     """Format all DSL files in place and report the violations that remain."""
@@ -48,8 +51,8 @@ def run_format(
         files=files,
         config=config,
         project_dir=project_dir,
-        compiled_project=compiled_project,
-        adapter=adapter,
+        discovered_inputs=discovered_inputs,
+        fixtures_only=fixtures_only,
     )
     formatted: list[Path] = []
     changes: list[FormatChange] = []
@@ -75,12 +78,16 @@ def run_format(
             with file_path.open("w", encoding="utf-8", newline="") as handle:
                 _ = handle.write(rendered_contents)
         formatted.append(file_path)
-    violations: list[LintViolation] = _lint_final_contents(
-        files=files,
-        updated_contents=updated_contents,
-        config=config,
-        project_dir=project_dir,
-        value_renderer=value_renderer,
+    violations: list[LintViolation] = (
+        []
+        if fixtures_only
+        else _lint_final_contents(
+            files=files,
+            updated_contents=updated_contents,
+            config=config,
+            project_dir=project_dir,
+            value_renderer=value_renderer,
+        )
     )
     final_contents: dict[Path, str] = {
         path: updated_contents.get(path, contents) for path, contents in files.items()
@@ -104,8 +111,8 @@ def _apply_fixes(
     files: dict[Path, str],
     config: LintConfig,
     project_dir: Path,
-    compiled_project: CompiledProject | None,
-    adapter: BaseAdapter | None,
+    discovered_inputs: DiscoveredProjectInputs | None,
+    fixtures_only: bool,
 ) -> dict[Path, str]:
     """Return contents after native header and supported SQL body fixes."""
 
@@ -113,17 +120,19 @@ def _apply_fixes(
         FixtureNullAutofix.apply(
             files=files,
             project_dir=project_dir,
-            project=compiled_project,
-            adapter=adapter,
+            discovered_inputs=discovered_inputs,
         )
-        if compiled_project is not None and adapter is not None
+        if discovered_inputs is not None
         else {}
     )
+    if fixtures_only:
+        return updated
     file_path: Path
     contents: str
     current_files: dict[Path, str] = {
         file_path: updated.get(file_path, contents) for file_path, contents in files.items()
     }
+    _ = prepare_native_header_cache(files=current_files)
     for file_path, contents in sorted(current_files.items()):
         native_result: tuple[str, tuple[LintViolation, ...]] = format_native_headers(
             contents=contents,
diff --git a/src/sqlbuild/lint/main/scan_fixture_typed_null_candidates.py b/src/sqlbuild/lint/main/scan_fixture_typed_null_candidates.py
new file mode 100644
index 0000000000..c5529ddeb6
--- /dev/null
+++ b/src/sqlbuild/lint/main/scan_fixture_typed_null_candidates.py
@@ -0,0 +1,19 @@
+"""Public lint entry for selected fixture-null candidate discovery."""
+
+from pathlib import Path
+
+from sqlbuild.lint._helpers.fixture_null_candidates import (
+    scan_fixture_typed_null_candidates as _scan_fixture_typed_null_candidates,
+)
+from sqlbuild.lint.models import FixtureNullCandidateScan
+
+
+def scan_fixture_typed_null_candidates(
+    *, project_dir: Path, selected_paths: frozenset[Path] | None = None
+) -> FixtureNullCandidateScan:
+    """Return selected SQL-test fixture-null candidates and referenced model names."""
+
+    return _scan_fixture_typed_null_candidates(
+        project_dir=project_dir,
+        selected_paths=selected_paths,
+    )
diff --git a/src/sqlbuild/lint/models.py b/src/sqlbuild/lint/models.py
index a611835cf0..680b371403 100644
--- a/src/sqlbuild/lint/models.py
+++ b/src/sqlbuild/lint/models.py
@@ -48,6 +48,14 @@ class LintBody:
     allows_dynamic_output_star: bool = False
 
 
+@dataclass(frozen=True)
+class FixtureNullCandidateScan:
+    """Cheaply discovered SQL-test fixtures that may contain removable typed nulls."""
+
+    paths: frozenset[Path] = field(default_factory=frozenset)
+    model_names: frozenset[str] = field(default_factory=frozenset)
+
+
 @dataclass(frozen=True)
 class LintEdit:
     """One deterministic authored-source edit proposed by a lint rule."""
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/format/__init__.py b/tests/e2e/src/sqlbuild/cli/commands/main/format/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/format/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/format/_test_types.py
new file mode 100644
index 0000000000..9aaf0535bf
--- /dev/null
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/format/_test_types.py
@@ -0,0 +1,34 @@
+"""Test case types for format command performance guards."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class FormatPerformanceGuardTestCase:
+    """One generated format workload and its performance budgets."""
+
+    description: str
+    model_count: int
+    test_count: int
+    expected_test_file_count: int
+    expected_typed_null_candidate_count: int
+    expected_max_elapsed_seconds: float
+    expected_max_format_seconds: float
+    hard_ceiling_seconds: float
+    expected_returncode: int
+    additional_arguments: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True)
+class FormatScalingGuardTestCase:
+    """One doubling profile for selected format workloads."""
+
+    description: str
+    test_counts: tuple[int, ...]
+    model_count: int
+    expected_max_doubling_ratio: float
+    expected_max_largest_elapsed_seconds: float
+    hard_ceiling_seconds: float
+    expected_returncode: int
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/format/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/format/helpers.py
new file mode 100644
index 0000000000..84ade5534b
--- /dev/null
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/format/helpers.py
@@ -0,0 +1,50 @@
+"""Generated project helpers for format performance guards."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def write_format_performance_project(
+    *, project_dir: Path, model_count: int, test_count: int
+) -> None:
+    """Write a deterministic contract-heavy project with unique SQL test fixtures."""
+
+    project_dir.mkdir()
+    (project_dir / "sqlbuild_project.toml").write_text(
+        'name = "format_guard"\nadapter = "duckdb"\n[defaults]\ncontract = "enforced"\n',
+        encoding="utf-8",
+    )
+    models_dir: Path = project_dir / "models"
+    tests_dir: Path = project_dir / "tests" / "unit" / "orders"
+    models_dir.mkdir()
+    tests_dir.mkdir(parents=True)
+    for model_index in range(model_count):
+        models_dir.joinpath(f"orders_{model_index:05d}.sql").write_text(
+            "MODEL (columns (order_id (type INTEGER), note (type VARCHAR)));\n"
+            f"SELECT {model_index} AS order_id, 'ready' AS note\n",
+            encoding="utf-8",
+        )
+    for test_index in range(test_count):
+        model_index: int = test_index % model_count
+        tests_dir.joinpath(f"test_orders_{test_index:05d}.sql").write_text(
+            "TEST();\n\n"
+            f"WITH __ref__orders_{model_index:05d} AS (\n"
+            "  SELECT\n"
+            f"    {test_index} AS order_id,\n"
+            "    CAST(NULL AS VARCHAR) AS note\n"
+            f"), __expected__orders_{model_index:05d} AS (\n"
+            f"  SELECT {test_index} AS order_id, NULL AS note\n"
+            ")\n"
+            "SELECT 1\n",
+            encoding="utf-8",
+        )
+
+
+def count_typed_null_candidates(*, project_dir: Path) -> int:
+    """Count generated typed-null projections before or after a check-only run."""
+
+    return sum(
+        path.read_text(encoding="utf-8").count("CAST(NULL AS VARCHAR)")
+        for path in (project_dir / "tests").rglob("*.sql")
+    )
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/format/test_format_performance.py b/tests/e2e/src/sqlbuild/cli/commands/main/format/test_format_performance.py
new file mode 100644
index 0000000000..15871ffe92
--- /dev/null
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/format/test_format_performance.py
@@ -0,0 +1,160 @@
+"""Opt-in performance guard for large contract-aware format workloads."""
+
+from __future__ import annotations
+
+import re
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+from tests.e2e.src.sqlbuild.cli.commands.main.format._test_types import (
+    FormatPerformanceGuardTestCase,
+    FormatScalingGuardTestCase,
+)
+from tests.e2e.src.sqlbuild.cli.commands.main.format.helpers import (
+    count_typed_null_candidates,
+    write_format_performance_project,
+)
+
+_FORMAT_ELAPSED_PATTERN: re.Pattern[str] = re.compile(r"Formatting SQL  OK  \(([0-9.]+)s;")
+
+
+@pytest.mark.performance
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FormatPerformanceGuardTestCase(
+            description="one thousand contract fixtures format within bounded work and time",
+            model_count=250,
+            test_count=1_000,
+            expected_test_file_count=1_000,
+            expected_typed_null_candidate_count=1_000,
+            expected_max_elapsed_seconds=12.0,
+            expected_max_format_seconds=5.0,
+            hard_ceiling_seconds=20.0,
+            expected_returncode=1,
+        ),
+        FormatPerformanceGuardTestCase(
+            description="one thousand fixture-only rewrites complete without canonical churn",
+            model_count=250,
+            test_count=1_000,
+            expected_test_file_count=1_000,
+            expected_typed_null_candidate_count=1_000,
+            expected_max_elapsed_seconds=6.0,
+            expected_max_format_seconds=2.0,
+            hard_ceiling_seconds=12.0,
+            expected_returncode=1,
+            additional_arguments=("--fixtures-only",),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_contract_heavy_project_when_formatting_tests_then_work_and_time_stay_bounded(
+    tmp_path: Path,
+    test_case: FormatPerformanceGuardTestCase,
+) -> None:
+    project_dir: Path = tmp_path / "format_performance"
+    write_format_performance_project(
+        project_dir=project_dir,
+        model_count=test_case.model_count,
+        test_count=test_case.test_count,
+    )
+    test_files: tuple[Path, ...] = tuple((project_dir / "tests").rglob("*.sql"))
+    candidate_count: int = count_typed_null_candidates(project_dir=project_dir)
+
+    started_at: float = time.perf_counter()
+    result: subprocess.CompletedProcess[str] = subprocess.run(
+        [
+            str(Path(sys.executable).with_name("sqb")),
+            "--project-dir",
+            str(project_dir),
+            "--no-color",
+            "format",
+            "--check",
+            *test_case.additional_arguments,
+            "--select",
+            "path:tests",
+        ],
+        check=False,
+        capture_output=True,
+        text=True,
+        timeout=test_case.hard_ceiling_seconds,
+    )
+    elapsed_seconds: float = time.perf_counter() - started_at
+    elapsed_match: re.Match[str] | None = _FORMAT_ELAPSED_PATTERN.search(result.stderr)
+
+    assert result.returncode == test_case.expected_returncode
+    assert len(test_files) == test_case.expected_test_file_count
+    assert candidate_count == test_case.expected_typed_null_candidate_count
+    assert count_typed_null_candidates(project_dir=project_dir) == candidate_count
+    assert f"{test_case.expected_test_file_count} files checked" in result.stderr
+    assert f"{test_case.expected_test_file_count} changed" in result.stderr
+    assert elapsed_match is not None
+    assert float(elapsed_match.group(1)) <= test_case.expected_max_format_seconds
+    assert elapsed_seconds <= test_case.expected_max_elapsed_seconds
+
+
+@pytest.mark.performance
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FormatScalingGuardTestCase(
+            description="contract fixture formatting scales linearly across doublings",
+            test_counts=(500, 1_000, 2_000),
+            model_count=250,
+            expected_max_doubling_ratio=2.75,
+            expected_max_largest_elapsed_seconds=12.0,
+            hard_ceiling_seconds=20.0,
+            expected_returncode=1,
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_doubling_fixture_counts_when_formatting_then_elapsed_time_scales_linearly(
+    tmp_path: Path,
+    test_case: FormatScalingGuardTestCase,
+) -> None:
+    elapsed_samples: list[float] = []
+    format_samples: list[float] = []
+    for test_count in test_case.test_counts:
+        project_dir: Path = tmp_path / f"format_scaling_{test_count}"
+        write_format_performance_project(
+            project_dir=project_dir,
+            model_count=test_case.model_count,
+            test_count=test_count,
+        )
+
+        started_at: float = time.perf_counter()
+        result: subprocess.CompletedProcess[str] = subprocess.run(
+            [
+                str(Path(sys.executable).with_name("sqb")),
+                "--project-dir",
+                str(project_dir),
+                "--no-color",
+                "format",
+                "--check",
+                "--select",
+                "path:tests",
+            ],
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=test_case.hard_ceiling_seconds,
+        )
+        elapsed_seconds: float = time.perf_counter() - started_at
+        elapsed_match: re.Match[str] | None = _FORMAT_ELAPSED_PATTERN.search(result.stderr)
+
+        assert result.returncode == test_case.expected_returncode
+        assert count_typed_null_candidates(project_dir=project_dir) == test_count
+        assert f"{test_count} files checked" in result.stderr
+        assert elapsed_match is not None
+        elapsed_samples.append(elapsed_seconds)
+        format_samples.append(float(elapsed_match.group(1)))
+
+    assert elapsed_samples[-1] <= test_case.expected_max_largest_elapsed_seconds
+    for samples in (elapsed_samples, format_samples):
+        for smaller, larger in zip(samples[:-1], samples[1:], strict=True):
+            assert larger / smaller <= test_case.expected_max_doubling_ratio
diff --git a/tests/unit/src/sqlbuild/lint/_test_types.py b/tests/unit/src/sqlbuild/lint/_test_types.py
index 798f55a895..1743e69a0d 100644
--- a/tests/unit/src/sqlbuild/lint/_test_types.py
+++ b/tests/unit/src/sqlbuild/lint/_test_types.py
@@ -55,6 +55,17 @@ class FormatCliTestCase:
     expected_writes: bool = False
 
 
+@dataclass(frozen=True)
+class FormatSelectionWorkTestCase:
+    """Work-count expectations for one selected format run."""
+
+    description: str
+    selected_file_count: int
+    expected_fixture_parse_count: int
+    expected_native_batch_count: int
+    expected_native_request_count: int
+
+
 @dataclass(frozen=True)
 class ExpandedLintTestCase:
     """Test case for linting the SQL a project actually produces."""
diff --git a/tests/unit/src/sqlbuild/lint/test_cli_format.py b/tests/unit/src/sqlbuild/lint/test_cli_format.py
index 06941f517c..7f6c0b9d4a 100644
--- a/tests/unit/src/sqlbuild/lint/test_cli_format.py
+++ b/tests/unit/src/sqlbuild/lint/test_cli_format.py
@@ -2,12 +2,21 @@
 
 from __future__ import annotations
 
+import json
 from pathlib import Path
+from unittest.mock import Mock
 
 import pytest
+from _pytest.capture import CaptureResult
 
+import sqlbuild._native as native_module
+import sqlbuild.cli.commands._helpers.lint.selection as lint_selection_module
+import sqlbuild.compiler.planner._helpers.sql_tests.fixture_formatting as fixture_formatting_module
 from sqlbuild.cli.commands.main.entrypoint.entry import main
-from tests.unit.src.sqlbuild.lint._test_types import FormatCliTestCase
+from tests.unit.src.sqlbuild.lint._test_types import (
+    FormatCliTestCase,
+    FormatSelectionWorkTestCase,
+)
 from tests.unit.src.sqlbuild.lint.helpers import write_fixture_format_project
 
 PROJECT_TOML: str = 'name = "demo"\nadapter = "duckdb"\n'
@@ -308,6 +317,70 @@ def test_given_union_fixture_when_formatting_then_branch_shapes_are_preserved(
     assert "UNION ALL" in formatted
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [FormatCliTestCase("non-prefix fixture", (), 0, "CAST(NULL AS VARCHAR) AS status")],
+    ids=lambda case: case.description,
+)
+def test_given_middle_typed_null_when_formatting_then_star_safe_prefix_is_preserved(
+    test_case: FormatCliTestCase,
+    tmp_path: Path,
+) -> None:
+    (tmp_path / "sqlbuild_project.toml").write_text(
+        PROJECT_TOML + '\n[defaults]\ncontract = "enforced"\n',
+        encoding="utf-8",
+    )
+    model: Path = tmp_path / "models" / "orders.sql"
+    model.parent.mkdir()
+    model.write_text(
+        'MODEL (description "Orders", columns ('
+        "order_id (type INTEGER), status (type VARCHAR), amount (type DOUBLE)));\n"
+        "SELECT 1 AS order_id, 'open' AS status, 1.0 AS amount\n",
+        encoding="utf-8",
+    )
+    test_file: Path = tmp_path / "tests" / "unit" / "test_orders.sql"
+    test_file.parent.mkdir(parents=True)
+    test_file.write_text(
+        "TEST();\n\n"
+        "WITH __ref__orders AS (\n"
+        "  SELECT\n"
+        "    1 AS order_id,\n"
+        "    CAST(NULL AS VARCHAR) AS status,\n"
+        "    1.0 AS amount\n"
+        "), __expected__orders AS (SELECT 1 AS order_id)\n"
+        "SELECT 1\n",
+        encoding="utf-8",
+    )
+
+    exit_code: int = main(["--project-dir", str(tmp_path), "format"])
+
+    assert exit_code == test_case.expected_exit_code
+    assert test_case.expected_fragment in test_file.read_text(encoding="utf-8")
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [FormatCliTestCase("empty explicit selector", ("--select", " "), 1, "matched no SQL files")],
+    ids=lambda case: case.description,
+)
+def test_given_empty_explicit_selector_when_formatting_then_no_files_are_modified(
+    test_case: FormatCliTestCase,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    (tmp_path / "sqlbuild_project.toml").write_text(PROJECT_TOML, encoding="utf-8")
+    model: Path = tmp_path / "models" / "orders.sql"
+    model.parent.mkdir()
+    original: str = 'MODEL (description "Orders");\nselect 1 as order_id\n'
+    model.write_text(original, encoding="utf-8")
+
+    exit_code: int = main(["--project-dir", str(tmp_path), "format", *test_case.arguments])
+
+    assert exit_code == test_case.expected_exit_code
+    assert test_case.expected_fragment in capsys.readouterr().err
+    assert model.read_text(encoding="utf-8") == original
+
+
 @pytest.mark.parametrize(
     "test_case",
     [FormatCliTestCase("SQL analysis disabled", (), 0, "CAST(NULL AS BOOLEAN)")],
@@ -374,3 +447,140 @@ def test_given_macro_expanded_fixture_when_formatting_then_authored_source_is_no
     formatted: str = test_file.read_text(encoding="utf-8")
     assert test_case.expected_fragment in formatted
     assert "CAST(NULL AS BOOLEAN) AS flag' AS fixture_text" in formatted
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FormatCliTestCase(
+            description="fixture-only formatting preserves unrelated SQL layout",
+            arguments=("--fixtures-only", "--select", "path:tests"),
+            expected_exit_code=0,
+            expected_fragment="__expected__rows AS (select 1 AS id, TRUE AS flag)",
+            expected_writes=True,
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_fixture_only_mode_when_formatting_then_only_fixture_nulls_are_rewritten(
+    test_case: FormatCliTestCase,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    test_file: Path = write_fixture_format_project(
+        tmp_path=tmp_path,
+        fixture_sql=("\n  SELECT\n    1 AS id,\n    CAST(NULL AS BOOLEAN) AS flag\n"),
+    )
+    authored: str = test_file.read_text(encoding="utf-8").replace(
+        "__expected__rows AS (SELECT 1 AS id, TRUE AS flag)",
+        "__expected__rows AS (select 1 AS id, TRUE AS flag)",
+    )
+    test_file.write_text(authored, encoding="utf-8")
+    native_format: Mock = Mock(wraps=native_module.format_sql_batch_json)
+    monkeypatch.setattr(native_module, "format_sql_batch_json", native_format)
+
+    exit_code: int = main(["--project-dir", str(tmp_path), "format", *test_case.arguments])
+
+    formatted: str = test_file.read_text(encoding="utf-8")
+    assert exit_code == test_case.expected_exit_code
+    assert "CAST(NULL AS BOOLEAN)" not in formatted
+    assert test_case.expected_fragment in formatted
+    assert native_format.call_count == 0
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FormatSelectionWorkTestCase(
+            description="selected test path bounds fixture parsing and native formatting",
+            selected_file_count=2,
+            expected_fixture_parse_count=1,
+            expected_native_batch_count=1,
+            expected_native_request_count=1,
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_test_path_selector_when_formatting_then_work_is_bounded_to_selected_files(
+    test_case: FormatSelectionWorkTestCase,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    (tmp_path / "sqlbuild_project.toml").write_text(
+        PROJECT_TOML + '\n[defaults]\ncontract = "enforced"\n',
+        encoding="utf-8",
+    )
+    model_dir: Path = tmp_path / "models"
+    model_dir.mkdir()
+    model_dir.joinpath("stg_orders.sql").write_text(
+        "MODEL (columns (order_id (type INTEGER), status (type VARCHAR)));\n"
+        "SELECT 1 AS order_id, 'open' AS status\n",
+        encoding="utf-8",
+    )
+    model_dir.joinpath("orders.sql").write_text(
+        "MODEL (columns (order_id (type INTEGER), status (type VARCHAR)));\n"
+        'SELECT order_id, status FROM __ref("stg_orders")\n',
+        encoding="utf-8",
+    )
+    fixture_sql: str = (
+        "TEST();\n\n"
+        "WITH __ref__stg_orders AS (\n"
+        "  SELECT\n"
+        "    1 AS order_id,\n"
+        "    CAST(NULL AS VARCHAR) AS status\n"
+        "), __expected__orders AS (\n"
+        "  SELECT 1 AS order_id, NULL AS status\n"
+        ")\n"
+        "SELECT 1\n"
+    )
+    selected_dir: Path = tmp_path / "tests" / "unit" / "orders"
+    selected_dir.mkdir(parents=True)
+    selected_files: tuple[Path, ...] = (
+        selected_dir / "test_first.sql",
+        selected_dir / "test_second.sql",
+    )
+    selected_files[0].write_text(fixture_sql, encoding="utf-8")
+    selected_files[1].write_text(
+        fixture_sql.replace(
+            "    1 AS order_id,\n    CAST(NULL AS VARCHAR) AS status\n",
+            "    1 AS order_id\n",
+        ),
+        encoding="utf-8",
+    )
+    unselected_file: Path = tmp_path / "tests" / "unit" / "customers" / "test_other.sql"
+    unselected_file.parent.mkdir()
+    unselected_file.write_text(fixture_sql, encoding="utf-8")
+    fixture_parse: Mock = Mock(wraps=fixture_formatting_module.SqlTestCteExtractor.extract)
+    native_format: Mock = Mock(wraps=native_module.format_sql_batch_json)
+    full_discovery: Mock = Mock(side_effect=AssertionError("full discovery must not run"))
+    project_graph: Mock = Mock(side_effect=AssertionError("project graph must not build"))
+    monkeypatch.setattr(fixture_formatting_module.SqlTestCteExtractor, "extract", fixture_parse)
+    monkeypatch.setattr(native_module, "format_sql_batch_json", native_format)
+    monkeypatch.setattr(lint_selection_module, "discover_project_inputs", full_discovery)
+    monkeypatch.setattr(lint_selection_module, "build_project_graph", project_graph)
+
+    exit_code: int = main(
+        [
+            "--project-dir",
+            str(tmp_path),
+            "format",
+            "--select",
+            "path:tests/unit/orders",
+        ]
+    )
+
+    captured: CaptureResult[str] = capsys.readouterr()
+    assert exit_code == 0
+    assert f"{test_case.selected_file_count} files checked" in captured.err
+    assert fixture_parse.call_count == test_case.expected_fixture_parse_count
+    assert native_format.call_count == test_case.expected_native_batch_count
+    native_requests: object = json.loads(native_format.call_args.args[0])
+    assert isinstance(native_requests, list)
+    assert len(native_requests) == test_case.expected_native_request_count
+    assert full_discovery.call_count == 0
+    assert project_graph.call_count == 0
+    assert all(
+        "CAST(NULL AS VARCHAR)" not in path.read_text(encoding="utf-8") for path in selected_files
+    )
+    assert "CAST(NULL AS VARCHAR)" in unselected_file.read_text(encoding="utf-8")