perf(planner): read fingerprint state once and keep plan generation linear

src/sqlbuild/cli/commands/_helpers/compile/pipeline.py

@@ -53,6 +53,7 @@
 from sqlbuild.presentation.classes.transient_status_reporter import TransientStatusReporter
 from sqlbuild.rule_engine.classes.early_sql_lint import EarlySqlLint
 from sqlbuild.rule_engine.main.load_config import load_rules_config
+from sqlbuild.rule_engine.main.render_rules_progress import format_rules_progress
 from sqlbuild.rule_engine.main.run_rules import run_rules
 from sqlbuild.rule_engine.models import RulesConfig, RulesRunResult
 from sqlbuild.runtime.observability.classes.operation_lifecycle import OperationLifecycle
@@ -204,6 +205,7 @@ def _analyze_compile_project(
         ):
             prepared_artifacts.start(project=graph.project, adapter=adapter)
         _ = start_compile_phase(status=status, message="Evaluating built-in and custom rules...")
+        rules_start: float = time.monotonic()
         rules_result = run_rules(
             graph=graph,
             discovered_inputs=discovered_inputs,
@@ -217,9 +219,8 @@ def _analyze_compile_project(
         )
         _ = complete_compile_phase(
             status=status,
-            message=(
-                f"Evaluated rules. (built-in {rules_result.built_in_ms / 1000:.2f}s, "
-                f"custom {rules_result.custom_ms / 1000:.2f}s)"
+            message=format_rules_progress(
+                elapsed_seconds=time.monotonic() - rules_start, result=rules_result
             ),
         )
     rule_diagnostics: tuple[CompilerDiagnostic, ...] = tuple(

src/sqlbuild/compiler/fingerprints/_helpers/latest.py

@@ -0,0 +1,25 @@
+"""Latest fingerprint set assembly shared by warehouse reads and in-memory selection."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+from sqlbuild.compiler.fingerprints.constants import NODE_TYPE_MODEL
+from sqlbuild.compiler.fingerprints.models import Fingerprint, FingerprintSet
+
+
+def build_latest_fingerprint_set(*, schema: str, latest: Iterable[Fingerprint]) -> FingerprintSet:
+    """Index latest rows by identity and by node name, preferring models on name collisions."""
+
+    fingerprints: dict[str, Fingerprint] = {}
+    fingerprints_by_identity: dict[tuple[str, str], Fingerprint] = {}
+    fingerprint: Fingerprint
+    for fingerprint in latest:
+        fingerprints_by_identity[(fingerprint.node_type, fingerprint.node_name)] = fingerprint
+        if fingerprint.node_type == NODE_TYPE_MODEL or fingerprint.node_name not in fingerprints:
+            fingerprints[fingerprint.node_name] = fingerprint
+    return FingerprintSet(
+        schema=schema,
+        fingerprints=fingerprints,
+        fingerprints_by_identity=fingerprints_by_identity,
+    )

src/sqlbuild/compiler/fingerprints/main/_select_latest.py

@@ -0,0 +1,28 @@
+"""In-memory node selection over an unfiltered latest fingerprint read."""
+
+from __future__ import annotations
+
+from sqlbuild.compiler.fingerprints._helpers.latest import build_latest_fingerprint_set
+from sqlbuild.compiler.fingerprints.models import FingerprintSet
+
+
+def select_latest_fingerprints(
+    *,
+    fingerprint_set: FingerprintSet,
+    node_names: tuple[str, ...],
+    filtered_node_types: tuple[str, ...] = (),
+) -> FingerprintSet:
+    """Apply the ``read_latest_fingerprints`` node filter to an unfiltered latest read."""
+
+    if not node_names and not filtered_node_types:
+        return FingerprintSet(schema=fingerprint_set.schema, fingerprints={})
+    names: frozenset[str] = frozenset(node_names)
+    types: frozenset[str] = frozenset(filtered_node_types)
+    return build_latest_fingerprint_set(
+        schema=fingerprint_set.schema,
+        latest=(
+            fingerprint
+            for fingerprint in (fingerprint_set.fingerprints_by_identity or {}).values()
+            if fingerprint.node_name in names or (types and fingerprint.node_type not in types)
+        ),
+    )

src/sqlbuild/compiler/fingerprints/main/read.py

@@ -9,10 +9,10 @@
 from typing import Any
 
 from sqlbuild.adapter.contract.types import AdapterExecute
+from sqlbuild.compiler.fingerprints._helpers.latest import build_latest_fingerprint_set
 from sqlbuild.compiler.fingerprints._helpers.sql import (
     build_qualified_table_name,
 )
-from sqlbuild.compiler.fingerprints.constants import NODE_TYPE_MODEL
 from sqlbuild.compiler.fingerprints.exceptions import FingerprintInputError
 from sqlbuild.compiler.fingerprints.models import Fingerprint, FingerprintSet
 
@@ -75,18 +75,9 @@ def read_latest_fingerprints(
             "fingerprint table to regenerate fingerprints."
         ) from error
     rows: list[tuple[Any, ...]] = result.fetchall()
-    fingerprints: dict[str, Fingerprint] = {}
-    fingerprints_by_identity: dict[tuple[str, str], Fingerprint] = {}
-    row: tuple[Any, ...]
-    for row in rows:
-        fingerprint: Fingerprint = _row_to_fingerprint(row=row, qualified_name=qualified_name)
-        fingerprints_by_identity[(fingerprint.node_type, fingerprint.node_name)] = fingerprint
-        if fingerprint.node_type == NODE_TYPE_MODEL or fingerprint.node_name not in fingerprints:
-            fingerprints[fingerprint.node_name] = fingerprint
-    return FingerprintSet(
+    return build_latest_fingerprint_set(
         schema=schema,
-        fingerprints=fingerprints,
-        fingerprints_by_identity=fingerprints_by_identity,
+        latest=(_row_to_fingerprint(row=row, qualified_name=qualified_name) for row in rows),
     )
 
 

src/sqlbuild/compiler/pipeline/main/compile.py

@@ -66,7 +66,7 @@
 from sqlbuild.rule_engine.exceptions import RulesError
 from sqlbuild.rule_engine.main.load_config import load_rules_config
 from sqlbuild.rule_engine.main.render_result import format_result
-from sqlbuild.rule_engine.main.render_skipped_rules import format_skipped_type_proof_rules
+from sqlbuild.rule_engine.main.render_rules_progress import format_rules_progress
 from sqlbuild.rule_engine.main.run_rules import run_rules
 from sqlbuild.rule_engine.models import RulesCacheConfig, RulesConfig, RulesResult, RulesRunResult
 from sqlbuild.runtime.contracts.main.open_connection import open_connection_with_hooks
@@ -106,6 +106,7 @@ def run_compile_pipeline(
         rules_config = replace(rules_config, cache=RulesCacheConfig(enabled=False))
     if on_progress is not None:
         on_progress("Evaluating configured rules...")
+    rules_start: float = time.monotonic()
     rules_result: RulesRunResult = run_rules(
         graph=graph,
         discovered_inputs=discovered_inputs,
@@ -120,13 +121,12 @@ def run_compile_pipeline(
         no_sql_analysis=resolved_options.no_sql_validation,
     )
     if on_progress is not None:
-        skipped_rules_note: str | None = format_skipped_type_proof_rules(
-            codes=rules_result.skipped_type_proof_rules
-        )
         on_progress(
-            f"Evaluated rules. (built-in {rules_result.built_in_ms / 1000:.2f}s, "
-            f"custom {rules_result.custom_ms / 1000:.2f}s)"
-            + ("" if skipped_rules_note is None else f"; {skipped_rules_note}")
+            format_rules_progress(
+                elapsed_seconds=time.monotonic() - rules_start,
+                result=rules_result,
+                note_skipped_rules=True,
+            )
         )
     if rules_result.findings:
         rendered: str = format_result(

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

@@ -3,7 +3,8 @@
 from __future__ import annotations
 
 import re
-from collections.abc import Sequence
+from collections import ChainMap
+from collections.abc import Mapping, Sequence
 from functools import lru_cache
 from typing import Any
 
@@ -91,7 +92,7 @@ def build_relation_fixture_completion(
         authoritative_columns=context.authoritative_columns,
         adapter=adapter,
     )
-    expected_types: dict[FixtureKey, dict[str, str]] = _expected_fixture_types(
+    expected_types: Mapping[FixtureKey, dict[str, str]] = _expected_fixture_types(
         base_expected_types=context.expected_types,
         ordered_model_names=ordered_model_names,
         model_map=model_map,
@@ -327,7 +328,7 @@ def complete_empty_fixture_sql(
 def _typed_null_fixture_types(
     *,
     null_literal_names_by_fixture: dict[FixtureKey, frozenset[str]],
-    expected_types: dict[FixtureKey, dict[str, str]],
+    expected_types: Mapping[FixtureKey, dict[str, str]],
 ) -> dict[FixtureKey, dict[str, str]]:
     result: dict[FixtureKey, dict[str, str]] = {}
     for key, null_literal_names in null_literal_names_by_fixture.items():
@@ -665,8 +666,8 @@ def _expected_fixture_types(
     base_expected_types: dict[FixtureKey, dict[str, str]],
     ordered_model_names: tuple[str, ...],
     model_map: dict[str, CompiledModel],
-) -> dict[FixtureKey, dict[str, str]]:
-    expected_types: dict[FixtureKey, dict[str, str]] = dict(base_expected_types)
+) -> Mapping[FixtureKey, dict[str, str]]:
+    expected_types: ChainMap[FixtureKey, dict[str, str]] = ChainMap({}, base_expected_types)
     for model_name in ordered_model_names:
         model: CompiledModel | None = model_map.get(model_name)
         if model is None:

src/sqlbuild/compiler/planner/_helpers/identity/graph.py

@@ -4,6 +4,7 @@
 
 from collections.abc import Mapping
 
+from sqlbuild.compiler.planner.classes.graph_identity_resolver import GraphIdentityResolver
 from sqlbuild.compiler.planner.models import GraphIdentityNode, GraphNodeKey
 from sqlbuild.compiler.planner.types import GraphIdentityComposer
 
@@ -14,37 +15,13 @@ def build_expected_graph_identity_hashes(
     execution_order: tuple[GraphNodeKey, ...],
     compose_identity: GraphIdentityComposer,
 ) -> dict[GraphNodeKey, str | None]:
-    hashes: dict[GraphNodeKey, str | None] = {}
-
-    def resolve(
-        *,
-        key: GraphNodeKey,
-        visiting: frozenset[GraphNodeKey],
-        cache: dict[GraphNodeKey, str | None],
-    ) -> tuple[str | None, dict[GraphNodeKey, str | None]]:
-        if key in cache:
-            return cache[key], cache
-        node: GraphIdentityNode | None = nodes.get(key)
-        if node is None or node.local_hash is None:
-            return None, {**cache, key: None}
-        if key in visiting:
-            return node.local_hash, {**cache, key: node.local_hash}
-        upstream_hashes: list[tuple[GraphNodeKey, str]] = []
-        upstream_key: GraphNodeKey
-        for upstream_key in node.upstream_keys:
-            upstream_hash, cache = resolve(key=upstream_key, visiting=visiting | {key}, cache=cache)
-            if upstream_hash is not None:
-                upstream_hashes.append((upstream_key, upstream_hash))
-        composed: str = compose_identity(
-            local_hash=node.local_hash,
-            upstream_hashes=tuple(upstream_hashes),
-        )
-        return composed, {**cache, key: composed}
-
+    resolver: GraphIdentityResolver = GraphIdentityResolver(
+        nodes=nodes, compose_identity=compose_identity, base_hashes={}, selected_keys=None
+    )
     key: GraphNodeKey
     for key in execution_order:
-        _, hashes = resolve(key=key, visiting=frozenset(), cache=hashes)
-    return hashes
+        _ = resolver.resolve_expected(key)
+    return resolver.hashes
 
 
 def build_graph_write_identity_hashes(
@@ -55,48 +32,13 @@ def build_graph_write_identity_hashes(
     base_identity_hashes: Mapping[GraphNodeKey, str],
     compose_identity: GraphIdentityComposer,
 ) -> dict[GraphNodeKey, str]:
-    hashes: dict[GraphNodeKey, str] = dict(base_identity_hashes)
-    resolved_selected: dict[GraphNodeKey, str] = {}
-
-    def resolve(
-        *,
-        key: GraphNodeKey,
-        visiting: frozenset[GraphNodeKey],
-        cache: dict[GraphNodeKey, str],
-        selected_cache: dict[GraphNodeKey, str],
-    ) -> tuple[str | None, dict[GraphNodeKey, str], dict[GraphNodeKey, str]]:
-        if key not in selected_keys:
-            return cache.get(key), cache, selected_cache
-        if key in selected_cache:
-            return selected_cache[key], cache, selected_cache
-        node: GraphIdentityNode | None = nodes.get(key)
-        if node is None or node.local_hash is None:
-            return cache.get(key), cache, selected_cache
-        if key in visiting:
-            return node.local_hash, cache, selected_cache
-        upstream_hashes: list[tuple[GraphNodeKey, str]] = []
-        upstream_key: GraphNodeKey
-        for upstream_key in node.upstream_keys:
-            upstream_hash, cache, selected_cache = resolve(
-                key=upstream_key,
-                visiting=visiting | {key},
-                cache=cache,
-                selected_cache=selected_cache,
-            )
-            if upstream_hash is not None:
-                upstream_hashes.append((upstream_key, upstream_hash))
-        composed: str = compose_identity(
-            local_hash=node.local_hash,
-            upstream_hashes=tuple(upstream_hashes),
-        )
-        return composed, {**cache, key: composed}, {**selected_cache, key: composed}
-
+    resolver: GraphIdentityResolver = GraphIdentityResolver(
+        nodes=nodes,
+        compose_identity=compose_identity,
+        base_hashes=base_identity_hashes,
+        selected_keys=selected_keys,
+    )
     key: GraphNodeKey
     for key in execution_order:
-        _, hashes, resolved_selected = resolve(
-            key=key,
-            visiting=frozenset(),
-            cache=hashes,
-            selected_cache=resolved_selected,
-        )
-    return hashes
+        _ = resolver.resolve_write(key)
+    return {hashed_key: value for hashed_key, value in resolver.hashes.items() if value is not None}

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

@@ -147,6 +147,7 @@ def plan_model_migrations(
         adapter=runtime.adapter,
         connection=runtime.connection,
         database=planning_database(runtime=runtime),
+        known_fingerprints=snapshot.fingerprints,
     )
     declares_migrations: bool = any(
         _declared_request(model) is not None for model in runtime.project.models