diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index 86739641ba..a08a616a8f 100644 --- a/.github/workflows/verify.yml +++ b/.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 ;; diff --git a/Makefile b/Makefile index c2513fe738..e0dcabcf55 100644 --- a/Makefile +++ b/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 diff --git a/src/sqlbuild/cli/commands/_helpers/build_planning/invocation.py b/src/sqlbuild/cli/commands/_helpers/build_planning/invocation.py index d53eb30a41..781c337405 100644 --- a/src/sqlbuild/cli/commands/_helpers/build_planning/invocation.py +++ b/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, diff --git a/src/sqlbuild/cli/commands/_helpers/plan/invocation.py b/src/sqlbuild/cli/commands/_helpers/plan/invocation.py index 26479888f1..1eb1c37796 100644 --- a/src/sqlbuild/cli/commands/_helpers/plan/invocation.py +++ b/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( diff --git a/src/sqlbuild/cli/commands/_helpers/plan/planning.py b/src/sqlbuild/cli/commands/_helpers/plan/planning.py index 58fa03ed69..41fd7ca62b 100644 --- a/src/sqlbuild/cli/commands/_helpers/plan/planning.py +++ b/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, diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/required_columns.py b/src/sqlbuild/compiler/compile/_helpers/analysis/required_columns.py index 95f06c4702..32c7deb5b4 100644 --- a/src/sqlbuild/compiler/compile/_helpers/analysis/required_columns.py +++ b/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 } diff --git a/src/sqlbuild/compiler/compile/_helpers/assembly/source_bindings.py b/src/sqlbuild/compiler/compile/_helpers/assembly/source_bindings.py index 1e0f573885..df2eb07e2d 100644 --- a/src/sqlbuild/compiler/compile/_helpers/assembly/source_bindings.py +++ b/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): diff --git a/src/sqlbuild/compiler/pipeline/main/compile.py b/src/sqlbuild/compiler/pipeline/main/compile.py index 89643c926c..1a4f08017c 100644 --- a/src/sqlbuild/compiler/pipeline/main/compile.py +++ b/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, diff --git a/src/sqlbuild/compiler/pipeline/models.py b/src/sqlbuild/compiler/pipeline/models.py index 6198d29e43..ae91eaec2d 100644 --- a/src/sqlbuild/compiler/pipeline/models.py +++ b/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) diff --git a/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py b/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py index 1df5659c55..d95c9230d8 100644 --- a/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py +++ b/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(), ) diff --git a/src/sqlbuild/compiler/planner/_helpers/graph/core.py b/src/sqlbuild/compiler/planner/_helpers/graph/core.py index 55d67821ec..c88875c2da 100644 --- a/src/sqlbuild/compiler/planner/_helpers/graph/core.py +++ b/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,14 +158,14 @@ 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, ()): @@ -171,8 +173,7 @@ def topologically_order_keys( 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) diff --git a/src/sqlbuild/compiler/planner/_helpers/migrations/discovery.py b/src/sqlbuild/compiler/planner/_helpers/migrations/discovery.py index 8b69f97752..d0a0c7daf2 100644 --- a/src/sqlbuild/compiler/planner/_helpers/migrations/discovery.py +++ b/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,11 +473,12 @@ 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, @@ -472,7 +486,7 @@ def _variant_fingerprints( ) 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: diff --git a/src/sqlbuild/compiler/planner/_helpers/migrations/entry_fingerprints.py b/src/sqlbuild/compiler/planner/_helpers/migrations/entry_fingerprints.py new file mode 100644 index 0000000000..36704edea2 --- /dev/null +++ b/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) diff --git a/src/sqlbuild/compiler/planner/_helpers/migrations/fingerprint.py b/src/sqlbuild/compiler/planner/_helpers/migrations/fingerprint.py index f47906b769..b41baeb9a2 100644 --- a/src/sqlbuild/compiler/planner/_helpers/migrations/fingerprint.py +++ b/src/sqlbuild/compiler/planner/_helpers/migrations/fingerprint.py @@ -6,11 +6,9 @@ import json import re from collections.abc import Mapping -from dataclasses import replace from typing import Any from sqlbuild.compiler.compile.constants import CURSOR_INPUTS_CONFIG_KEY -from sqlbuild.compiler.compile.models import CompiledModel from sqlbuild.compiler.planner._helpers.changes.metadata import version_identity_metadata_payload from sqlbuild.compiler.planner.constants import ( MIGRATION_FINGERPRINT_EXCLUDED_CONFIG_KEYS, @@ -18,7 +16,6 @@ MIGRATION_MODEL_NAME_METADATA_KEY, MIGRATION_REF_PLACEHOLDER_PREFIX, ) -from sqlbuild.compiler.planner.models import ModelPlanEntry from sqlbuild.compiler.sql_analysis.main.import_polyglot_sql import import_polyglot_sql _REF_MARKER: re.Pattern[str] = re.compile(r"""__ref\(\s*(?:"([^"]+)"|'([^']+)')\s*\)""") @@ -67,6 +64,28 @@ def build_migration_fingerprint( return hashlib.sha256(payload.encode("utf-8")).hexdigest() +def migration_fingerprint_ref_names(*, query_sql: str, metadata_json: str) -> frozenset[str]: + """Return every model name whose rename identity can change the migration fingerprint.""" + + names: set[str] = { + match.group(1) or match.group(2) for match in _REF_MARKER.finditer(query_sql) + } + payload: Any = version_identity_metadata_payload(metadata_json) + config: Any = ( + {str(key): value for key, value in payload.items()}.get(_CONFIG_KEY) + if isinstance(payload, dict) + else None + ) + cursor_inputs: Any = ( + {str(key): value for key, value in config.items()}.get(CURSOR_INPUTS_CONFIG_KEY) + if isinstance(config, dict) + else None + ) + if isinstance(cursor_inputs, dict): + names.update(str(name) for name in cursor_inputs) + return frozenset(names) + + def _identity_metadata(*, metadata_json: str, ref_identities: Mapping[str, str]) -> Any: payload: Any = version_identity_metadata_payload(metadata_json) if not isinstance(payload, dict): @@ -111,9 +130,9 @@ def _normalized_query_sql( except polyglot.PolyglotError: return None stripped: Any = strip_ast_formatting(parsed) - if not _has_unambiguous_local_scopes(stripped): - return None definitions: list[tuple[str, bool]] = _local_names(stripped) + if not _has_unambiguous_local_scopes(root=stripped, definitions=definitions): + return None ordered_names: list[str] = list(dict.fromkeys(name for name, _ in definitions)) local_names: dict[str, str] = { name: f"{MIGRATION_LOCAL_NAME_PREFIX}{index}" @@ -144,7 +163,7 @@ def strip_ast_formatting(node: Any) -> Any: return node -def _has_unambiguous_local_scopes(root: Any) -> bool: +def _has_unambiguous_local_scopes(*, root: Any, definitions: list[tuple[str, bool]]) -> bool: """Accept one top-level WITH whose names never shadow physical relations.""" statements: list[Any] = list(root.values()) if isinstance(root, dict) else [] @@ -166,7 +185,7 @@ def _has_unambiguous_local_scopes(root: Any) -> bool: if set(_relation_names(cte.get(_CTE_BODY_KEY))) & set(cte_names[index:]): return False physical: set[str] = set(_relation_names(root)) - set(cte_names) - aliases: set[str] = {name for name, is_cte in _local_names(root) if not is_cte} + aliases: set[str] = {name for name, is_cte in definitions if not is_cte} return not physical & aliases @@ -314,32 +333,3 @@ def _is_identifier(node: Any) -> bool: and isinstance(node.get(_IDENTIFIER_NAME_KEY), str) and _IDENTIFIER_QUOTED_KEY in node ) - - -def with_migration_fingerprints( - *, - entries: tuple[ModelPlanEntry, ...], - models_by_name: Mapping[str, CompiledModel], - dialect: str | None, -) -> tuple[ModelPlanEntry, ...]: - """Attach the current-name migration fingerprint each build stores with its fingerprint.""" - - 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: - attached.append(entry) - continue - attached.append( - replace( - entry, - migration_fingerprint=build_migration_fingerprint( - query_sql=model.query_sql, - metadata_json=entry.fingerprint_metadata_json, - ref_identities={}, - dialect=dialect, - ), - ) - ) - return tuple(attached) diff --git a/src/sqlbuild/compiler/planner/_helpers/migrations/planning.py b/src/sqlbuild/compiler/planner/_helpers/migrations/planning.py index 7508bebf58..e759fa032b 100644 --- a/src/sqlbuild/compiler/planner/_helpers/migrations/planning.py +++ b/src/sqlbuild/compiler/planner/_helpers/migrations/planning.py @@ -36,6 +36,9 @@ from sqlbuild.compiler.planner._helpers.warehouse.snapshot import ( gather_redirected_cursor_snapshots, ) +from sqlbuild.compiler.planner.classes.migration_fingerprint_cache import ( + MigrationFingerprintCache, +) from sqlbuild.compiler.planner.classes.migration_state_inspection import ( MigrationStateInspection, ) @@ -130,6 +133,7 @@ def plan_model_migrations( snapshot: WarehouseSnapshot, overrides: PlannerOverrides, deferral: DeferralInputs, + fingerprints: MigrationFingerprintCache, ) -> ModelMigrationPlanning: """Decide every declared or discovered migration and project the post-migration snapshot.""" @@ -153,6 +157,7 @@ def plan_model_migrations( declarations=_project_declarations(runtime=runtime, state=state), state=state, project_schemas=schemas, + fingerprints=fingerprints, ) if not discovery.requests: return ModelMigrationPlanning(snapshot=snapshot, warnings=discovery.warnings) diff --git a/src/sqlbuild/compiler/planner/_helpers/planning/output_assembly.py b/src/sqlbuild/compiler/planner/_helpers/planning/output_assembly.py index 8189e2536d..0a1ad90d5e 100644 --- a/src/sqlbuild/compiler/planner/_helpers/planning/output_assembly.py +++ b/src/sqlbuild/compiler/planner/_helpers/planning/output_assembly.py @@ -5,7 +5,9 @@ from dataclasses import replace from sqlbuild.compiler.compile.types import CompiledResourceType -from sqlbuild.compiler.planner._helpers.migrations.fingerprint import with_migration_fingerprints +from sqlbuild.compiler.planner._helpers.migrations.entry_fingerprints import ( + with_migration_fingerprints, +) from sqlbuild.compiler.planner._helpers.output.plan_output import build_plan_output from sqlbuild.compiler.planner._helpers.planning.retention import plan_retention, plan_table_types from sqlbuild.compiler.planner._helpers.pruning.selection_staleness import ( @@ -80,6 +82,7 @@ def with_storage_policies( runtime: PlannerRuntime, warehouse: PlannerWarehouseState, scopes: PlannerScopeResolution, + policies: PlannerPolicies, ) -> PlanOutput: """Attach migrations and selected Snowflake table-type and retention work to the plan.""" @@ -89,8 +92,11 @@ def with_storage_policies( entries=with_migration_fingerprints( entries=plan_output.model_entries, models_by_name={model.name: model for model in runtime.project.models}, + fingerprints=warehouse.migration_fingerprints, dialect=runtime.adapter.sql_analysis_dialect(), - ), + ) + if policies.record_migration_fingerprints + else plan_output.model_entries, hints=warehouse.column_rename_hints, ), migration_entries=warehouse.migration_entries, diff --git a/src/sqlbuild/compiler/planner/_helpers/planning/warehouse_state.py b/src/sqlbuild/compiler/planner/_helpers/planning/warehouse_state.py index 4ac3230fa9..d8da979bcd 100644 --- a/src/sqlbuild/compiler/planner/_helpers/planning/warehouse_state.py +++ b/src/sqlbuild/compiler/planner/_helpers/planning/warehouse_state.py @@ -12,6 +12,9 @@ effectively_full_refreshed_model_names, ) from sqlbuild.compiler.planner._helpers.warehouse.snapshot import gather_warehouse_snapshot +from sqlbuild.compiler.planner.classes.migration_fingerprint_cache import ( + MigrationFingerprintCache, +) from sqlbuild.compiler.planner.models import ( ColumnMigrationPlanning, CursorSnapshotScope, @@ -37,6 +40,9 @@ def gather_planner_warehouse_state( """Gather the warehouse snapshot and inspection relations in one pass.""" warehouse_start: float = time.monotonic() + fingerprints: MigrationFingerprintCache = MigrationFingerprintCache( + root=runtime.project.compile_cache_dir + ) if runtime.on_progress is not None: runtime.on_progress("Inspecting warehouse state...") snapshot: WarehouseSnapshot = gather_warehouse_snapshot( @@ -67,7 +73,9 @@ def gather_planner_warehouse_state( snapshot=snapshot, overrides=overrides, deferral=deferral, + fingerprints=fingerprints, ) + fingerprints.persist() inspection_relations: PlannerRelationsContext = build_planner_relations_context( project=runtime.project, adapter=runtime.adapter, @@ -115,4 +123,5 @@ def gather_planner_warehouse_state( column_rename_hints=columns.hints, old_name_view_entries=old_names.entries, old_name_views=old_names.views, + migration_fingerprints=fingerprints, ) diff --git a/src/sqlbuild/compiler/planner/_helpers/sql_tests/fixture_validation.py b/src/sqlbuild/compiler/planner/_helpers/sql_tests/fixture_validation.py index 95e88e958e..5d79332892 100644 --- a/src/sqlbuild/compiler/planner/_helpers/sql_tests/fixture_validation.py +++ b/src/sqlbuild/compiler/planner/_helpers/sql_tests/fixture_validation.py @@ -5,7 +5,6 @@ import re from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter -from sqlbuild.compiler.compile.main._infer_fixture_columns import infer_fixture_column_facts from sqlbuild.compiler.compile.models import ( CompiledProject, CompiledSqlTest, @@ -84,7 +83,7 @@ def build_validated_test_fixtures( f"expected output '{name}' {empty_error}" ) completed_expected_outputs[name] = completed_sql - inference: FixtureColumnInference | None = infer_fixture_column_facts( + inference: FixtureColumnInference | None = fixture_context.fixture_inferences.infer( query_sql=completed_sql, inference_profile=adapter.expression_inference_profile(), ) diff --git a/src/sqlbuild/compiler/planner/classes/fixture_column_inferences.py b/src/sqlbuild/compiler/planner/classes/fixture_column_inferences.py new file mode 100644 index 0000000000..0bd817703e --- /dev/null +++ b/src/sqlbuild/compiler/planner/classes/fixture_column_inferences.py @@ -0,0 +1,46 @@ +"""Fixture column inference computed once per fixture body and inference profile.""" + +from __future__ import annotations + +from sqlbuild.adapter.contract.models import ExpressionInferenceProfile +from sqlbuild.compiler.compile.main._infer_fixture_columns import infer_fixture_column_facts +from sqlbuild.compiler.compile.models import FixtureColumnInference + + +class FixtureColumnInferences: + """Reuse inference for identical fixture SQL under an equal, catalog-free profile.""" + + def __init__(self) -> None: + self._by_profile: list[ + tuple[ExpressionInferenceProfile, dict[str, FixtureColumnInference | None]] + ] = [] + + def infer( + self, *, query_sql: str, inference_profile: ExpressionInferenceProfile + ) -> FixtureColumnInference | None: + """Return the inference for one fixture query, computing each distinct query once.""" + + if inference_profile.binding_catalog is not None: + return infer_fixture_column_facts( + query_sql=query_sql, inference_profile=inference_profile + ) + inferred: dict[str, FixtureColumnInference | None] = self._profile_results( + inference_profile + ) + if query_sql not in inferred: + inferred[query_sql] = infer_fixture_column_facts( + query_sql=query_sql, inference_profile=inference_profile + ) + return inferred[query_sql] + + def _profile_results( + self, inference_profile: ExpressionInferenceProfile + ) -> dict[str, FixtureColumnInference | None]: + profile: ExpressionInferenceProfile + results: dict[str, FixtureColumnInference | None] + for profile, results in self._by_profile: + if profile == inference_profile: + return results + results = {} + self._by_profile.append((inference_profile, results)) + return results diff --git a/src/sqlbuild/compiler/planner/classes/migration_fingerprint_cache.py b/src/sqlbuild/compiler/planner/classes/migration_fingerprint_cache.py new file mode 100644 index 0000000000..a8b7454141 --- /dev/null +++ b/src/sqlbuild/compiler/planner/classes/migration_fingerprint_cache.py @@ -0,0 +1,232 @@ +"""Migration fingerprints computed at most once per model definition and reused across runs.""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import sqlite3 +from collections.abc import Mapping +from importlib.metadata import PackageNotFoundError, version +from pathlib import Path +from typing import Any, cast + +from sqlbuild.compiler.planner._helpers.migrations.fingerprint import ( + build_migration_fingerprint, + migration_fingerprint_ref_names, +) +from sqlbuild.compiler.planner.constants import MIGRATION_FINGERPRINT_ALGORITHM + +type _FingerprintKey = tuple[str, str, str | None, tuple[tuple[str, str], ...]] + +_CACHE_VERSION: int = 1 +_CACHE_DIRECTORY_NAME: str = f"migration-fingerprints-v{_CACHE_VERSION}" +_CACHE_DATABASE_NAME: str = "migration-fingerprints.sqlite3" +_CACHE_ENTRY_SEPARATOR: str = "\n" +_MAX_CACHE_ENTRY_BYTES: int = 4_096 +_SQLITE_TIMEOUT_SECONDS: float = 0.1 +_CREATE_CACHE_TABLE_SQL: str = """ +CREATE TABLE IF NOT EXISTS migration_fingerprint ( + cache_key TEXT PRIMARY KEY, + payload TEXT NOT NULL +) +""" + + +class MigrationFingerprintCache: + """Reuse migration fingerprints across rename maps, planning phases, and invocations.""" + + def __init__(self, *, root: Path | None = None) -> None: + self._ref_names: dict[tuple[str, str], frozenset[str]] = {} + self._fingerprints: dict[_FingerprintKey, str | None] = {} + self._database_path: Path | None = ( + None if root is None else root / _CACHE_DIRECTORY_NAME / _CACHE_DATABASE_NAME + ) + self._connection: sqlite3.Connection | None = None + self._connection_opened: bool = False + self._pending: dict[str, str] = {} + self._shared_digest: str = _shared_digest() if root is not None else "" + + def fingerprint( + self, + *, + query_sql: str, + metadata_json: str, + ref_identities: Mapping[str, str], + dialect: str | None, + ) -> str | None: + """Return the migration fingerprint, keyed only by the renames this model can see.""" + + definition: tuple[str, str] = (query_sql, metadata_json) + names: frozenset[str] | None = self._ref_names.get(definition) + if names is None: + names = migration_fingerprint_ref_names( + query_sql=query_sql, metadata_json=metadata_json + ) + self._ref_names[definition] = names + visible: dict[str, str] = { + name: ref_identities[name] for name in sorted(names) if name in ref_identities + } + key: _FingerprintKey = (query_sql, metadata_json, dialect, tuple(visible.items())) + if key in self._fingerprints: + return self._fingerprints[key] + cache_key: str = self._cache_key(key) + found: bool + stored: str | None + found, stored = self._stored(cache_key) + if not found: + stored = build_migration_fingerprint( + query_sql=query_sql, + metadata_json=metadata_json, + ref_identities=visible, + dialect=dialect, + ) + self._remember(cache_key=cache_key, fingerprint=stored) + self._fingerprints[key] = stored + return stored + + def persist(self) -> None: + """Write fingerprints computed since the last persist; failures only skip reuse.""" + + pending: dict[str, str] = self._pending + self._pending = {} + connection: sqlite3.Connection | None = self._connection + self._connection = None + self._connection_opened = False + try: + if pending and self._database_path is not None: + if connection is None: + self._database_path.parent.mkdir(parents=True, exist_ok=True) + connection = sqlite3.connect( + self._database_path, timeout=_SQLITE_TIMEOUT_SECONDS + ) + _ = connection.execute(_CREATE_CACHE_TABLE_SQL) + _ = connection.executemany( + "INSERT OR REPLACE INTO migration_fingerprint (cache_key, payload) " + "VALUES (?, ?)", + pending.items(), + ) + connection.commit() + except (OSError, sqlite3.DatabaseError): + pass + finally: + _close(connection) + + def _cache_key(self, key: _FingerprintKey) -> str: + if self._database_path is None: + return "" + return hashlib.sha256( + json.dumps( + [self._shared_digest, *key[:3], [list(item) for item in key[3]]], + ensure_ascii=False, + separators=(",", ":"), + ).encode() + ).hexdigest() + + def _stored(self, cache_key: str) -> tuple[bool, str | None]: + connection: sqlite3.Connection | None = self._open() + if connection is None: + return False, None + try: + row: tuple[object] | None = connection.execute( + "SELECT payload FROM migration_fingerprint WHERE cache_key = ?", (cache_key,) + ).fetchone() + except sqlite3.DatabaseError: + self._connection = None + _close(connection) + return False, None + return ( + (False, None) + if row is None + else _fingerprint_from_contents(contents=row[0], expected_cache_key=cache_key) + ) + + def _open(self) -> sqlite3.Connection | None: + if self._connection_opened: + return self._connection + self._connection_opened = True + if self._database_path is None or not self._database_path.is_file(): + return None + try: + self._connection = sqlite3.connect(self._database_path, timeout=_SQLITE_TIMEOUT_SECONDS) + except (OSError, sqlite3.DatabaseError): + self._connection = None + return self._connection + + def _remember(self, *, cache_key: str, fingerprint: str | None) -> None: + if self._database_path is None: + return + serialized: str = json.dumps( + {"v": _CACHE_VERSION, "k": cache_key, "f": fingerprint}, + separators=(",", ":"), + sort_keys=True, + ) + self._pending[cache_key] = _CACHE_ENTRY_SEPARATOR.join( + (_entry_digest(cache_key=cache_key, serialized=serialized), serialized) + ) + + +def _fingerprint_from_contents( + *, contents: object, expected_cache_key: str +) -> tuple[bool, str | None]: + if not isinstance(contents, str) or len(contents) > _MAX_CACHE_ENTRY_BYTES: + return False, None + stored_digest, separator, serialized = contents.partition(_CACHE_ENTRY_SEPARATOR) + if ( + not separator + or not stored_digest.isascii() + or not hmac.compare_digest( + stored_digest, _entry_digest(cache_key=expected_cache_key, serialized=serialized) + ) + ): + return False, None + try: + payload: object = json.loads(serialized) + except (ValueError, RecursionError): + return False, None + values: dict[str, Any] = cast(dict[str, Any], payload) if isinstance(payload, dict) else {} + fingerprint: object = values.get("f") + if ( + values.get("v") != _CACHE_VERSION + or values.get("k") != expected_cache_key + or not (fingerprint is None or isinstance(fingerprint, str)) + ): + return False, None + return True, cast(str | None, fingerprint) + + +def _entry_digest(*, cache_key: str, serialized: str) -> str: + digest: Any = hashlib.sha256(cache_key.encode()) + digest.update(b"\0") + digest.update(serialized.encode()) + return digest.hexdigest() + + +def _shared_digest() -> str: + return hashlib.sha256( + json.dumps( + { + "algorithm": MIGRATION_FINGERPRINT_ALGORITHM, + "cache_version": _CACHE_VERSION, + "sqlbuild_version": _package_version("sqlbuild"), + "polyglot_version": _package_version("polyglot-sql-chio"), + }, + sort_keys=True, + ).encode() + ).hexdigest() + + +def _package_version(package: str) -> str: + try: + return version(package) + except PackageNotFoundError: + return "unknown" + + +def _close(connection: sqlite3.Connection | None) -> None: + if connection is None: + return + try: + connection.close() + except sqlite3.DatabaseError: + pass diff --git a/src/sqlbuild/compiler/planner/constants.py b/src/sqlbuild/compiler/planner/constants.py index 5e5bdc928b..991de00b28 100644 --- a/src/sqlbuild/compiler/planner/constants.py +++ b/src/sqlbuild/compiler/planner/constants.py @@ -62,6 +62,7 @@ MIGRATION_FINGERPRINT_METADATA_KEY: str = "migration_fingerprint" MIGRATION_REF_PLACEHOLDER_PREFIX: str = "__sqb_migration_ref__" MIGRATION_LOCAL_NAME_PREFIX: str = "__sqb_local_" +MIGRATION_FINGERPRINT_ALGORITHM: str = "migration-fingerprint-v1" MIGRATION_FINGERPRINT_EXCLUDED_CONFIG_KEYS: frozenset[str] = frozenset( { "batch_size", diff --git a/src/sqlbuild/compiler/planner/main/execution/execution.py b/src/sqlbuild/compiler/planner/main/execution/execution.py index 70fe0cb496..3348cf14a0 100644 --- a/src/sqlbuild/compiler/planner/main/execution/execution.py +++ b/src/sqlbuild/compiler/planner/main/execution/execution.py @@ -166,7 +166,11 @@ def build_execution_plan( planned_sql_tests=planned_sql_tests, ) plan_output = with_storage_policies( - plan_output=plan_output, runtime=runtime, warehouse=warehouse, scopes=scopes + plan_output=plan_output, + runtime=runtime, + warehouse=warehouse, + scopes=scopes, + policies=policies, ) plan_output = with_plan_warnings( runtime=runtime, diff --git a/src/sqlbuild/compiler/planner/models.py b/src/sqlbuild/compiler/planner/models.py index b7d9aa9678..9a715e90db 100644 --- a/src/sqlbuild/compiler/planner/models.py +++ b/src/sqlbuild/compiler/planner/models.py @@ -39,6 +39,10 @@ MigrationPromotion, OldNameViewAction, ) +from sqlbuild.compiler.planner.classes.fixture_column_inferences import FixtureColumnInferences +from sqlbuild.compiler.planner.classes.migration_fingerprint_cache import ( + MigrationFingerprintCache, +) from sqlbuild.compiler.planner.exceptions import PlannerInputError from sqlbuild.compiler.planner.types import ( BackfillAction, @@ -144,6 +148,9 @@ class RelationFixturePlanningContext: relations: dict[FixtureKey, FixtureRelationMetadata] authoritative_columns: dict[FixtureKey, frozenset[str]] expected_types: dict[FixtureKey, dict[str, str]] + fixture_inferences: FixtureColumnInferences = field( + default_factory=FixtureColumnInferences, compare=False + ) @dataclass(frozen=True) @@ -1538,6 +1545,7 @@ class PlannerPolicies: auto_load_sources: bool = False selection_diagnostics: bool = False plan_sql_tests: bool = True + record_migration_fingerprints: bool = True @dataclass(frozen=True) @@ -1574,6 +1582,9 @@ class PlannerWarehouseState: column_rename_hints: tuple[ColumnRenameHint, ...] = () old_name_view_entries: tuple[OldNameViewPlanEntry, ...] = () old_name_views: tuple[OldNameView, ...] = () + migration_fingerprints: MigrationFingerprintCache = field( + default_factory=MigrationFingerprintCache + ) @dataclass(frozen=True) diff --git a/src/sqlbuild/runtime/compute_logs/classes/diagnostic_log_handler.py b/src/sqlbuild/runtime/compute_logs/classes/diagnostic_log_handler.py index 5ac6792940..45e6f11387 100644 --- a/src/sqlbuild/runtime/compute_logs/classes/diagnostic_log_handler.py +++ b/src/sqlbuild/runtime/compute_logs/classes/diagnostic_log_handler.py @@ -5,6 +5,7 @@ import logging from collections.abc import Callable from datetime import UTC, datetime +from functools import cache from importlib.metadata import PackageNotFoundError, version from typing import cast @@ -146,6 +147,7 @@ def _severity(level: int) -> str: return "debug" +@cache def _sqlbuild_version() -> str: try: return version("sqlbuild") diff --git a/src/sqlbuild/runtime/observability/_helpers/factory.py b/src/sqlbuild/runtime/observability/_helpers/factory.py index 9d926721b5..b1bc0e5631 100644 --- a/src/sqlbuild/runtime/observability/_helpers/factory.py +++ b/src/sqlbuild/runtime/observability/_helpers/factory.py @@ -2,6 +2,7 @@ from collections.abc import Mapping from datetime import UTC, datetime +from functools import cache from importlib.metadata import version from types import MappingProxyType from uuid import uuid4 @@ -41,9 +42,7 @@ def create_lifecycle_event( event_type=event_type, schema_version=CURRENT_LIFECYCLE_EVENT_SCHEMA_VERSION, producer=producer, - producer_version=( - version(_DEFAULT_PRODUCER) if producer_version is None else producer_version - ), + producer_version=(_sqlbuild_version() if producer_version is None else producer_version), occurred_at=datetime.now(UTC) if occurred_at is None else occurred_at, invocation_id=identity.invocation_id, invocation_sequence=0, @@ -55,3 +54,8 @@ def create_lifecycle_event( external_context=current_invocation_external_context(), payload=payload, ) + + +@cache +def _sqlbuild_version() -> str: + return version(_DEFAULT_PRODUCER) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_runtime_artifact_preservation.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_runtime_artifact_preservation.py index 28f254fb39..6f18ac6227 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_runtime_artifact_preservation.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_runtime_artifact_preservation.py @@ -32,7 +32,7 @@ "target/run/functions/python/is_completed_order_py.sql", ), expected_exit_code=0, - expected_compiler_cache_count=2, + expected_compiler_cache_count=3, expected_local_history=False, expected_compiled_paths=( "target/compiled/models/marts/fact_orders.sql", diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/COMPILE_BENCHMARKS.md b/tests/e2e/src/sqlbuild/cli/commands/main/compile/COMPILE_BENCHMARKS.md index 92bf38481a..65fb6bec1a 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/COMPILE_BENCHMARKS.md +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/COMPILE_BENCHMARKS.md @@ -69,3 +69,17 @@ each in a fresh process. Every command has a wall-time ceiling, a peak-RSS ceiling and an output-line bound linear in the model count, so a renderer or graph walk that re-expands shared nodes per path fails on size before it can hide behind a loose timing budget. It runs in the 3,000-model fresh-process compile job. + +## Plan and build guards + +The same inspection project also bounds `sqb plan --json` against an empty DuckDB warehouse. The +plan budget is a ratio to a warm `sqb compile --json` measured in the same test on the same runner, +plus a peak-RSS cap. A second case plans a 300-model copy of the benchmark and fails when the +planning CPU of the large project (plan CPU minus compile CPU) exceeds twice its linear projection +from the small one, so super-linear planning work fails on any runner. + +`test_build_performance.py` builds a 1,000-model DuckDB benchmark without tests or audits from an +empty warehouse once per module, with wall and peak-RSS limits. Its existing-state cases copy the +built project, change the queries of a few models (and rename one table in the second case so +rename discovery matches it against stored fingerprints), and bound the next `sqb plan --json`. +All cases carry `models_3000` identifiers so they run only in the 3,000-model fresh-process job. diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py index f309d36440..863289062a 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py @@ -289,3 +289,36 @@ class InspectionCommandPerformanceGuardTestCase: expected_max_output_lines: int expected_max_wall_seconds: float expected_max_rss_bytes: int + + +@dataclass(frozen=True) +class PlanComparedToCompileGuardTestCase: + description: str + expected_max_compile_ratio: float + expected_max_plan_wall_seconds: float + expected_max_rss_bytes: int + + +@dataclass(frozen=True) +class PlanScalingGuardTestCase: + description: str + small_model_count: int + expected_max_linear_factor: float + + +@dataclass(frozen=True) +class BuildPerformanceGuardTestCase: + description: str + expected_max_wall_seconds: float + expected_max_rss_bytes: int + + +@dataclass(frozen=True) +class ExistingStatePlanGuardTestCase: + description: str + edited_models: tuple[str, ...] + renamed_models: tuple[tuple[str, str], ...] + expected_query_changed: tuple[str, ...] + expected_migrations: tuple[tuple[str, str, str], ...] + expected_max_wall_seconds: float + expected_max_rss_bytes: int diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/conftest.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/conftest.py index 01872affcb..bc2754dd34 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/conftest.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/conftest.py @@ -9,7 +9,11 @@ assert_required_cgroup_memory_limit, ) from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import ( + BuiltBenchmark, + InspectionCommandMeasurement, + prepare_build_benchmark_project, prepare_inspection_benchmark_project, + run_fresh_process_inspection_command, ) from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import run_sqb @@ -35,3 +39,17 @@ def inspection_benchmark_project(tmp_path_factory: pytest.TempPathFactory) -> Pa project_dir: Path = tmp_path_factory.mktemp("inspection") / "semantic_inspection" prepare_inspection_benchmark_project(project_dir=project_dir) return project_dir + + +@pytest.fixture(scope="module") +def built_benchmark(tmp_path_factory: pytest.TempPathFactory) -> BuiltBenchmark: + assert_required_cgroup_memory_limit() + project_dir: Path = tmp_path_factory.mktemp("build") / "semantic_build" + prepare_build_benchmark_project(project_dir=project_dir) + build: InspectionCommandMeasurement = run_fresh_process_inspection_command( + project_dir=project_dir, + label="build-benchmark-build", + sqb_args=("build",), + expected_max_wall_seconds=300.0, + ) + return BuiltBenchmark(project_dir=project_dir, build=build) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py index 23e70d68fc..b40d029121 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py @@ -773,6 +773,28 @@ def prepare_inspection_benchmark_project(*, project_dir: Path) -> None: ) +def prepare_small_inspection_project(*, project_dir: Path, model_count: int) -> None: + """Write and compile the semantic benchmark scaled down to model_count models.""" + + scale: float = model_count / 3_000 + write_semantic_compile_project( + project_dir=project_dir, + model_count=model_count, + source_count=max(1, round(713 * scale)), + seed_count=max(1, round(141 * scale)), + function_count=max(1, round(71 * scale)), + macro_count=max(1, round(37 * scale)), + test_count=max(1, round(2_945 * scale)), + audit_count=max(1, round(5_056 * scale)), + ) + _ = run_fresh_process_inspection_command( + project_dir=project_dir, + label="small-inspection-compile", + sqb_args=("compile", "--json"), + expected_max_wall_seconds=60.0, + ) + + def _append_benchmark_edit(path: Path, label: str) -> None: path.write_text( path.read_text(encoding="utf-8") + f"\n-- one {label} edit\n", @@ -2156,3 +2178,85 @@ def build_empty_input_test_project_files( ")\nSELECT 1\n" ), } + + +class BuiltBenchmark(NamedTuple): + project_dir: Path + build: InspectionCommandMeasurement + + +BUILD_BENCHMARK_MODEL_COUNT: int = 1_000 +_BUILD_BENCHMARK_CONNECTION: str = '\n[connection]\ndatabase = "benchmark.duckdb"\n' +_BUILD_BENCHMARK_EDIT: tuple[str, str] = ("amount + 0 WHEN", "amount + 100 WHEN") + + +def prepare_build_benchmark_project(*, project_dir: Path) -> None: + """Write a 1,000-model benchmark without tests or audits and warm its compile cache.""" + + write_semantic_compile_project( + project_dir=project_dir, + model_count=BUILD_BENCHMARK_MODEL_COUNT, + source_count=238, + seed_count=47, + function_count=24, + macro_count=12, + test_count=0, + audit_count=0, + ) + config: Path = project_dir / "sqlbuild_project.toml" + config.write_text( + config.read_text(encoding="utf-8") + _BUILD_BENCHMARK_CONNECTION, encoding="utf-8" + ) + _ = run_fresh_process_inspection_command( + project_dir=project_dir, + label="build-benchmark-compile", + sqb_args=("compile", "--json"), + expected_max_wall_seconds=60.0, + ) + + +def change_benchmark_models( + *, + project_dir: Path, + edited_models: tuple[str, ...], + renamed_models: tuple[tuple[str, str], ...], +) -> None: + """Change the query of each edited model and rename each renamed model file.""" + + name: str + for name in edited_models: + _replace_benchmark_text( + path=_benchmark_model_path(project_dir=project_dir, name=name), + old=_BUILD_BENCHMARK_EDIT[0], + new=_BUILD_BENCHMARK_EDIT[1], + ) + old_name: str + new_name: str + for old_name, new_name in renamed_models: + path: Path = _benchmark_model_path(project_dir=project_dir, name=old_name) + _ = path.rename(path.with_name(f"{new_name}.sql")) + + +def _benchmark_model_path(*, project_dir: Path, name: str) -> Path: + return next((project_dir / "models").rglob(f"{name}.sql")) + + +def plan_reasons_and_migrations( + payload: object, +) -> tuple[tuple[str, ...], tuple[tuple[str, str, str], ...]]: + """Return query-changed model names and (origin, destination, decision) migrations.""" + + plan: dict[str, Any] = cast(dict[str, Any], payload) + return ( + tuple(sorted(model["name"] for model in filter(_query_changed, plan["models"]))), + tuple( + sorted( + (migration["origin_model"], migration["model"], migration["decision"]) + for migration in plan["migrations"] + ) + ), + ) + + +def _query_changed(model: dict[str, Any]) -> bool: + return model["reason"] == "query_changed" diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_build_performance.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_build_performance.py new file mode 100644 index 0000000000..999f5e3af7 --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_build_performance.py @@ -0,0 +1,124 @@ +"""Fresh-process speed and memory guards for sqb build and planning against existing state. + +The guarded project is a 1,000-model DuckDB benchmark without tests or audits. It is built once +from an empty warehouse; later cases copy the built project, change a few models or rename one, +and plan again so change detection and rename discovery run against real stored state. +""" + +from __future__ import annotations + +import logging +import shutil +from pathlib import Path + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import ( + BuildPerformanceGuardTestCase, + ExistingStatePlanGuardTestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import ( + BuiltBenchmark, + InspectionCommandMeasurement, + change_benchmark_models, + plan_reasons_and_migrations, + run_fresh_process_inspection_command, +) + +_LOGGER: logging.Logger = logging.getLogger(__name__) +_MIB: int = 1024 * 1024 + + +@pytest.mark.performance +@pytest.mark.cold_compile_performance +@pytest.mark.parametrize( + "test_case", + [ + BuildPerformanceGuardTestCase( + description="models_3000_job_build_1000_model_benchmark_from_empty_warehouse", + expected_max_wall_seconds=90.0, + expected_max_rss_bytes=1_600 * _MIB, + ) + ], + ids=lambda case: case.description, +) +def test_given_empty_warehouse_when_building_benchmark_then_preserves_resource_budgets( + built_benchmark: BuiltBenchmark, test_case: BuildPerformanceGuardTestCase +) -> None: + build: InspectionCommandMeasurement = built_benchmark.build + _LOGGER.info( + "build %s wall=%.3fs cpu=%.3fs peak_rss_bytes=%d", + test_case.description, + build.elapsed_seconds, + build.cpu_seconds, + build.peak_rss_bytes, + ) + + assert "Completed successfully" in build.output + assert build.elapsed_seconds < test_case.expected_max_wall_seconds + assert build.peak_rss_bytes < test_case.expected_max_rss_bytes + + +@pytest.mark.performance +@pytest.mark.cold_compile_performance +@pytest.mark.parametrize( + "test_case", + [ + ExistingStatePlanGuardTestCase( + description="models_3000_job_plan_after_editing_three_built_models", + edited_models=("model_00994", "model_00995", "model_00996"), + renamed_models=(), + expected_query_changed=("model_00994", "model_00995", "model_00996"), + expected_migrations=(), + expected_max_wall_seconds=9.0, + expected_max_rss_bytes=768 * _MIB, + ), + ExistingStatePlanGuardTestCase( + description="models_3000_job_plan_after_editing_and_renaming_built_models", + edited_models=("model_00994",), + renamed_models=(("model_00999", "model_00999_renamed"),), + expected_query_changed=("model_00994",), + expected_migrations=(("model_00999", "model_00999_renamed", "renamed"),), + expected_max_wall_seconds=9.0, + expected_max_rss_bytes=768 * _MIB, + ), + ], + ids=lambda case: case.description, +) +def test_given_built_benchmark_when_models_change_then_plan_preserves_resource_budgets( + built_benchmark: BuiltBenchmark, + test_case: ExistingStatePlanGuardTestCase, + tmp_path: Path, +) -> None: + project_dir: Path = tmp_path / "semantic_build" + _ = shutil.copytree(built_benchmark.project_dir, project_dir) + change_benchmark_models( + project_dir=project_dir, + edited_models=test_case.edited_models, + renamed_models=test_case.renamed_models, + ) + + plan: InspectionCommandMeasurement = run_fresh_process_inspection_command( + project_dir=project_dir, + label=test_case.description, + sqb_args=("plan", "--json"), + expected_max_wall_seconds=test_case.expected_max_wall_seconds, + ) + + _LOGGER.info( + "existing-state plan %s wall=%.3fs cpu=%.3fs peak_rss_bytes=%d", + test_case.description, + plan.elapsed_seconds, + plan.cpu_seconds, + plan.peak_rss_bytes, + ) + assert plan_reasons_and_migrations(plan.payload) == ( + test_case.expected_query_changed, + test_case.expected_migrations, + ) + assert plan.elapsed_seconds < test_case.expected_max_wall_seconds + assert plan.peak_rss_bytes < test_case.expected_max_rss_bytes + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py index 40a0c8719e..54f0750776 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py @@ -11,17 +11,21 @@ import logging from pathlib import Path +from typing import cast import pytest from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import ( InspectionCommandPerformanceGuardTestCase, + PlanComparedToCompileGuardTestCase, + PlanScalingGuardTestCase, ) from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import ( INSPECTION_BENCHMARK_MODEL_COUNT, SHARED_DIAMOND_HUB, SHARED_DIAMOND_ROLLUP, InspectionCommandMeasurement, + prepare_small_inspection_project, run_fresh_process_inspection_command, ) @@ -44,7 +48,7 @@ "(already shown)", ), expected_max_output_lines=_MODEL_COUNT, - expected_max_wall_seconds=2.5, + expected_max_wall_seconds=3.0, expected_max_rss_bytes=128 * _MIB, ), InspectionCommandPerformanceGuardTestCase( @@ -57,7 +61,7 @@ "(already shown)", ), expected_max_output_lines=_MODEL_COUNT, - expected_max_wall_seconds=2.5, + expected_max_wall_seconds=3.0, expected_max_rss_bytes=128 * _MIB, ), InspectionCommandPerformanceGuardTestCase( @@ -75,7 +79,7 @@ f'"id": "model:{SHARED_DIAMOND_ROLLUP}"', ), expected_max_output_lines=20 * _MODEL_COUNT, - expected_max_wall_seconds=2.5, + expected_max_wall_seconds=3.0, expected_max_rss_bytes=128 * _MIB, ), InspectionCommandPerformanceGuardTestCase( @@ -94,7 +98,7 @@ '"id": "model:model_00000"', ), expected_max_output_lines=20 * _MODEL_COUNT, - expected_max_wall_seconds=2.5, + expected_max_wall_seconds=3.0, expected_max_rss_bytes=128 * _MIB, ), InspectionCommandPerformanceGuardTestCase( @@ -154,3 +158,101 @@ def test_given_shared_dependency_benchmark_when_inspecting_then_preserves_resour assert output_lines <= test_case.expected_max_output_lines assert result.elapsed_seconds < test_case.expected_max_wall_seconds assert result.peak_rss_bytes < test_case.expected_max_rss_bytes + + +@pytest.mark.performance +@pytest.mark.cold_compile_performance +@pytest.mark.parametrize( + "test_case", + [ + PlanComparedToCompileGuardTestCase( + description="models_3000_plan_json_relative_to_warm_compile", + expected_max_compile_ratio=2.6, + expected_max_plan_wall_seconds=45.0, + expected_max_rss_bytes=1_280 * _MIB, + ) + ], + ids=lambda case: case.description, +) +def test_given_shared_dependency_benchmark_when_planning_then_stays_near_compile( + inspection_benchmark_project: Path, + test_case: PlanComparedToCompileGuardTestCase, +) -> None: + compiled: InspectionCommandMeasurement = run_fresh_process_inspection_command( + project_dir=inspection_benchmark_project, + label=f"{test_case.description}-compile", + sqb_args=("compile", "--json"), + expected_max_wall_seconds=test_case.expected_max_plan_wall_seconds, + ) + + planned: InspectionCommandMeasurement = run_fresh_process_inspection_command( + project_dir=inspection_benchmark_project, + label=test_case.description, + sqb_args=("plan", "--json"), + expected_max_wall_seconds=test_case.expected_max_plan_wall_seconds, + ) + + ratio: float = planned.elapsed_seconds / compiled.elapsed_seconds + _LOGGER.info( + "plan %s wall=%.3fs cpu=%.3fs peak_rss_bytes=%d compile_wall=%.3fs ratio=%.2f", + test_case.description, + planned.elapsed_seconds, + planned.cpu_seconds, + planned.peak_rss_bytes, + compiled.elapsed_seconds, + ratio, + ) + assert len(cast(dict[str, list[object]], planned.payload)["models"]) == _MODEL_COUNT + assert ratio < test_case.expected_max_compile_ratio + assert planned.peak_rss_bytes < test_case.expected_max_rss_bytes + + +@pytest.mark.performance +@pytest.mark.cold_compile_performance +@pytest.mark.parametrize( + "test_case", + [ + PlanScalingGuardTestCase( + description="models_3000_plan_work_scales_linearly_from_300_models", + small_model_count=300, + expected_max_linear_factor=2.0, + ) + ], + ids=lambda case: case.description, +) +def test_given_benchmarks_of_two_sizes_when_planning_then_planning_work_scales_linearly( + inspection_benchmark_project: Path, + test_case: PlanScalingGuardTestCase, + tmp_path: Path, +) -> None: + small_project: Path = tmp_path / "semantic_small" + prepare_small_inspection_project( + project_dir=small_project, model_count=test_case.small_model_count + ) + planning_cpu: dict[str, float] = {} + label: str + project_dir: Path + for label, project_dir in (("small", small_project), ("large", inspection_benchmark_project)): + compiled: InspectionCommandMeasurement = run_fresh_process_inspection_command( + project_dir=project_dir, + label=f"{test_case.description}-{label}-compile", + sqb_args=("compile", "--json"), + expected_max_wall_seconds=60.0, + ) + planned: InspectionCommandMeasurement = run_fresh_process_inspection_command( + project_dir=project_dir, + label=f"{test_case.description}-{label}-plan", + sqb_args=("plan", "--json"), + expected_max_wall_seconds=60.0, + ) + planning_cpu[label] = planned.cpu_seconds - compiled.cpu_seconds + + linear_cpu: float = planning_cpu["small"] * _MODEL_COUNT / test_case.small_model_count + _LOGGER.info( + "plan scaling %s small_planning_cpu=%.3fs large_planning_cpu=%.3fs linear=%.3fs", + test_case.description, + planning_cpu["small"], + planning_cpu["large"], + linear_cpu, + ) + assert planning_cpu["large"] < test_case.expected_max_linear_factor * linear_cpu diff --git a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py index 7012831de2..7926643061 100644 --- a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py +++ b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py @@ -787,3 +787,25 @@ class UnicodeRuleLocationTestCase: rule_code: str sql: str expected_anchors: tuple[str, ...] + + +@dataclass(frozen=True) +class SourceRebindingTestCase: + """Expression-source inspection that must rebind downstream models only on new evidence.""" + + description: str + source_expression: str + model_sql: str + expected_exit_code: int + expected_rebinding: bool + expected_fragment: str = "" + + +@dataclass(frozen=True) +class DiscoveryWorkTestCase: + """Commands that must discover a project with the same work as compile.""" + + description: str + command: tuple[str, ...] + expected_exit_code: int + expected_eager_output_column_scans: int diff --git a/tests/integration/src/sqlbuild/cli/commands/main/helpers.py b/tests/integration/src/sqlbuild/cli/commands/main/helpers.py index 4be88f914d..2f984de28d 100644 --- a/tests/integration/src/sqlbuild/cli/commands/main/helpers.py +++ b/tests/integration/src/sqlbuild/cli/commands/main/helpers.py @@ -1,9 +1,10 @@ from __future__ import annotations import json +from collections.abc import Callable from pathlib import Path from tempfile import TemporaryDirectory -from typing import cast +from typing import Any, cast import duckdb import pytest @@ -16,6 +17,10 @@ from sqlbuild.adapter.contract.types import RetentionChangePhase from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter from sqlbuild.cli.commands.main.entrypoint.entry import main +from sqlbuild.compiler.compile._helpers.assembly import source_bindings as source_bindings_module +from sqlbuild.compiler.compile.models import PolyglotAnalysisResult +from sqlbuild.compiler.discovery._helpers.filesystem import core as discovery_core_module +from sqlbuild.spec.contracts.models import SourceLocation def unavailable_artifact_directory(*, prefix: str) -> TemporaryDirectory[str]: @@ -396,3 +401,60 @@ def compile_duckdb_projection( result: dict[str, object] = json.loads(capsys.readouterr().out) diagnostics: list[dict[str, object]] = cast(list[dict[str, object]], result["diagnostics"]) return exit_code, tuple(item["code"] for item in diagnostics) + + +def write_expression_source_project( + *, project_dir: Path, source_expression: str, model_sql: str +) -> None: + """Write a DuckDB project with one typed expression source and one reading model.""" + + (project_dir / "sources").mkdir(parents=True) + (project_dir / "models").mkdir() + (project_dir / "sqlbuild_project.toml").write_text( + 'name = "orders"\nadapter = "duckdb"\n[connection]\ndatabase = "orders.duckdb"\n', + encoding="utf-8", + ) + (project_dir / "sources" / "orders.yml").write_text( + "sources:\n" + " - name: typed_orders\n" + f' expression: "{source_expression}"\n' + " columns:\n" + " - name: order_id\n" + " type: INTEGER\n", + encoding="utf-8", + ) + (project_dir / "models" / "orders.sql").write_text(model_sql, encoding="utf-8") + + +def record_source_rebinding_analyses(monkeypatch: pytest.MonkeyPatch) -> list[str]: + """Record every model query analysed again against inspected source columns.""" + + analysed: list[str] = [] + original: Callable[..., PolyglotAnalysisResult] = ( + source_bindings_module.analyze_columns_and_lineage_with_polyglot + ) + + def recording(**kwargs: Any) -> PolyglotAnalysisResult: + analysed.append(str(kwargs["query_sql"])) + return original(**kwargs) + + monkeypatch.setattr( + source_bindings_module, "analyze_columns_and_lineage_with_polyglot", recording + ) + return analysed + + +def record_eager_output_column_scans(monkeypatch: pytest.MonkeyPatch) -> list[Path]: + """Record every model file whose output columns discovery locates eagerly.""" + + scanned: list[Path] = [] + original: Callable[..., dict[str, SourceLocation]] = ( + discovery_core_module.model_output_column_locations + ) + + def recording(**kwargs: Any) -> dict[str, SourceLocation]: + scanned.append(cast(Path, kwargs["relative_path"])) + return original(**kwargs) + + monkeypatch.setattr(discovery_core_module, "model_output_column_locations", recording) + return scanned diff --git a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/_test_types.py b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/_test_types.py index 1fb66bd9b3..b79aa64a37 100644 --- a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/_test_types.py +++ b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/_test_types.py @@ -132,3 +132,14 @@ class CompletedMigrationPlanTextTestCase: description: str project_toml: str expected_target_label: str + + +@dataclass(frozen=True) +class MigrationFingerprintWorkTestCase: + description: str + prior_models: tuple[Callable[[], dict[str, str]], ...] + models: Callable[[], dict[str, str]] + command: tuple[str, ...] + expected_computations: int + expected_migrations: int + expected_stored_models: tuple[str, ...] = () diff --git a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/helpers.py b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/helpers.py index e916d75518..4469ccbccc 100644 --- a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/helpers.py +++ b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/helpers.py @@ -3,7 +3,7 @@ from __future__ import annotations import json -from collections.abc import Callable +from collections.abc import Callable, Mapping from dataclasses import dataclass from pathlib import Path from textwrap import dedent @@ -13,6 +13,7 @@ import pytest from _pytest.capture import CaptureResult +import sqlbuild.compiler.planner.classes.migration_fingerprint_cache as fingerprint_cache_module from sqlbuild.adapter.contract.models import MigrationStagePlan from sqlbuild.adapter.contract.types import MigrationTransfer from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter @@ -728,3 +729,63 @@ def state_tables(*, project_dir: Path, schema: str) -> tuple[str, ...]: ), ) ) + + +type FingerprintComputation = tuple[str, str, tuple[tuple[str, str], ...], str | None] + + +def record_fingerprint_computations( + monkeypatch: pytest.MonkeyPatch, +) -> list[FingerprintComputation]: + """Record every migration fingerprint the planner actually computes.""" + + computed: list[FingerprintComputation] = [] + original: Callable[..., str | None] = fingerprint_cache_module.build_migration_fingerprint + + def recording( + *, + query_sql: str, + metadata_json: str, + ref_identities: Mapping[str, str], + dialect: str | None, + ) -> str | None: + computed.append((query_sql, metadata_json, tuple(sorted(ref_identities.items())), dialect)) + return original( + query_sql=query_sql, + metadata_json=metadata_json, + ref_identities=ref_identities, + dialect=dialect, + ) + + monkeypatch.setattr(fingerprint_cache_module, "build_migration_fingerprint", recording) + return computed + + +def models_with_stored_migration_fingerprints(project_dir: Path) -> tuple[str, ...]: + """Return model names whose stored fingerprint rows carry a migration fingerprint.""" + + return tuple( + str(row[0]) + for row in query( + project_dir=project_dir, + sql=( + "SELECT DISTINCT node_name FROM main._sqlbuild_fingerprints " + "WHERE json_extract_string(decode(from_base64(metadata_json_b64)), " + "'$.migration_fingerprint') IS NOT NULL ORDER BY 1" + ), + ) + ) + + +def build_each( + *, + project_dir: Path, + model_sets: tuple[Callable[[], dict[str, str]], ...], + capsys: pytest.CaptureFixture[str], +) -> None: + """Build each model set in turn over the current raw orders.""" + + models: Callable[[], dict[str, str]] + for models in model_sets: + write_project(project_dir=project_dir, models=models()) + _ = build_ok(project_dir=project_dir, capsys=capsys) diff --git a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/test_fingerprint_work.py b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/test_fingerprint_work.py new file mode 100644 index 0000000000..5053547d82 --- /dev/null +++ b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/test_fingerprint_work.py @@ -0,0 +1,137 @@ +"""Integration coverage for how much migration fingerprint work plan and build perform.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from tests.integration.src.sqlbuild.cli.commands.main.model_migrations._test_types import ( + MigrationFingerprintWorkTestCase, +) +from tests.integration.src.sqlbuild.cli.commands.main.model_migrations.helpers import ( + CliRun, + FingerprintComputation, + build_each, + load_raw_orders, + models_with_stored_migration_fingerprints, + original_order_models, + plan_json, + planned_migrations, + record_fingerprint_computations, + renamed_order_models, + run_sqb, + write_project, +) + +_ORIGINAL_NAMES: tuple[str, ...] = ("daily_order_totals", "orders_enriched", "stg_orders") +_RENAMED_NAMES: tuple[str, ...] = ( + "customer_daily_order_totals", + "customer_orders_enriched", + "stg_customer_orders", +) + + +@pytest.mark.parametrize( + "test_case", + [ + MigrationFingerprintWorkTestCase( + description="plan without earlier state fingerprints nothing", + prior_models=(), + models=original_order_models, + command=("plan", "--json"), + expected_computations=0, + expected_migrations=0, + ), + MigrationFingerprintWorkTestCase( + description="plan with renamed models fingerprints each candidate once", + prior_models=(original_order_models,), + models=renamed_order_models, + command=("plan", "--json"), + expected_computations=3, + expected_migrations=3, + ), + ], + ids=lambda case: case.description, +) +def test_given_project_state_when_planning_then_fingerprints_only_migration_candidates_once( + test_case: MigrationFingerprintWorkTestCase, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + load_raw_orders(project_dir=tmp_path, first_day=1, last_day=5) + build_each(project_dir=tmp_path, model_sets=test_case.prior_models, capsys=capsys) + write_project(project_dir=tmp_path, models=test_case.models()) + computed: list[FingerprintComputation] = record_fingerprint_computations(monkeypatch) + + migrations: int = len(planned_migrations(plan_json(project_dir=tmp_path, capsys=capsys))) + + assert migrations == test_case.expected_migrations + assert len(computed) == len(set(computed)) == test_case.expected_computations + + +@pytest.mark.parametrize( + "test_case", + [ + MigrationFingerprintWorkTestCase( + description="first build fingerprints each built model once", + prior_models=(), + models=original_order_models, + command=("build",), + expected_computations=3, + expected_migrations=0, + expected_stored_models=_ORIGINAL_NAMES, + ), + MigrationFingerprintWorkTestCase( + description="rebuild with a warm compile cache computes no fingerprint", + prior_models=(original_order_models,), + models=original_order_models, + command=("build",), + expected_computations=0, + expected_migrations=0, + expected_stored_models=_ORIGINAL_NAMES, + ), + MigrationFingerprintWorkTestCase( + description="rebuild without the compile cache computes each fingerprint once", + prior_models=(original_order_models,), + models=original_order_models, + command=("build", "--no-cache"), + expected_computations=3, + expected_migrations=0, + expected_stored_models=_ORIGINAL_NAMES, + ), + MigrationFingerprintWorkTestCase( + description="renamed build after a plan computes only unplanned storage fingerprints", + prior_models=(original_order_models,), + models=renamed_order_models, + command=("build",), + expected_computations=2, + expected_migrations=3, + expected_stored_models=tuple(sorted((*_ORIGINAL_NAMES, *_RENAMED_NAMES))), + ), + ], + ids=lambda case: case.description, +) +def test_given_project_state_when_building_then_stores_fingerprints_computed_once( + test_case: MigrationFingerprintWorkTestCase, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + load_raw_orders(project_dir=tmp_path, first_day=1, last_day=5) + build_each(project_dir=tmp_path, model_sets=test_case.prior_models, capsys=capsys) + write_project(project_dir=tmp_path, models=test_case.models()) + migrations: int = len(planned_migrations(plan_json(project_dir=tmp_path, capsys=capsys))) + computed: list[FingerprintComputation] = record_fingerprint_computations(monkeypatch) + + result: CliRun = run_sqb(project_dir=tmp_path, args=test_case.command, capsys=capsys) + + assert result.exit_code == 0, result.output + assert migrations == test_case.expected_migrations + assert len(computed) == len(set(computed)) == test_case.expected_computations + assert models_with_stored_migration_fingerprints(tmp_path) == (test_case.expected_stored_models) + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_discovery_work.py b/tests/integration/src/sqlbuild/cli/commands/main/test_discovery_work.py new file mode 100644 index 0000000000..62ce0ec451 --- /dev/null +++ b/tests/integration/src/sqlbuild/cli/commands/main/test_discovery_work.py @@ -0,0 +1,71 @@ +"""Integration coverage for discovery work shared by compile, plan, and build.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from sqlbuild.cli.commands.main.entrypoint.entry import main +from tests.integration.src.sqlbuild.cli.commands.main._test_types import DiscoveryWorkTestCase +from tests.integration.src.sqlbuild.cli.commands.main.helpers import ( + execute_duckdb_sql, + record_eager_output_column_scans, + write_expression_source_project, +) + + +@pytest.mark.parametrize( + "test_case", + [ + DiscoveryWorkTestCase( + description="compile", + command=("compile",), + expected_exit_code=0, + expected_eager_output_column_scans=0, + ), + DiscoveryWorkTestCase( + description="plan", + command=("plan", "--json"), + expected_exit_code=0, + expected_eager_output_column_scans=0, + ), + DiscoveryWorkTestCase( + description="build", + command=("build",), + expected_exit_code=0, + expected_eager_output_column_scans=0, + ), + ], + ids=lambda case: case.description, +) +def test_given_project_when_running_command_then_discovery_skips_eager_column_locations( + test_case: DiscoveryWorkTestCase, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + execute_duckdb_sql( + db_path=tmp_path / "orders.duckdb", + sql="CREATE TABLE raw_orders AS SELECT 1 AS order_id", + ) + write_expression_source_project( + project_dir=tmp_path, + source_expression="(SELECT 1 AS order_id, 'open' AS status)", + model_sql=( + "MODEL (materialized table);\n" + 'SELECT o.order_id, o.status FROM __source("typed_orders") AS o\n' + ), + ) + scanned: list[Path] = record_eager_output_column_scans(monkeypatch) + _ = capsys.readouterr() + + exit_code: int = main(["--project-dir", str(tmp_path), "--no-color", *test_case.command]) + + output: str = "".join(capsys.readouterr()) + assert exit_code == test_case.expected_exit_code, output + assert len(scanned) == test_case.expected_eager_output_column_scans + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_source_rebinding.py b/tests/integration/src/sqlbuild/cli/commands/main/test_source_rebinding.py new file mode 100644 index 0000000000..52690a0815 --- /dev/null +++ b/tests/integration/src/sqlbuild/cli/commands/main/test_source_rebinding.py @@ -0,0 +1,71 @@ +"""Integration coverage for rebinding models to inspected expression-source columns.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from sqlbuild.cli.commands.main.entrypoint.entry import main +from tests.integration.src.sqlbuild.cli.commands.main._test_types import SourceRebindingTestCase +from tests.integration.src.sqlbuild.cli.commands.main.helpers import ( + execute_duckdb_sql, + record_source_rebinding_analyses, + write_expression_source_project, +) + + +@pytest.mark.parametrize( + "test_case", + [ + SourceRebindingTestCase( + description="names already known at compile time add no rebinding", + source_expression="(SELECT 1 AS order_id, 'open' AS status)", + model_sql=( + "MODEL (materialized table);\n" + 'SELECT o.order_id, o.status FROM __source("typed_orders") AS o\n' + ), + expected_exit_code=0, + expected_rebinding=False, + ), + SourceRebindingTestCase( + description="names hidden from compile rebind and reject a missing column", + source_expression="(SELECT * FROM main.raw_orders)", + model_sql=( + 'MODEL (materialized table);\nSELECT o.missing FROM __source("typed_orders") AS o\n' + ), + expected_exit_code=1, + expected_rebinding=True, + expected_fragment="error[B002]", + ), + ], + ids=lambda case: case.description, +) +def test_given_expression_source_when_planning_then_rebinds_only_on_new_column_evidence( + test_case: SourceRebindingTestCase, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + execute_duckdb_sql( + db_path=tmp_path / "orders.duckdb", + sql="CREATE TABLE raw_orders AS SELECT 1 AS order_id", + ) + write_expression_source_project( + project_dir=tmp_path, + source_expression=test_case.source_expression, + model_sql=test_case.model_sql, + ) + analysed: list[str] = record_source_rebinding_analyses(monkeypatch) + _ = capsys.readouterr() + + exit_code: int = main(["--project-dir", str(tmp_path), "--no-color", "plan"]) + + output: str = "".join(capsys.readouterr()) + assert exit_code == test_case.expected_exit_code, output + assert test_case.expected_fragment in output + assert bool(analysed) is test_case.expected_rebinding + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/unit/src/sqlbuild/compiler/compile/_helpers/_test_types.py b/tests/unit/src/sqlbuild/compiler/compile/_helpers/_test_types.py index 8f11a18d13..667e9f22b7 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/compile/_helpers/_test_types.py @@ -557,6 +557,13 @@ class RequiredExternalColumnsTestCase: expected_columns: tuple[tuple[str, str, str], ...] +@dataclass(frozen=True) +class LongUnionRequiredColumnsTestCase: + description: str + branch_count: int + expected_max_seconds: float + + @dataclass(frozen=True) class ValidateSqlSyntaxTestCase: description: str diff --git a/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_required_columns.py b/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_required_columns.py index b962f202aa..5c38d3d855 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_required_columns.py +++ b/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_required_columns.py @@ -1,5 +1,7 @@ from __future__ import annotations +import time + import pytest from sqlbuild.compiler.compile._helpers.analysis.required_columns import ( @@ -9,6 +11,7 @@ from sqlbuild.compiler.compile.types import CompiledResourceType from sqlbuild.compiler.references.types import SqlReferenceKind from tests.unit.src.sqlbuild.compiler.compile._helpers._test_types import ( + LongUnionRequiredColumnsTestCase, RequiredExternalColumnsTestCase, ) @@ -174,6 +177,26 @@ ("source", "partner_orders", "order_id"), ), ), + RequiredExternalColumnsTestCase( + description="resolves set operation branches matched by name", + query_sql=( + "WITH combined AS (" + 'SELECT order_id, payload FROM __source("web_orders") ' + "UNION ALL BY NAME " + 'SELECT payload, order_id FROM __source("partner_orders")' + ") SELECT order_id FROM combined" + ), + references=( + CompileSqlReference(SqlReferenceKind.SOURCE, "web_orders"), + CompileSqlReference(SqlReferenceKind.SOURCE, "partner_orders"), + ), + expected_columns=( + ("source", "web_orders", "order_id"), + ("source", "web_orders", "payload"), + ("source", "partner_orders", "payload"), + ("source", "partner_orders", "order_id"), + ), + ), RequiredExternalColumnsTestCase( description="does not resolve generated local columns against an outer scope", query_sql=( @@ -217,5 +240,39 @@ def test_given_query_when_resolving_required_columns_then_returns_external_reads assert frozenset(received) == frozenset(test_case.expected_columns) +@pytest.mark.parametrize( + "test_case", + [ + LongUnionRequiredColumnsTestCase( + description="two hundred branch union stays linear", + branch_count=200, + expected_max_seconds=3.0, + ) + ], + ids=lambda case: case.description, +) +def test_given_long_union_chain_when_resolving_required_columns_then_finishes_quickly( + test_case: LongUnionRequiredColumnsTestCase, +) -> None: + branch_count: int = test_case.branch_count + branches: str = " UNION ALL ".join( + f'SELECT order_id, amount_{index} AS amount FROM __source("orders")' + for index in range(branch_count) + ) + started: float = time.perf_counter() + + result: tuple[CompiledLineageSourceFact, ...] = resolve_required_external_columns( + query_sql=f"WITH combined AS ({branches}) SELECT order_id, amount FROM combined", + references=(CompileSqlReference(SqlReferenceKind.SOURCE, "orders"),), + dialect="duckdb", + ) + + elapsed: float = time.perf_counter() - started + assert frozenset(source.column_name for source in result) == frozenset( + {"order_id", *(f"amount_{index}" for index in range(branch_count))} + ) + assert elapsed < test_case.expected_max_seconds + + if __name__ == "__main__": pytest.main([__file__, "-vv"]) diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_fixture_planning_work.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_fixture_planning_work.py index 5e1557d5de..7c1878596c 100644 --- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_fixture_planning_work.py +++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_fixture_planning_work.py @@ -16,6 +16,7 @@ from sqlbuild.compiler.planner._helpers.fixtures import completion as fixture_completion from sqlbuild.compiler.planner._helpers.output.plan_output import build_selected_test_entries from sqlbuild.compiler.planner._helpers.sql_tests import assembly as sql_test_assembly +from sqlbuild.compiler.planner.classes import fixture_column_inferences as fixture_inferences from sqlbuild.compiler.planner.models import ( FixtureColumnMetadata, FixtureRelationMetadata, @@ -235,9 +236,9 @@ def test_given_complete_authoritative_fixture_when_completing_then_skips_column_ "analyze_resolved_column_reads", fallback_analyzer, ) - fixture_inference: Mock = Mock(wraps=fixture_completion.infer_fixture_column_facts) + fixture_inference: Mock = Mock(wraps=fixture_inferences.infer_fixture_column_facts) monkeypatch.setattr( - fixture_completion, + fixture_inferences, "infer_fixture_column_facts", fixture_inference, ) diff --git a/tests/unit/src/sqlbuild/compiler/planner/classes/_test_types.py b/tests/unit/src/sqlbuild/compiler/planner/classes/_test_types.py index a6f6f5f8f1..c45447080c 100644 --- a/tests/unit/src/sqlbuild/compiler/planner/classes/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/planner/classes/_test_types.py @@ -1,4 +1,8 @@ +from collections.abc import Callable from dataclasses import dataclass +from pathlib import Path + +import pytest @dataclass(frozen=True) @@ -7,3 +11,24 @@ class BackgroundSqlTestPlanningTestCase: enabled: bool expected_planner_calls: int expected_error: str | None = None + + +@dataclass(frozen=True) +class MigrationFingerprintCacheTestCase: + description: str + requests: tuple[tuple[str, dict[str, str], str], ...] + expected_computations: int + + +@dataclass(frozen=True) +class PersistedMigrationFingerprintTestCase: + description: str + between_runs: Callable[[Path, pytest.MonkeyPatch], None] + expected_second_run_computations: int + + +@dataclass(frozen=True) +class FixtureColumnInferencesTestCase: + description: str + requests: tuple[tuple[str, str], ...] + expected_inferences: int diff --git a/tests/unit/src/sqlbuild/compiler/planner/classes/helpers.py b/tests/unit/src/sqlbuild/compiler/planner/classes/helpers.py new file mode 100644 index 0000000000..016fc0b8cd --- /dev/null +++ b/tests/unit/src/sqlbuild/compiler/planner/classes/helpers.py @@ -0,0 +1,129 @@ +"""Cache-state changes applied between planner runs in migration fingerprint cache tests.""" + +from __future__ import annotations + +import json +import sqlite3 +from collections.abc import Mapping +from pathlib import Path + +import pytest + +import sqlbuild.compiler.planner.classes.fixture_column_inferences as fixture_inferences +import sqlbuild.compiler.planner.classes.migration_fingerprint_cache as fingerprint_cache +from sqlbuild.adapter.contract.models import ExpressionInferenceProfile +from sqlbuild.compiler.compile.main._infer_fixture_columns import infer_fixture_column_facts +from sqlbuild.compiler.compile.models import FixtureColumnInference +from sqlbuild.compiler.planner._helpers.migrations.fingerprint import build_migration_fingerprint + +ORDERS_SQL: str = 'SELECT o.order_id, o.amount_cents FROM __ref("stg_orders") AS o' +CURSOR_METADATA: str = json.dumps( + { + "model_name": "daily_orders", + "config": {"materialized": "incremental", "cursor_inputs": {"order_events": "event_date"}}, + } +) +PLAIN_METADATA: str = json.dumps( + {"model_name": "daily_orders", "config": {"materialized": "table"}} +) +DATABASE_GLOB: str = "migration-fingerprints-v*/migration-fingerprints.sqlite3" + + +def count_fingerprint_computations(monkeypatch: pytest.MonkeyPatch) -> list[int]: + """Count every fingerprint computed instead of reused.""" + + computations: list[int] = [] + + def counting( + *, + query_sql: str, + metadata_json: str, + ref_identities: Mapping[str, str], + dialect: str | None, + ) -> str | None: + computations.append(1) + return build_migration_fingerprint( + query_sql=query_sql, + metadata_json=metadata_json, + ref_identities=ref_identities, + dialect=dialect, + ) + + monkeypatch.setattr(fingerprint_cache, "build_migration_fingerprint", counting) + return computations + + +def keep_cache(root: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Leave the persisted cache untouched.""" + + del root, monkeypatch + + +def corrupt_database(root: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Overwrite the cache database with bytes that are not SQLite.""" + + del monkeypatch + next(root.glob(DATABASE_GLOB)).write_bytes(b"not a database") + + +def tamper_entries(root: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Change every stored fingerprint without updating its digest.""" + + del monkeypatch + connection: sqlite3.Connection = sqlite3.connect(next(root.glob(DATABASE_GLOB))) + try: + _ = connection.execute( + 'UPDATE migration_fingerprint SET payload = replace(payload, \'"f":"\', \'"f":"0\')' + ) + connection.commit() + finally: + connection.close() + + +def replace_digests_with_non_ascii(root: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Replace every stored digest with non-ASCII text, keeping the entry separator.""" + + del monkeypatch + connection: sqlite3.Connection = sqlite3.connect(next(root.glob(DATABASE_GLOB))) + try: + _ = connection.execute( + "UPDATE migration_fingerprint " + "SET payload = 'é' || substr(payload, instr(payload, char(10)))" + ) + connection.commit() + finally: + connection.close() + + +def upgrade_sqlbuild(root: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Report a different installed sqlbuild version for the next run.""" + + del root + monkeypatch.setattr(fingerprint_cache, "_package_version", lambda package: f"{package}-next") + + +FIXTURE_PROFILES: dict[str, ExpressionInferenceProfile] = { + "duckdb": ExpressionInferenceProfile(sql_analysis_dialect="duckdb"), + "duckdb_again": ExpressionInferenceProfile(sql_analysis_dialect="duckdb"), + "snowflake": ExpressionInferenceProfile(sql_analysis_dialect="snowflake"), + "catalog": ExpressionInferenceProfile(sql_analysis_dialect="duckdb", binding_catalog=object()), +} +FIXTURE_QUERIES: dict[str, str] = { + "orders": "SELECT 1 AS order_id, 'open' AS status UNION ALL SELECT 2, 'closed'", + "customers": "SELECT 7 AS customer_id, NULL AS segment", +} + + +def count_fixture_inferences(monkeypatch: pytest.MonkeyPatch) -> list[int]: + """Count every fixture inference computed instead of reused.""" + + inferences: list[int] = [] + + def counting( + *, query_sql: str, inference_profile: ExpressionInferenceProfile + ) -> FixtureColumnInference | None: + inferences.append(1) + return infer_fixture_column_facts(query_sql=query_sql, inference_profile=inference_profile) + + monkeypatch.setattr(fixture_inferences, "infer_fixture_column_facts", counting) + return inferences diff --git a/tests/unit/src/sqlbuild/compiler/planner/classes/test_fixture_column_inferences.py b/tests/unit/src/sqlbuild/compiler/planner/classes/test_fixture_column_inferences.py new file mode 100644 index 0000000000..b474b847d8 --- /dev/null +++ b/tests/unit/src/sqlbuild/compiler/planner/classes/test_fixture_column_inferences.py @@ -0,0 +1,63 @@ +"""Unit coverage for reusing fixture column inference within one planning run.""" + +from __future__ import annotations + +import pytest + +from sqlbuild.compiler.compile.main._infer_fixture_columns import infer_fixture_column_facts +from sqlbuild.compiler.compile.models import FixtureColumnInference +from sqlbuild.compiler.planner.classes.fixture_column_inferences import FixtureColumnInferences +from tests.unit.src.sqlbuild.compiler.planner.classes._test_types import ( + FixtureColumnInferencesTestCase, +) +from tests.unit.src.sqlbuild.compiler.planner.classes.helpers import ( + FIXTURE_PROFILES, + FIXTURE_QUERIES, + count_fixture_inferences, +) + + +@pytest.mark.parametrize( + "test_case", + [ + FixtureColumnInferencesTestCase( + description="identical fixture bodies under equal profiles infer once", + requests=(("orders", "duckdb"), ("orders", "duckdb_again"), ("orders", "duckdb")), + expected_inferences=1, + ), + FixtureColumnInferencesTestCase( + description="different bodies and dialects infer separately", + requests=(("orders", "duckdb"), ("customers", "duckdb"), ("orders", "snowflake")), + expected_inferences=3, + ), + FixtureColumnInferencesTestCase( + description="profiles with a binding catalog are never reused", + requests=(("orders", "catalog"), ("orders", "catalog")), + expected_inferences=2, + ), + ], + ids=lambda case: case.description, +) +def test_given_fixture_queries_when_inferring_then_matches_direct_inference_with_minimal_work( + test_case: FixtureColumnInferencesTestCase, monkeypatch: pytest.MonkeyPatch +) -> None: + inferences: list[int] = count_fixture_inferences(monkeypatch) + memo: FixtureColumnInferences = FixtureColumnInferences() + + results: list[FixtureColumnInference | None] = [ + memo.infer(query_sql=FIXTURE_QUERIES[query], inference_profile=FIXTURE_PROFILES[profile]) + for query, profile in test_case.requests + ] + + assert results == [ + infer_fixture_column_facts( + query_sql=FIXTURE_QUERIES[query], inference_profile=FIXTURE_PROFILES[profile] + ) + for query, profile in test_case.requests + ] + assert all(result is not None for result in results) + assert len(inferences) == test_case.expected_inferences + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/unit/src/sqlbuild/compiler/planner/classes/test_migration_fingerprint_cache.py b/tests/unit/src/sqlbuild/compiler/planner/classes/test_migration_fingerprint_cache.py new file mode 100644 index 0000000000..7c38f4c20a --- /dev/null +++ b/tests/unit/src/sqlbuild/compiler/planner/classes/test_migration_fingerprint_cache.py @@ -0,0 +1,198 @@ +"""Unit coverage for reusing migration fingerprints within one plan.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping +from pathlib import Path + +import pytest + +import sqlbuild.compiler.planner.classes.migration_fingerprint_cache as fingerprint_cache +from sqlbuild.compiler.planner._helpers.migrations.fingerprint import build_migration_fingerprint +from sqlbuild.compiler.planner.classes.migration_fingerprint_cache import ( + MigrationFingerprintCache, +) +from tests.unit.src.sqlbuild.compiler.planner.classes._test_types import ( + MigrationFingerprintCacheTestCase, + PersistedMigrationFingerprintTestCase, +) +from tests.unit.src.sqlbuild.compiler.planner.classes.helpers import ( + CURSOR_METADATA, + ORDERS_SQL, + PLAIN_METADATA, + corrupt_database, + count_fingerprint_computations, + keep_cache, + replace_digests_with_non_ascii, + tamper_entries, + upgrade_sqlbuild, +) + +_ORDERS_SQL: str = 'SELECT o.order_id, o.amount_cents FROM __ref("stg_orders") AS o' +_CURSOR_METADATA: str = json.dumps( + { + "model_name": "daily_orders", + "config": { + "materialized": "incremental", + "cursor_inputs": {"order_events": "event_date"}, + }, + } +) +_PLAIN_METADATA: str = json.dumps( + {"model_name": "daily_orders", "config": {"materialized": "table"}} +) +_METADATA: dict[str, str] = {"plain": _PLAIN_METADATA, "cursor": _CURSOR_METADATA} + + +@pytest.mark.parametrize( + "test_case", + [ + MigrationFingerprintCacheTestCase( + description="renames of models the query never reads reuse the fingerprint", + requests=( + ("plain", {}, "duckdb"), + ("plain", {"customers": "old_customers"}, "duckdb"), + ("plain", {}, "duckdb"), + ), + expected_computations=1, + ), + MigrationFingerprintCacheTestCase( + description="a rename of a referenced model is a distinct fingerprint", + requests=( + ("plain", {}, "duckdb"), + ("plain", {"stg_orders": "old_orders"}, "duckdb"), + ("plain", {"stg_orders": "old_orders", "customers": "x"}, "duckdb"), + ), + expected_computations=2, + ), + MigrationFingerprintCacheTestCase( + description="a rename of a cursor input is a distinct fingerprint", + requests=( + ("cursor", {}, "duckdb"), + ("cursor", {"order_events": "old_events"}, "duckdb"), + ), + expected_computations=2, + ), + MigrationFingerprintCacheTestCase( + description="dialects and metadata never share fingerprints", + requests=( + ("plain", {}, "duckdb"), + ("plain", {}, "snowflake"), + ("cursor", {}, "duckdb"), + ), + expected_computations=3, + ), + ], + ids=lambda case: case.description, +) +def test_given_fingerprint_requests_when_cached_then_matches_direct_build_with_minimal_work( + test_case: MigrationFingerprintCacheTestCase, monkeypatch: pytest.MonkeyPatch +) -> None: + computations: list[int] = [] + + def counting( + *, + query_sql: str, + metadata_json: str, + ref_identities: Mapping[str, str], + dialect: str | None, + ) -> str | None: + computations.append(1) + return build_migration_fingerprint( + query_sql=query_sql, + metadata_json=metadata_json, + ref_identities=ref_identities, + dialect=dialect, + ) + + monkeypatch.setattr(fingerprint_cache, "build_migration_fingerprint", counting) + cache: MigrationFingerprintCache = MigrationFingerprintCache() + + results: list[str | None] = [ + cache.fingerprint( + query_sql=_ORDERS_SQL, + metadata_json=_METADATA[metadata], + ref_identities=identities, + dialect=dialect, + ) + for metadata, identities, dialect in test_case.requests + ] + + assert results == [ + build_migration_fingerprint( + query_sql=_ORDERS_SQL, + metadata_json=_METADATA[metadata], + ref_identities=identities, + dialect=dialect, + ) + for metadata, identities, dialect in test_case.requests + ] + assert all(result is not None for result in results) + assert len(computations) == test_case.expected_computations + + +@pytest.mark.parametrize( + "test_case", + [ + PersistedMigrationFingerprintTestCase( + description="a warm cache computes nothing", + between_runs=keep_cache, + expected_second_run_computations=0, + ), + PersistedMigrationFingerprintTestCase( + description="a corrupt database is recomputed", + between_runs=corrupt_database, + expected_second_run_computations=2, + ), + PersistedMigrationFingerprintTestCase( + description="tampered entries are recomputed", + between_runs=tamper_entries, + expected_second_run_computations=2, + ), + PersistedMigrationFingerprintTestCase( + description="non-ASCII digests are recomputed", + between_runs=replace_digests_with_non_ascii, + expected_second_run_computations=2, + ), + PersistedMigrationFingerprintTestCase( + description="a new sqlbuild version is recomputed", + between_runs=upgrade_sqlbuild, + expected_second_run_computations=2, + ), + ], + ids=lambda case: case.description, +) +def test_given_persisted_fingerprints_when_planning_again_then_reuses_only_valid_entries( + test_case: PersistedMigrationFingerprintTestCase, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + requests: tuple[str, ...] = (PLAIN_METADATA, CURSOR_METADATA) + first: MigrationFingerprintCache = MigrationFingerprintCache(root=tmp_path) + expected: list[str | None] = [ + first.fingerprint( + query_sql=ORDERS_SQL, metadata_json=metadata, ref_identities={}, dialect="duckdb" + ) + for metadata in requests + ] + first.persist() + test_case.between_runs(tmp_path, monkeypatch) + computations: list[int] = count_fingerprint_computations(monkeypatch) + second: MigrationFingerprintCache = MigrationFingerprintCache(root=tmp_path) + + results: list[str | None] = [ + second.fingerprint( + query_sql=ORDERS_SQL, metadata_json=metadata, ref_identities={}, dialect="duckdb" + ) + for metadata in requests + ] + second.persist() + + assert results == expected + assert all(result is not None for result in results) + assert len(computations) == test_case.expected_second_run_computations + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"])