fix(planner): avoid duplicate planning work

src/sqlbuild/compiler/planner/_helpers/changes/detect.py

@@ -60,32 +60,13 @@ def detect_changes(
 ) -> PlannerChangeResults:
     """Detect selected model and function changes."""
 
-    model_changes: dict[str, ChangeDetectionResult] = {}
-    function_local_hashes: dict[str, str] = build_function_local_hashes(
-        functions=project.functions,
+    model_changes: dict[str, ChangeDetectionResult] = detect_model_changes_in_scope(
+        project=project,
+        scope=scope,
+        snapshot=snapshot,
+        expected_version_hashes=expected_version_hashes,
+        expected_metadata_jsons=expected_metadata_jsons,
     )
-    hook_version_hashes: dict[str, str] = {
-        name: identity.version_hash
-        for name, identity in build_hook_identities(project.hook_functions).items()
-    }
-    key: CompiledObjectKey
-    for key in scope.execution_order:
-        if key not in scope.selected_keys or key.resource_type != CompiledResourceType.MODEL:
-            continue
-        model: CompiledModel | None = scope.models_by_name.get(key.name)
-        if model is None:
-            continue
-        model_changes[model.name] = detect_model_changes(
-            model=model,
-            snapshot=snapshot,
-            sql_analysis_enabled=project.settings.sql_analysis,
-            query_change_tracking=project.settings.query_change_tracking,
-            full_refresh=False,
-            function_local_hashes=function_local_hashes,
-            hook_version_hashes=hook_version_hashes,
-            expected_version_hash=(expected_version_hashes or {}).get(model.name),
-            expected_metadata_json=(expected_metadata_jsons or {}).get(model.name),
-        )
 
     function_changes: dict[str, FunctionChangeResult] = {}
     function: CompiledFunction
@@ -116,6 +97,51 @@ def detect_changes(
     return PlannerChangeResults(models=model_changes, functions=function_changes)
 
 
+def detect_model_changes_in_scope(
+    *,
+    project: CompiledProject,
+    scope: PlannerScope,
+    snapshot: WarehouseSnapshot,
+    query_change_tracking: bool | None = None,
+    expected_version_hashes: dict[str, str] | None = None,
+    expected_metadata_jsons: dict[str, str] | None = None,
+) -> dict[str, ChangeDetectionResult]:
+    """Detect changes for selected models without inspecting functions."""
+
+    model_changes: dict[str, ChangeDetectionResult] = {}
+    effective_query_change_tracking: bool = (
+        project.settings.query_change_tracking
+        if query_change_tracking is None
+        else query_change_tracking
+    )
+    function_local_hashes: dict[str, str] = build_function_local_hashes(
+        functions=project.functions,
+    )
+    hook_version_hashes: dict[str, str] = {
+        name: identity.version_hash
+        for name, identity in build_hook_identities(project.hook_functions).items()
+    }
+    key: CompiledObjectKey
+    for key in scope.execution_order:
+        if key not in scope.selected_keys or key.resource_type != CompiledResourceType.MODEL:
+            continue
+        model: CompiledModel | None = scope.models_by_name.get(key.name)
+        if model is None:
+            continue
+        model_changes[model.name] = detect_model_changes(
+            model=model,
+            snapshot=snapshot,
+            sql_analysis_enabled=project.settings.sql_analysis,
+            query_change_tracking=effective_query_change_tracking,
+            full_refresh=False,
+            function_local_hashes=function_local_hashes,
+            hook_version_hashes=hook_version_hashes,
+            expected_version_hash=(expected_version_hashes or {}).get(model.name),
+            expected_metadata_json=(expected_metadata_jsons or {}).get(model.name),
+        )
+    return model_changes
+
+
 def detect_model_changes(
     *,
     model: CompiledModel,

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

@@ -198,19 +198,27 @@ def _fixture_keys_requiring_analysis(
     return frozenset(required)
 
 
-def build_relation_fixture_context(*, project: CompiledProject) -> RelationFixturePlanningContext:
+def build_relation_fixture_context(
+    *,
+    project: CompiledProject,
+    models_by_name: dict[str, CompiledModel] | None = None,
+) -> RelationFixturePlanningContext:
     """Build immutable project-wide fixture metadata once per planning invocation."""
 
-    models_by_name: dict[str, CompiledModel] = {model.name: model for model in project.models}
+    model_map: dict[str, CompiledModel] = (
+        {model.name: model for model in project.models}
+        if models_by_name is None
+        else models_by_name
+    )
     relations: dict[FixtureKey, FixtureRelationMetadata] = _fixture_relation_metadata(
         project=project,
-        model_map=models_by_name,
+        model_map=model_map,
     )
     authoritative_columns: dict[FixtureKey, frozenset[str]] = _authoritative_fixture_columns(
         relations=relations
     )
     return RelationFixturePlanningContext(
-        models_by_name=models_by_name,
+        models_by_name=model_map,
         relations=relations,
         authoritative_columns=authoritative_columns,
         expected_types=_base_expected_fixture_types(relations=relations),

src/sqlbuild/compiler/planner/_helpers/output/plan_output.py

@@ -33,7 +33,10 @@
     scope_overlaps,
 )
 from sqlbuild.compiler.planner._helpers.resolve.resolve import resolve_function_sql
-from sqlbuild.compiler.planner._helpers.sql_tests.assembly import plan_test
+from sqlbuild.compiler.planner._helpers.sql_tests.assembly import (
+    build_sql_test_planning_context,
+    plan_test,
+)
 from sqlbuild.compiler.planner.exceptions import (
     PlannerInputError,
     SqlTestFixtureValidationError,
@@ -56,6 +59,7 @@
     SeedPlanEntry,
     SourceLoadPlanEntry,
     SqlTestPlanEntry,
+    SqlTestPlanningContext,
     WarehouseSnapshot,
 )
 from sqlbuild.compiler.planner.types import PlanReason
@@ -445,15 +449,29 @@ def build_selected_test_entries(
     entries: list[SqlTestPlanEntry] = []
     warnings: list[PlanWarning] = []
     fixture_diagnostics: list[str] = []
-    fixture_planning_context: RelationFixturePlanningContext | None = None
+    selected_tests: list[CompiledSqlTest] = []
     sql_test: CompiledSqlTest
     for sql_test in project.sql_tests:
         if not scope_overlaps(scope_deps=sql_test.scope_deps, selected_keys=selected_keys):
             continue
         if case_name is not None and sql_test.case_name != case_name:
             continue
-        if project.settings.sql_analysis and fixture_planning_context is None:
-            fixture_planning_context = build_relation_fixture_context(project=project)
+        selected_tests.append(sql_test)
+    if not selected_tests:
+        return entries, warnings
+    test_planning_context: SqlTestPlanningContext = build_sql_test_planning_context(
+        project=project,
+        tests=tuple(selected_tests),
+    )
+    fixture_planning_context: RelationFixturePlanningContext | None = (
+        build_relation_fixture_context(
+            project=project,
+            models_by_name=test_planning_context.models_by_name,
+        )
+        if project.settings.sql_analysis
+        else None
+    )
+    for sql_test in selected_tests:
         test_entry: SqlTestPlanEntry
         test_warnings: tuple[PlanWarning, ...]
         try:
@@ -464,6 +482,7 @@ def build_selected_test_entries(
                 sql_analysis_enabled=project.settings.sql_analysis,
                 validate_fixtures=True,
                 fixture_planning_context=fixture_planning_context,
+                planning_context=test_planning_context,
             )
         except SqlTestFixtureValidationError as error:
             fixture_diagnostics.extend(error.diagnostics)

src/sqlbuild/compiler/planner/_helpers/planning/identities.py

@@ -4,12 +4,15 @@
 
 from dataclasses import replace
 
-from sqlbuild.compiler.compile.models import CompiledProject
-from sqlbuild.compiler.planner._helpers.changes.detect import detect_changes
+from sqlbuild.compiler.compile.models import CompiledObjectKey, CompiledProject
+from sqlbuild.compiler.compile.types import CompiledResourceType
+from sqlbuild.compiler.planner._helpers.changes.detect import detect_model_changes_in_scope
 from sqlbuild.compiler.planner._helpers.identity.direct import (
     build_direct_model_version_identities,
 )
 from sqlbuild.compiler.planner.models import (
+    ChangeDetectionResult,
+    DirectModelVersionIdentities,
     PlannerChangeResults,
     PlannerIdentityContext,
     PlannerScopeResolution,
@@ -21,21 +24,27 @@ def build_planner_identity_context(
     *,
     project: CompiledProject,
     scopes: PlannerScopeResolution,
+    include_stale_warning_identities: bool = True,
 ) -> PlannerIdentityContext:
     """Build expected version identities for the inspection and stale-warning scopes."""
 
+    version_identities: DirectModelVersionIdentities = build_direct_model_version_identities(
+        functions=project.functions,
+        seeds=project.seeds,
+        scope=scopes.inspection_scope,
+        hook_functions=project.hook_functions,
+    )
     return PlannerIdentityContext(
-        version_identities=build_direct_model_version_identities(
-            functions=project.functions,
-            seeds=project.seeds,
-            scope=scopes.inspection_scope,
-            hook_functions=project.hook_functions,
-        ),
-        stale_warning_identities=build_direct_model_version_identities(
-            functions=project.functions,
-            seeds=project.seeds,
-            scope=scopes.stale_warning_scope,
-            hook_functions=project.hook_functions,
+        version_identities=version_identities,
+        stale_warning_identities=(
+            build_direct_model_version_identities(
+                functions=project.functions,
+                seeds=project.seeds,
+                scope=scopes.stale_warning_scope,
+                hook_functions=project.hook_functions,
+            )
+            if include_stale_warning_identities
+            else version_identities
         ),
     )
 
@@ -46,17 +55,64 @@ def detect_stale_warning_changes(
     scopes: PlannerScopeResolution,
     snapshot: WarehouseSnapshot,
     identities: PlannerIdentityContext,
+    execution_changes: PlannerChangeResults,
 ) -> PlannerChangeResults:
-    """Detect project-wide changes used for stale-out-of-selection warnings."""
+    """Detect only stale-warning model changes not already known from execution planning."""
 
-    return detect_changes(
+    reusable_model_names: frozenset[str] = _reusable_execution_model_names(
+        identities=identities,
+        execution_changes=execution_changes,
+        query_change_tracking=project.settings.query_change_tracking,
+    )
+    stale_keys: frozenset[CompiledObjectKey] = frozenset(
+        scopes.stale_warning_scope.all_keys.values()
+    )
+    remaining_model_keys: frozenset[CompiledObjectKey] = frozenset(
+        key
+        for key in stale_keys
+        if key.resource_type == CompiledResourceType.MODEL and key.name not in reusable_model_names
+    )
+    remaining_model_changes: dict[str, ChangeDetectionResult] = detect_model_changes_in_scope(
         project=project,
         scope=replace(
             scopes.stale_warning_scope,
-            selected_keys=frozenset(scopes.stale_warning_scope.all_keys.values()),
+            selected_keys=remaining_model_keys,
         ),
         snapshot=snapshot,
-        full_refresh=False,
+        query_change_tracking=True,
         expected_version_hashes=identities.stale_warning_identities.model_version_hashes,
         expected_metadata_jsons=identities.stale_warning_identities.model_metadata_jsons,
     )
+    return PlannerChangeResults(
+        models={
+            **{
+                name: change
+                for name, change in execution_changes.models.items()
+                if name in reusable_model_names
+            },
+            **remaining_model_changes,
+        },
+        functions={},
+    )
+
+
+def _reusable_execution_model_names(
+    *,
+    identities: PlannerIdentityContext,
+    execution_changes: PlannerChangeResults,
+    query_change_tracking: bool,
+) -> frozenset[str]:
+    if not query_change_tracking:
+        return frozenset()
+    execution_identities: DirectModelVersionIdentities = identities.version_identities
+    stale_identities: DirectModelVersionIdentities = identities.stale_warning_identities
+    return frozenset(
+        model_name
+        for model_name in execution_changes.models
+        if (
+            execution_identities.model_version_hashes.get(model_name)
+            == stale_identities.model_version_hashes.get(model_name)
+            and execution_identities.model_metadata_jsons.get(model_name)
+            == stale_identities.model_metadata_jsons.get(model_name)
+        )
+    )

src/sqlbuild/compiler/planner/_helpers/planning/output_assembly.py

@@ -93,6 +93,7 @@ def with_plan_warnings(
             version_identities=identities.stale_warning_identities,
             source_freshness=source_freshness,
             include_sources=False,
+            model_changes_complete=True,
         )
     )
     if stale_out_of_selection_warnings:

src/sqlbuild/compiler/planner/_helpers/pruning/selection_staleness.py

@@ -34,6 +34,7 @@ def build_stale_out_of_selection_warnings(
     version_identities: DirectModelVersionIdentities,
     source_freshness: DirectSourceFreshnessPlanningResult | None,
     include_sources: bool = True,
+    model_changes_complete: bool = False,
 ) -> tuple[PlanWarning, ...]:
     """Warn for selected models stale through changed upstreams outside the run set."""
 
@@ -67,6 +68,7 @@ def build_stale_out_of_selection_warnings(
             changes=changes,
             snapshot=snapshot,
             version_identities=version_identities,
+            model_changes_complete=model_changes_complete,
         ),
         changed_seed_names=_changed_seed_names(
             snapshot=snapshot,
@@ -129,6 +131,7 @@ def _changed_model_names(
     changes: PlannerChangeResults,
     snapshot: WarehouseSnapshot,
     version_identities: DirectModelVersionIdentities,
+    model_changes_complete: bool,
 ) -> frozenset[str]:
     changed: set[str] = {
         model_name
@@ -150,6 +153,7 @@ def _changed_model_names(
             changes=changes,
             snapshot=snapshot,
             version_identities=version_identities,
+            model_changes_complete=model_changes_complete,
         )
     )
     return frozenset(changed)
@@ -186,16 +190,25 @@ def _model_own_identity_changed(
     changes: PlannerChangeResults,
     snapshot: WarehouseSnapshot,
     version_identities: DirectModelVersionIdentities,
+    model_changes_complete: bool,
 ) -> bool:
     selected_change: ChangeDetectionResult | None = changes.models.get(model_name)
-    if selected_change is not None:
-        if selected_change.change_kind in {
+    if selected_change is not None and (
+        model_changes_complete
+        or selected_change.change_kind
+        in {
             ChangeKind.FIRST_RUN,
             ChangeKind.QUERY_CHANGED,
             ChangeKind.CONFIG_CHANGED,
             ChangeKind.SCHEMA_CHANGED,
-        }:
-            return True
+        }
+    ):
+        return selected_change.change_kind in {
+            ChangeKind.FIRST_RUN,
+            ChangeKind.QUERY_CHANGED,
+            ChangeKind.CONFIG_CHANGED,
+            ChangeKind.SCHEMA_CHANGED,
+        }
     model: CompiledModel | None = original_scope.models_by_name.get(model_name)
     if model is None:
         return False