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

* perf(planner): read fingerprint state once per plan

Full-project plans and projects declaring migrations read the latest
fingerprint per identity without a node filter, apply the selection in
memory, and share the rows with migration inspection instead of reading
the state table a second time. Narrow selections keep the filtered read.

* perf(observability): cut per-statement monitoring overhead

Statement monitors without a query-ID provider now start their thread
only once a statement outlives the heartbeat threshold, through one
shared deferred-start daemon, instead of starting and joining a thread
for every statement. Lifecycle events are validated once at
construction rather than again on publish and on sequencing, JSON
freezing builds error paths only on failure, and statements hash their
SQL once. Recorded events, diagnostics, and -v output are unchanged.

* perf(planner): resolve graph identity hashes in linear time

Expected and write identity resolution copied the whole hash cache for
every resolved node, which made identity context building and change
reconciliation quadratic in project size. A memoized resolver now
records each node once with the same resolution order, cycle handling,
and results.

* perf(planner): overlay per-test fixture types instead of copying them

SQL-test fixture completion copied the project-wide expected-type map
for every test, which made planning test-heavy projects quadratic. Each
test now overlays only the models of its chain on the shared map.

* fix(cli): report rules phase wall time on its progress line

The rules completion line reported only built-in and custom rule time,
which run concurrently and are not phase wall time, so users could not
see how long the rules phase took next to the compile line. It now leads
with the phase wall time in the same form as other phase lines.

* test(planner): guard plan phase scaling on the inspection benchmark

Plan the inspection benchmark at 1,000 and 3,000 models without the
compile cache and fail when warehouse inspection or plan generation
grows more than 1.5 times faster than the project, or when either
phase exceeds a loose ceiling.

* fix(observability): release finished statements from deferred monitor starts

A stopped monitor now cancels its pending deferred start and drops its
callbacks and query-ID provider, so a finished statement and its SQL are
no longer retained until the heartbeat deadline. A monitor thread is
recorded only after it starts, so a refused start no longer makes stop()
join an unstarted thread and fail an otherwise successful statement.

* test(pipeline): account for the rules wall-time clock read in phase timings

The rules progress line reads the clock once before rules run, between
the compile and planning windows. The phase-timing test now supplies
that read, gives rules a measurable duration, and proves compile and
planning stay disjoint from it, with and without progress reporting.

Author: kvlonge

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