@@ -31,6 +31,7 @@
     binding_query_fields,
     lineage_reference_map,
     prepare_binding_queries,
+    relation_stubs_collide,
     remembered_shared_results,
     reused_shared_results,
     shared_result_is_exact,
@@ -723,22 +724,24 @@ def _prepare_compact_analysis_batch(
         shared: SharedBindingQuery | None = (
             binding_query.shared if binding_query is not None else None
         )
-        qualified_reference_names: frozenset[str] = (
-            _qualified_reference_names(
+        stubbed_reference_names: frozenset[str] = (
+            frozenset()
+            if binding_schema is not None
+            or relation_stubs_collide(cleaned_sql=cleaned_sql, names=lineage_references)
+            else frozenset(lineage_references)
+            - _qualified_reference_names(
                 query_sql=cleaned_sql,
                 reference_names=lineage_references.keys(),
             )
-            if binding_schema is None
-            else frozenset()
         )
         canonical_stubs: dict[str, str] = (
             shared.stubs
             if shared is not None
             else {
                 name: (
-                    name
-                    if binding_schema is not None or name in qualified_reference_names
-                    else f"{COMPACT_RELATION_STUB_PREFIX}{index}"
+                    f"{COMPACT_RELATION_STUB_PREFIX}{index}"
+                    if name in stubbed_reference_names
+                    else name
                 )
                 for index, name in enumerate(lineage_references)
             }
 
@@ -4,7 +4,7 @@
 
 import re
 from collections import Counter
-from collections.abc import Mapping, Sequence
+from collections.abc import Iterable, Mapping, Sequence
 from dataclasses import replace
 from typing import Any, cast
 
@@ -190,6 +190,15 @@ def shared_result_is_exact(*, validation: object, template: object) -> bool:
     return isinstance(validation, dict) and cast(dict[str, object], validation).get("errors") == []
 
 
+def relation_stubs_collide(*, cleaned_sql: str, names: Iterable[str]) -> bool:
+    """Return whether authored SQL or relation names already use the relation stub prefix."""
+
+    return (
+        any(name.startswith(COMPACT_RELATION_STUB_PREFIX) for name in names)
+        or COMPACT_RELATION_STUB_PREFIX.casefold() in cleaned_sql.casefold()
+    )
+
+
 def _shared_reference_names(
     *,
     cleaned_sql: str,
@@ -199,9 +208,7 @@ def _shared_reference_names(
     names: list[str] = list(lineage_reference_map(references))
     if not names or len({name.casefold() for name in names}) != len(names):
         return None
-    if any(name.startswith(COMPACT_RELATION_STUB_PREFIX) for name in names):
-        return None
-    if COMPACT_RELATION_STUB_PREFIX.casefold() in cleaned_sql.casefold():
+    if relation_stubs_collide(cleaned_sql=cleaned_sql, names=names):
         return None
     if SQL_QUOTED_IDENTIFIER_DELIMITER in cleaned_sql:
         return None
 
@@ -12,6 +12,14 @@ class SourceSemanticBindingCase:
     expected_code: str = "B002"
 
 
+@dataclass(frozen=True)
+class RelationStubNameCase:
+    description: str
+    stub_cte_name: str
+    expected_downstream_type: str
+    expected_ghost_column_diagnostics: frozenset[tuple[str, str]]
+
+
 @dataclass(frozen=True)
 class SemanticCorpusCase:
     description: str
@@ -106,6 +114,19 @@ class LayeredProductionCompilePerformanceGuardTestCase:
     expected_config_edit_max_seconds: float
 
 
+@dataclass(frozen=True)
+class LayeredProductionCompileTestCase:
+    description: str
+    model_count: int
+    source_count: int
+    seed_count: int
+    function_count: int
+    macro_count: int
+    test_count: int
+    audit_count: int
+    expected_diagnostic_codes: tuple[str, ...]
+
+
 @dataclass(frozen=True)
 class SemanticCompilePerformanceGuardTestCase:
     description: str
 
@@ -30,6 +30,7 @@
     FreshProcessCompileCachePerformanceGuardTestCase,
     SemanticCorpusCase,
 )
+from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import prepare_inline_project
 
 _DBT_SHAPED_SQL_SIZE_PROFILE: tuple[tuple[float, int], ...] = (
     (0.50, 1_800),
@@ -1474,8 +1475,12 @@ def _scope_single_folder_macros(*, project_dir: Path) -> None:
 
 
 def _layered_write_schemas(*, project_dir: Path) -> None:
-    schemas_dir: Path = project_dir / "schemas"
-    schemas_dir.mkdir()
+    """Write the contract schema beside its only consumer, the first staging model."""
+
+    schemas_dir: Path = (
+        project_dir / "models" / _layered_model_folder(index=0) / "_sqlbuild" / "_schemas"
+    )
+    schemas_dir.mkdir(parents=True)
     (schemas_dir / "benchmark_row.sql").write_text(
         """SCHEMA (
   name benchmark_row,
@@ -1619,16 +1624,17 @@ def _layered_dependent_model_sql(
     previous_name: str = f"model_{index - 1:05d}"
     macro_index: int = (index // _MACRO_INTERVAL) % macro_count
     id_expression: str = {
-        True: f'@macro_{macro_index:05d}("id")',
-        False: f"id + {index % 7}",
+        True: f'@macro_{macro_index:05d}("previous.id")',
+        False: f"previous.id + {index % 7}",
     }[index % _MACRO_INTERVAL == 0]
     generated_amount_expression: str = (
-        f"amount + {index % 11} + {_layered_generated_mapping_expression()} "
+        f"previous.amount + {index % 11} + "
+        f"{_layered_generated_mapping_expression(column='previous.id')} "
         "+ CAST(@@benchmark_revision AS INTEGER)"
     )
     function_index: int = index % function_count
     amount_expression: str = {
-        True: f'__udf("fn_{function_index:05d}")(amount)',
+        True: f'__udf("fn_{function_index:05d}")(previous.amount)',
         False: generated_amount_expression,
     }[index % _FUNCTION_INTERVAL == 0]
     seed_index: int = index % seed_count
@@ -1639,7 +1645,7 @@ def _layered_dependent_model_sql(
     query: str = f'''SELECT
   {id_expression} AS id,
   {amount_expression} AS amount,
-  CASE WHEN id % 2 = 0 THEN 'even' ELSE 'odd' END AS status
+  CASE WHEN previous.id % 2 = 0 THEN 'even' ELSE 'odd' END AS status
 FROM __ref("{previous_name}") AS previous{join_sql}
 '''
     direct_sql: str = f"{_layered_model_header(index=index, audit_count=audit_count)}\n\n{query}"
@@ -2061,8 +2067,10 @@ def _layered_model_sql_size_target(*, index: int, model_count: int) -> int:
     return round(lower_size + position * (upper_size - lower_size))
 
 
-def _layered_generated_mapping_expression() -> str:
-    clauses: str = "".join(f"WHEN id = {value:05d} THEN {value % 13:02d} " for value in range(45))
+def _layered_generated_mapping_expression(*, column: str = "id") -> str:
+    clauses: str = "".join(
+        f"WHEN {column} = {value:05d} THEN {value % 13:02d} " for value in range(45)
+    )
     return f"CASE {clauses} ELSE 0 END"
 
 
@@ -2276,3 +2284,44 @@ def plan_reasons_and_migrations(
 
 def _query_changed(model: dict[str, Any]) -> bool:
     return model["reason"] == "query_changed"
+
+
+def write_relation_stub_project(
+    *, tmp_path: Path, stub_cte_name: str, selected_column: str
+) -> Path:
+    """Write a project whose middle model authors a CTE named like a relation stub."""
+
+    return prepare_inline_project(
+        tmp_path=tmp_path,
+        project_name="relation_stub_names",
+        repo_files={
+            "sqlbuild_project.toml": (
+                'name = "relation_stub_names"\n'
+                'adapter = "duckdb"\n'
+                'default_target = "dev"\n\n'
+                "[connection]\n"
+                'database = "orders.duckdb"\n\n'
+                "[targets.dev]\n"
+                'schema = "main"\n'
+            ),
+            "models/orders.sql": (
+                "MODEL (materialized table);\n\nSELECT CAST(7 AS INTEGER) AS order_id\n"
+            ),
+            "models/order_status.sql": (
+                "MODEL (materialized table);\n\n"
+                f"WITH {stub_cte_name} AS (\n"
+                "  SELECT 'pending' AS order_id, 'pending' AS ghost_status\n"
+                ")\n"
+                f'SELECT {selected_column} FROM __ref("orders")\n'
+            ),
+            "models/next_order.sql": (
+                "MODEL (\n"
+                "  materialized table,\n"
+                "  contract enforced,\n"
+                "  columns (order_id (type INTEGER), next_order_id (type INTEGER)),\n"
+                ");\n\n"
+                "SELECT order_id, order_id + 1 AS next_order_id\n"
+                'FROM __ref("order_status")\n'
+            ),
+        },
+    )
@@ -237,7 +237,7 @@ def test_given_test_heavy_project_when_compiling_then_finishes_within_budget(
             macro_count=12,
             test_count=130,
             expected_audit_count=700,
-            expected_hook_count=2,
+            expected_hook_count=0,
             expected_min_model_sql_bytes=5_500_000,
             expected_max_model_sql_bytes=7_000_000,
             expected_min_compiled_test_bytes=3_000_000,
 
@@ -0,0 +1,77 @@
+"""E2E coverage that the layered production benchmark project compiles cleanly."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import (
+    LayeredProductionCompileTestCase,
+)
+from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import (
+    write_layered_production_compile_project,
+)
+from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import run_sqb
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        LayeredProductionCompileTestCase(
+            description="small layered project with seed joins, macros, functions and contract",
+            model_count=350,
+            source_count=83,
+            seed_count=16,
+            function_count=8,
+            macro_count=4,
+            test_count=47,
+            audit_count=251,
+            expected_diagnostic_codes=(),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_layered_production_project_when_compiling_then_reports_no_diagnostics(
+    test_case: LayeredProductionCompileTestCase,
+    tmp_path: Path,
+) -> None:
+    project_dir: Path = tmp_path / "layered_production"
+    write_layered_production_compile_project(
+        project_dir=project_dir,
+        model_count=test_case.model_count,
+        source_count=test_case.source_count,
+        seed_count=test_case.seed_count,
+        function_count=test_case.function_count,
+        macro_count=test_case.macro_count,
+        test_count=test_case.test_count,
+        audit_count=test_case.audit_count,
+    )
+    model_sql: str = "".join(
+        path.read_text(encoding="utf-8") for path in (project_dir / "models").rglob("*.sql")
+    )
+    for fragment in ("__seed(", "__udf(", "@macro_", "model_schema benchmark_row"):
+        assert fragment in model_sql
+
+    result: subprocess.CompletedProcess[str] = run_sqb(
+        project_dir=project_dir,
+        command=("compile", "--no-cache", "--json"),
+    )
+
+    assert result.returncode == 0, result.stdout + result.stderr
+    payload: dict[str, Any] = json.loads(result.stdout)
+    diagnostic_codes: tuple[str, ...] = tuple(
+        str(diagnostic["code"]) for diagnostic in payload["diagnostics"]
+    )
+    assert diagnostic_codes == test_case.expected_diagnostic_codes, payload["diagnostics"]
+    assert payload["summary"]["errors"] == 0
+    assert payload["summary"]["warnings"] == 0
+    assert payload["summary"]["models"] == test_case.model_count
+    assert payload["summary"]["tests"] == test_case.test_count
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-vv"])