perf(plan): cut plan and build planning time on large projects

.github/workflows/verify.yml

@@ -84,8 +84,8 @@ jobs:
                 broad=true
                 performance=true
                 ;;
-              src/sqlbuild/cli/commands/_helpers/lineage/*|src/sqlbuild/cli/commands/_helpers/plan/*|src/sqlbuild/cli/commands/_helpers/scope/*|src/sqlbuild/cli/commands/main/inspection/*|src/sqlbuild/cli/commands/main/project/*|src/sqlbuild/cli/output/*|src/sqlbuild/presentation/*)
-                # Inspection command guards run inside the fresh-process compile jobs.
+              src/sqlbuild/cli/commands/_helpers/lineage/*|src/sqlbuild/cli/commands/_helpers/plan/*|src/sqlbuild/cli/commands/_helpers/scope/*|src/sqlbuild/cli/commands/_helpers/build_planning/*|src/sqlbuild/cli/commands/main/inspection/*|src/sqlbuild/cli/commands/main/project/*|src/sqlbuild/cli/output/*|src/sqlbuild/executor/build/*|src/sqlbuild/executor/run/*|src/sqlbuild/presentation/*)
+                # Inspection, plan, and build guards run inside the fresh-process compile jobs.
                 broad=true
                 performance=true
                 ;;

Makefile

@@ -237,6 +237,7 @@ test-e2e-cold-compile-performance:
 	env PYTHONUNBUFFERED=1 SQLBUILD_CONCURRENCY=4 uv run pytest \
 		tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_performance.py \
 		tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py \
+		tests/e2e/src/sqlbuild/cli/commands/main/compile/test_build_performance.py \
 		-m cold_compile_performance -k "models_$(SQLBUILD_BENCHMARK_MODELS)" \
 		-vv --log-cli-level=INFO --color=yes
 

src/sqlbuild/cli/commands/_helpers/build_planning/invocation.py

@@ -35,7 +35,7 @@ def resolve_build_invocation(*, request: BuildCommandRequest) -> BuildInvocation
         request.project_dir if request.project_dir is not None else Path.cwd()
     )
     discovered_inputs: DiscoveredProjectInputs = discover_project_inputs(
-        project_dir=effective_project_dir
+        project_dir=effective_project_dir, extract_output_column_locations=False
     )
     effective_defer_clone_from: str | None = _resolve_defer_clone_from(
         discovered_inputs=discovered_inputs,

src/sqlbuild/cli/commands/_helpers/plan/invocation.py

@@ -32,7 +32,7 @@ def resolve_plan_invocation(*, request: PlanCommandRequest) -> PlanInvocation:
         request.project_dir if request.project_dir is not None else Path.cwd()
     )
     discovered_inputs: DiscoveredProjectInputs = discover_project_inputs(
-        project_dir=effective_project_dir
+        project_dir=effective_project_dir, extract_output_column_locations=False
     )
     _validate_preview_target(discovered_inputs=discovered_inputs, as_target=request.as_target)
     adapter_context: AdapterConnectionContext = resolve_adapter_connection_context(

src/sqlbuild/cli/commands/_helpers/plan/planning.py

@@ -53,6 +53,7 @@ def compile_plan_pipeline(
             resolve_python_run_selectors=request.include_python,
             max_microbatches=request.max_microbatches,
             selection_diagnostics=request.selection_diagnostics,
+            record_migration_fingerprints=False,
         ),
         hooks=ConnectionHooks(
             on_progress=invocation.planning_progress.on_progress,

src/sqlbuild/compiler/compile/_helpers/analysis/required_columns.py

@@ -217,7 +217,7 @@ def _resolve_set_operation_reads(
     branches: list[_RelationReads] = []
     required: list[CompiledLineageSourceFact] = []
     for side in ("left", "right"):
-        branch: Any | None = getattr(operation, "args", {}).get(side)
+        branch: Any | None = operation.arg(side)
         if branch is None:
             return _RelationReads(), _deduplicated_sources(required)
         branch_reads, branch_required = _resolve_query_reads(
@@ -228,7 +228,7 @@ def _resolve_set_operation_reads(
         branches.append(branch_reads)
         required.extend(branch_required)
     left, right = branches
-    if bool(getattr(operation, "args", {}).get("by_name")):
+    if bool(operation.arg("by_name")):
         right_by_name: dict[str, tuple[CompiledLineageSourceFact, ...]] = {
             name.casefold(): leaves for name, leaves in right.slots
         }

src/sqlbuild/compiler/compile/_helpers/assembly/source_bindings.py

@@ -82,7 +82,7 @@ def get_source_binding_diagnostics(
     for name, value in columns.items():
         if value:
             raw_shape: dict[str, str] = {column.name: column.type for column in value}
-            if raw_shape != shapes.get(name) or (
+            if _adds_physical_evidence(physical=raw_shape, known=shapes.get(name)) or (
                 profile.sql_analysis_dialect in CASE_SENSITIVE_BINDING_DIALECTS
                 and not profile.quoted_identifiers_ignore_case
             ):
@@ -202,6 +202,16 @@ def get_source_binding_diagnostics(
     return tuple(diagnostics)
 
 
+def _adds_physical_evidence(*, physical: dict[str, str], known: dict[str, str] | None) -> bool:
+    """Return whether inspected columns differ from the shape; untyped names match any type."""
+
+    if known is None or physical.keys() != known.keys():
+        return True
+    return any(
+        physical_type and physical_type != known[name] for name, physical_type in physical.items()
+    )
+
+
 def _placeholders(model: CompiledModel) -> dict[str, str] | None:
     values: object = model.config.values.get("placeholders")
     if isinstance(values, dict):

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

@@ -247,6 +247,7 @@ def _build_result(
             auto_load_sources=options.auto_load_sources,
             selection_diagnostics=options.selection_diagnostics,
             plan_sql_tests=options.plan_sql_tests,
+            record_migration_fingerprints=options.record_migration_fingerprints,
         ),
         on_progress=on_progress,
         project_config=discovered_inputs.project_config,

src/sqlbuild/compiler/pipeline/models.py

@@ -61,6 +61,7 @@ class CompilePipelineOptions:
     max_microbatches: int | None = None
     selection_diagnostics: bool = False
     plan_sql_tests: bool = True
+    record_migration_fingerprints: bool = True
 
 
 @dataclass(frozen=True)

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

@@ -11,7 +11,6 @@
 from sqlbuild.compiler.compile.main._analyze_columns_and_lineage import (
     analyze_resolved_column_reads,
 )
-from sqlbuild.compiler.compile.main._infer_fixture_columns import infer_fixture_column_facts
 from sqlbuild.compiler.compile.models import (
     CompiledLineageColumnFact,
     CompiledLineageSourceFact,
@@ -28,6 +27,7 @@
     ColumnTransformKind,
 )
 from sqlbuild.compiler.planner._helpers.fixtures.empty_fixture import is_empty_fixture_query
+from sqlbuild.compiler.planner.classes.fixture_column_inferences import FixtureColumnInferences
 from sqlbuild.compiler.planner.models import (
     FixtureColumnMetadata,
     FixtureRelationMetadata,
@@ -74,6 +74,7 @@ def build_relation_fixture_completion(
         fixture_groups=fixture_groups,
         relations=relations,
         adapter=adapter,
+        inferences=context.fixture_inferences,
     )
 
     fixture_keys_requiring_analysis: frozenset[FixtureKey] = _fixture_keys_requiring_analysis(
@@ -217,6 +218,7 @@ def _prepare_fixture_inputs(
     fixture_groups: FixtureGroups,
     relations: dict[FixtureKey, FixtureRelationMetadata],
     adapter: BaseAdapter,
+    inferences: FixtureColumnInferences,
 ) -> tuple[
     dict[FixtureKey, str],
     dict[FixtureKey, tuple[InferredColumn, ...]],
@@ -250,7 +252,7 @@ def _prepare_fixture_inputs(
                     continue
                 fixture_sql = empty_sql
             fixture_sql_by_key[key] = fixture_sql
-            inference: FixtureColumnInference | None = infer_fixture_column_facts(
+            inference: FixtureColumnInference | None = inferences.infer(
                 query_sql=fixture_sql,
                 inference_profile=adapter.expression_inference_profile(),
             )

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

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import heapq
+
 from sqlbuild.compiler.compile.models import (
     CompiledObjectKey,
     CompiledProject,
@@ -156,23 +158,22 @@ def topologically_order_keys(
     downstream: dict[CompiledObjectKey, tuple[CompiledObjectKey, ...]] = build_downstream_deps(
         upstream
     )
-    ready: list[CompiledObjectKey] = sorted(
-        (key for key, count in indegree.items() if count == 0),
-        key=_key_sort_key,
-    )
+    ready: list[tuple[tuple[str, str], CompiledObjectKey]] = [
+        (_key_sort_key(key), key) for key, count in indegree.items() if count == 0
+    ]
+    heapq.heapify(ready)
     ordered: list[CompiledObjectKey] = []
 
     while ready:
-        current: CompiledObjectKey = ready.pop(0)
+        current: CompiledObjectKey = heapq.heappop(ready)[1]
         ordered.append(current)
         downstream_key: CompiledObjectKey
         for downstream_key in downstream.get(current, ()):
             if downstream_key not in indegree:
                 continue
             indegree[downstream_key] -= 1
             if indegree[downstream_key] == 0:
-                ready.append(downstream_key)
-                ready.sort(key=_key_sort_key)
+                heapq.heappush(ready, (_key_sort_key(downstream_key), downstream_key))
 
     if len(ordered) != len(node_keys):
         ordered_set: set[CompiledObjectKey] = set(ordered)

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

@@ -19,7 +19,9 @@
 )
 from sqlbuild.compiler.migrations.models import MigrationEvent, MigrationRelation
 from sqlbuild.compiler.migrations.types import MigrationDecision, MigrationDiscovery
-from sqlbuild.compiler.planner._helpers.migrations.fingerprint import build_migration_fingerprint
+from sqlbuild.compiler.planner.classes.migration_fingerprint_cache import (
+    MigrationFingerprintCache,
+)
 from sqlbuild.compiler.planner.classes.migration_state_inspection import (
     MigrationStateInspection,
 )
@@ -62,6 +64,7 @@ def discover_model_migrations(
     declarations: tuple[ModelMigrationDeclaration, ...],
     state: MigrationStateInspection,
     project_schemas: set[str],
+    fingerprints: MigrationFingerprintCache,
 ) -> ModelMigrationDiscovery:
     """Add unique one-to-one fingerprint matches to the explicit migration requests."""
 
@@ -89,6 +92,7 @@ def discover_model_migrations(
                 variants=seed_variants,
                 metadata_jsons=metadata_jsons,
                 dialect=dialect,
+                fingerprints=fingerprints,
             ),
         )
     unbuilt: frozenset[str] = frozenset(
@@ -115,6 +119,7 @@ def discover_model_migrations(
         candidates=candidates,
         metadata_jsons=metadata_jsons,
         dialect=dialect,
+        fingerprints=fingerprints,
     )
     requests: tuple[ModelMigrationRequest, ...] = (
         *manual_requests,
@@ -126,7 +131,11 @@ def discover_model_migrations(
         requests=requests,
         warnings=_ambiguity_warnings(models=models, scope=scope, ambiguous=ambiguous),
         destination_fingerprints=_destination_fingerprints(
-            requests=requests, variants=variants, metadata_jsons=metadata_jsons, dialect=dialect
+            requests=requests,
+            variants=variants,
+            metadata_jsons=metadata_jsons,
+            dialect=dialect,
+            fingerprints=fingerprints,
         ),
     )
 
@@ -405,9 +414,12 @@ def _match(
     candidates: tuple[Fingerprint, ...],
     metadata_jsons: dict[str, str],
     dialect: str | None,
+    fingerprints: MigrationFingerprintCache,
 ) -> tuple[tuple[dict[str, str], ...], dict[str, Fingerprint], dict[str, str]]:
     """Match unbuilt models to candidates under every rename-map variant."""
 
+    if not candidates:
+        return tuple(dict(seed) for seed in seed_variants), {}, {}
     by_fingerprint: dict[str, list[Fingerprint]] = defaultdict(list)
     candidate: Fingerprint
     for candidate in candidates:
@@ -428,6 +440,7 @@ def _match(
                     metadata_json=metadata_jsons[model.name],
                     variants=variants,
                     dialect=dialect,
+                    fingerprints=fingerprints,
                 ),
                 by_fingerprint=by_fingerprint,
                 blocked=blocked,
@@ -460,19 +473,20 @@ def _variant_fingerprints(
     metadata_json: str,
     variants: tuple[dict[str, str], ...],
     dialect: str | None,
+    fingerprints: MigrationFingerprintCache,
 ) -> tuple[str, ...]:
     """Return the distinct migration fingerprints of one model under each rename map."""
 
-    fingerprints: list[str | None] = [
-        build_migration_fingerprint(
+    computed: list[str | None] = [
+        fingerprints.fingerprint(
             query_sql=model.query_sql,
             metadata_json=metadata_json,
             ref_identities=variant,
             dialect=dialect,
         )
         for variant in variants
     ]
-    return tuple(dict.fromkeys(item for item in fingerprints if item is not None))
+    return tuple(dict.fromkeys(item for item in computed if item is not None))
 
 
 def _match_options(
@@ -514,17 +528,22 @@ def _destination_fingerprints(
     variants: tuple[dict[str, str], ...],
     metadata_jsons: dict[str, str],
     dialect: str | None,
+    fingerprints: MigrationFingerprintCache,
 ) -> dict[str, tuple[str, ...]]:
-    fingerprints: dict[str, tuple[str, ...]] = {}
+    destinations: dict[str, tuple[str, ...]] = {}
     request: ModelMigrationRequest
     for request in requests:
         metadata_json: str | None = metadata_jsons.get(request.model.name)
         if metadata_json is None:
             continue
-        fingerprints[request.model.name] = _variant_fingerprints(
-            model=request.model, metadata_json=metadata_json, variants=variants, dialect=dialect
+        destinations[request.model.name] = _variant_fingerprints(
+            model=request.model,
+            metadata_json=metadata_json,
+            variants=variants,
+            dialect=dialect,
+            fingerprints=fingerprints,
         )
-    return fingerprints
+    return destinations
 
 
 def _moves_history(*, model: CompiledModel, origin: Fingerprint) -> bool:

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

@@ -0,0 +1,48 @@
+"""Attach the migration fingerprint each built model stores with its fingerprint row."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import replace
+
+from sqlbuild.compiler.compile.models import CompiledModel
+from sqlbuild.compiler.planner.classes.migration_fingerprint_cache import (
+    MigrationFingerprintCache,
+)
+from sqlbuild.compiler.planner.models import ModelPlanEntry
+from sqlbuild.compiler.planner.types import PlanAction
+
+
+def with_migration_fingerprints(
+    *,
+    entries: tuple[ModelPlanEntry, ...],
+    models_by_name: Mapping[str, CompiledModel],
+    fingerprints: MigrationFingerprintCache,
+    dialect: str | None,
+) -> tuple[ModelPlanEntry, ...]:
+    """Attach current-name migration fingerprints to the entries that will be built."""
+
+    attached: list[ModelPlanEntry] = []
+    entry: ModelPlanEntry
+    for entry in entries:
+        model: CompiledModel | None = models_by_name.get(entry.name)
+        if (
+            model is None
+            or entry.fingerprint_metadata_json is None
+            or entry.action == PlanAction.SKIP
+        ):
+            attached.append(entry)
+            continue
+        attached.append(
+            replace(
+                entry,
+                migration_fingerprint=fingerprints.fingerprint(
+                    query_sql=model.query_sql,
+                    metadata_json=entry.fingerprint_metadata_json,
+                    ref_identities={},
+                    dialect=dialect,
+                ),
+            )
+        )
+    fingerprints.persist()
+    return tuple(attached)