From c71fa191c55358c4e5d6f068edef2acc184f5368 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Sun, 20 Sep 2026 07:45:32 +0100 Subject: [PATCH] fix(planner): avoid duplicate planning work --- .../planner/_helpers/changes/detect.py | 76 +++++++---- .../planner/_helpers/fixtures/completion.py | 16 ++- .../planner/_helpers/output/plan_output.py | 27 +++- .../planner/_helpers/planning/identities.py | 90 +++++++++--- .../_helpers/planning/output_assembly.py | 1 + .../_helpers/pruning/selection_staleness.py | 21 ++- .../planner/_helpers/sql_tests/assembly.py | 129 +++++++++++++++--- .../planner/main/execution/execution.py | 48 +++++-- src/sqlbuild/compiler/planner/models.py | 11 ++ .../compiler/planner/main/_test_types.py | 11 ++ .../planner/main/test_build_execution_plan.py | 106 ++++++++++++++ .../compiler/planner/_helpers/_test_types.py | 7 + .../planner/_helpers/planning/_test_types.py | 9 ++ .../_helpers/planning/test_identities.py | 88 ++++++++++++ .../_helpers/test_fixture_planning_work.py | 5 + .../_helpers/test_selection_staleness.py | 67 +++++++++ 16 files changed, 625 insertions(+), 87 deletions(-) create mode 100644 tests/unit/src/sqlbuild/compiler/planner/_helpers/planning/test_identities.py diff --git a/src/sqlbuild/compiler/planner/_helpers/changes/detect.py b/src/sqlbuild/compiler/planner/_helpers/changes/detect.py index 952ffd0c1b..e21637a2ed 100644 --- a/src/sqlbuild/compiler/planner/_helpers/changes/detect.py +++ b/src/sqlbuild/compiler/planner/_helpers/changes/detect.py @@ -60,32 +60,13 @@ def detect_changes( ) -> PlannerChangeResults: """Detect selected model and function changes.""" - model_changes: dict[str, ChangeDetectionResult] = {} - function_local_hashes: dict[str, str] = build_function_local_hashes( - functions=project.functions, + model_changes: dict[str, ChangeDetectionResult] = detect_model_changes_in_scope( + project=project, + scope=scope, + snapshot=snapshot, + expected_version_hashes=expected_version_hashes, + expected_metadata_jsons=expected_metadata_jsons, ) - hook_version_hashes: dict[str, str] = { - name: identity.version_hash - for name, identity in build_hook_identities(project.hook_functions).items() - } - key: CompiledObjectKey - for key in scope.execution_order: - if key not in scope.selected_keys or key.resource_type != CompiledResourceType.MODEL: - continue - model: CompiledModel | None = scope.models_by_name.get(key.name) - if model is None: - continue - model_changes[model.name] = detect_model_changes( - model=model, - snapshot=snapshot, - sql_analysis_enabled=project.settings.sql_analysis, - query_change_tracking=project.settings.query_change_tracking, - full_refresh=False, - function_local_hashes=function_local_hashes, - hook_version_hashes=hook_version_hashes, - expected_version_hash=(expected_version_hashes or {}).get(model.name), - expected_metadata_json=(expected_metadata_jsons or {}).get(model.name), - ) function_changes: dict[str, FunctionChangeResult] = {} function: CompiledFunction @@ -116,6 +97,51 @@ def detect_changes( return PlannerChangeResults(models=model_changes, functions=function_changes) +def detect_model_changes_in_scope( + *, + project: CompiledProject, + scope: PlannerScope, + snapshot: WarehouseSnapshot, + query_change_tracking: bool | None = None, + expected_version_hashes: dict[str, str] | None = None, + expected_metadata_jsons: dict[str, str] | None = None, +) -> dict[str, ChangeDetectionResult]: + """Detect changes for selected models without inspecting functions.""" + + model_changes: dict[str, ChangeDetectionResult] = {} + effective_query_change_tracking: bool = ( + project.settings.query_change_tracking + if query_change_tracking is None + else query_change_tracking + ) + function_local_hashes: dict[str, str] = build_function_local_hashes( + functions=project.functions, + ) + hook_version_hashes: dict[str, str] = { + name: identity.version_hash + for name, identity in build_hook_identities(project.hook_functions).items() + } + key: CompiledObjectKey + for key in scope.execution_order: + if key not in scope.selected_keys or key.resource_type != CompiledResourceType.MODEL: + continue + model: CompiledModel | None = scope.models_by_name.get(key.name) + if model is None: + continue + model_changes[model.name] = detect_model_changes( + model=model, + snapshot=snapshot, + sql_analysis_enabled=project.settings.sql_analysis, + query_change_tracking=effective_query_change_tracking, + full_refresh=False, + function_local_hashes=function_local_hashes, + hook_version_hashes=hook_version_hashes, + expected_version_hash=(expected_version_hashes or {}).get(model.name), + expected_metadata_json=(expected_metadata_jsons or {}).get(model.name), + ) + return model_changes + + def detect_model_changes( *, model: CompiledModel, diff --git a/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py b/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py index cebbf65996..bb398968df 100644 --- a/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py +++ b/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py @@ -198,19 +198,27 @@ def _fixture_keys_requiring_analysis( return frozenset(required) -def build_relation_fixture_context(*, project: CompiledProject) -> RelationFixturePlanningContext: +def build_relation_fixture_context( + *, + project: CompiledProject, + models_by_name: dict[str, CompiledModel] | None = None, +) -> RelationFixturePlanningContext: """Build immutable project-wide fixture metadata once per planning invocation.""" - models_by_name: dict[str, CompiledModel] = {model.name: model for model in project.models} + model_map: dict[str, CompiledModel] = ( + {model.name: model for model in project.models} + if models_by_name is None + else models_by_name + ) relations: dict[FixtureKey, FixtureRelationMetadata] = _fixture_relation_metadata( project=project, - model_map=models_by_name, + model_map=model_map, ) authoritative_columns: dict[FixtureKey, frozenset[str]] = _authoritative_fixture_columns( relations=relations ) return RelationFixturePlanningContext( - models_by_name=models_by_name, + models_by_name=model_map, relations=relations, authoritative_columns=authoritative_columns, expected_types=_base_expected_fixture_types(relations=relations), diff --git a/src/sqlbuild/compiler/planner/_helpers/output/plan_output.py b/src/sqlbuild/compiler/planner/_helpers/output/plan_output.py index 0c5fb35cd3..6d9b7422bf 100644 --- a/src/sqlbuild/compiler/planner/_helpers/output/plan_output.py +++ b/src/sqlbuild/compiler/planner/_helpers/output/plan_output.py @@ -33,7 +33,10 @@ scope_overlaps, ) from sqlbuild.compiler.planner._helpers.resolve.resolve import resolve_function_sql -from sqlbuild.compiler.planner._helpers.sql_tests.assembly import plan_test +from sqlbuild.compiler.planner._helpers.sql_tests.assembly import ( + build_sql_test_planning_context, + plan_test, +) from sqlbuild.compiler.planner.exceptions import ( PlannerInputError, SqlTestFixtureValidationError, @@ -56,6 +59,7 @@ SeedPlanEntry, SourceLoadPlanEntry, SqlTestPlanEntry, + SqlTestPlanningContext, WarehouseSnapshot, ) from sqlbuild.compiler.planner.types import PlanReason @@ -445,15 +449,29 @@ def build_selected_test_entries( entries: list[SqlTestPlanEntry] = [] warnings: list[PlanWarning] = [] fixture_diagnostics: list[str] = [] - fixture_planning_context: RelationFixturePlanningContext | None = None + selected_tests: list[CompiledSqlTest] = [] sql_test: CompiledSqlTest for sql_test in project.sql_tests: if not scope_overlaps(scope_deps=sql_test.scope_deps, selected_keys=selected_keys): continue if case_name is not None and sql_test.case_name != case_name: continue - if project.settings.sql_analysis and fixture_planning_context is None: - fixture_planning_context = build_relation_fixture_context(project=project) + selected_tests.append(sql_test) + if not selected_tests: + return entries, warnings + test_planning_context: SqlTestPlanningContext = build_sql_test_planning_context( + project=project, + tests=tuple(selected_tests), + ) + fixture_planning_context: RelationFixturePlanningContext | None = ( + build_relation_fixture_context( + project=project, + models_by_name=test_planning_context.models_by_name, + ) + if project.settings.sql_analysis + else None + ) + for sql_test in selected_tests: test_entry: SqlTestPlanEntry test_warnings: tuple[PlanWarning, ...] try: @@ -464,6 +482,7 @@ def build_selected_test_entries( sql_analysis_enabled=project.settings.sql_analysis, validate_fixtures=True, fixture_planning_context=fixture_planning_context, + planning_context=test_planning_context, ) except SqlTestFixtureValidationError as error: fixture_diagnostics.extend(error.diagnostics) diff --git a/src/sqlbuild/compiler/planner/_helpers/planning/identities.py b/src/sqlbuild/compiler/planner/_helpers/planning/identities.py index 928d1c296b..bac63e0d53 100644 --- a/src/sqlbuild/compiler/planner/_helpers/planning/identities.py +++ b/src/sqlbuild/compiler/planner/_helpers/planning/identities.py @@ -4,12 +4,15 @@ from dataclasses import replace -from sqlbuild.compiler.compile.models import CompiledProject -from sqlbuild.compiler.planner._helpers.changes.detect import detect_changes +from sqlbuild.compiler.compile.models import CompiledObjectKey, CompiledProject +from sqlbuild.compiler.compile.types import CompiledResourceType +from sqlbuild.compiler.planner._helpers.changes.detect import detect_model_changes_in_scope from sqlbuild.compiler.planner._helpers.identity.direct import ( build_direct_model_version_identities, ) from sqlbuild.compiler.planner.models import ( + ChangeDetectionResult, + DirectModelVersionIdentities, PlannerChangeResults, PlannerIdentityContext, PlannerScopeResolution, @@ -21,21 +24,27 @@ def build_planner_identity_context( *, project: CompiledProject, scopes: PlannerScopeResolution, + include_stale_warning_identities: bool = True, ) -> PlannerIdentityContext: """Build expected version identities for the inspection and stale-warning scopes.""" + version_identities: DirectModelVersionIdentities = build_direct_model_version_identities( + functions=project.functions, + seeds=project.seeds, + scope=scopes.inspection_scope, + hook_functions=project.hook_functions, + ) return PlannerIdentityContext( - version_identities=build_direct_model_version_identities( - functions=project.functions, - seeds=project.seeds, - scope=scopes.inspection_scope, - hook_functions=project.hook_functions, - ), - stale_warning_identities=build_direct_model_version_identities( - functions=project.functions, - seeds=project.seeds, - scope=scopes.stale_warning_scope, - hook_functions=project.hook_functions, + version_identities=version_identities, + stale_warning_identities=( + build_direct_model_version_identities( + functions=project.functions, + seeds=project.seeds, + scope=scopes.stale_warning_scope, + hook_functions=project.hook_functions, + ) + if include_stale_warning_identities + else version_identities ), ) @@ -46,17 +55,64 @@ def detect_stale_warning_changes( scopes: PlannerScopeResolution, snapshot: WarehouseSnapshot, identities: PlannerIdentityContext, + execution_changes: PlannerChangeResults, ) -> PlannerChangeResults: - """Detect project-wide changes used for stale-out-of-selection warnings.""" + """Detect only stale-warning model changes not already known from execution planning.""" - return detect_changes( + reusable_model_names: frozenset[str] = _reusable_execution_model_names( + identities=identities, + execution_changes=execution_changes, + query_change_tracking=project.settings.query_change_tracking, + ) + stale_keys: frozenset[CompiledObjectKey] = frozenset( + scopes.stale_warning_scope.all_keys.values() + ) + remaining_model_keys: frozenset[CompiledObjectKey] = frozenset( + key + for key in stale_keys + if key.resource_type == CompiledResourceType.MODEL and key.name not in reusable_model_names + ) + remaining_model_changes: dict[str, ChangeDetectionResult] = detect_model_changes_in_scope( project=project, scope=replace( scopes.stale_warning_scope, - selected_keys=frozenset(scopes.stale_warning_scope.all_keys.values()), + selected_keys=remaining_model_keys, ), snapshot=snapshot, - full_refresh=False, + query_change_tracking=True, expected_version_hashes=identities.stale_warning_identities.model_version_hashes, expected_metadata_jsons=identities.stale_warning_identities.model_metadata_jsons, ) + return PlannerChangeResults( + models={ + **{ + name: change + for name, change in execution_changes.models.items() + if name in reusable_model_names + }, + **remaining_model_changes, + }, + functions={}, + ) + + +def _reusable_execution_model_names( + *, + identities: PlannerIdentityContext, + execution_changes: PlannerChangeResults, + query_change_tracking: bool, +) -> frozenset[str]: + if not query_change_tracking: + return frozenset() + execution_identities: DirectModelVersionIdentities = identities.version_identities + stale_identities: DirectModelVersionIdentities = identities.stale_warning_identities + return frozenset( + model_name + for model_name in execution_changes.models + if ( + execution_identities.model_version_hashes.get(model_name) + == stale_identities.model_version_hashes.get(model_name) + and execution_identities.model_metadata_jsons.get(model_name) + == stale_identities.model_metadata_jsons.get(model_name) + ) + ) diff --git a/src/sqlbuild/compiler/planner/_helpers/planning/output_assembly.py b/src/sqlbuild/compiler/planner/_helpers/planning/output_assembly.py index b988d09643..b18331c4bc 100644 --- a/src/sqlbuild/compiler/planner/_helpers/planning/output_assembly.py +++ b/src/sqlbuild/compiler/planner/_helpers/planning/output_assembly.py @@ -93,6 +93,7 @@ def with_plan_warnings( version_identities=identities.stale_warning_identities, source_freshness=source_freshness, include_sources=False, + model_changes_complete=True, ) ) if stale_out_of_selection_warnings: diff --git a/src/sqlbuild/compiler/planner/_helpers/pruning/selection_staleness.py b/src/sqlbuild/compiler/planner/_helpers/pruning/selection_staleness.py index 41487377f8..c0816766ee 100644 --- a/src/sqlbuild/compiler/planner/_helpers/pruning/selection_staleness.py +++ b/src/sqlbuild/compiler/planner/_helpers/pruning/selection_staleness.py @@ -34,6 +34,7 @@ def build_stale_out_of_selection_warnings( version_identities: DirectModelVersionIdentities, source_freshness: DirectSourceFreshnessPlanningResult | None, include_sources: bool = True, + model_changes_complete: bool = False, ) -> tuple[PlanWarning, ...]: """Warn for selected models stale through changed upstreams outside the run set.""" @@ -67,6 +68,7 @@ def build_stale_out_of_selection_warnings( changes=changes, snapshot=snapshot, version_identities=version_identities, + model_changes_complete=model_changes_complete, ), changed_seed_names=_changed_seed_names( snapshot=snapshot, @@ -129,6 +131,7 @@ def _changed_model_names( changes: PlannerChangeResults, snapshot: WarehouseSnapshot, version_identities: DirectModelVersionIdentities, + model_changes_complete: bool, ) -> frozenset[str]: changed: set[str] = { model_name @@ -150,6 +153,7 @@ def _changed_model_names( changes=changes, snapshot=snapshot, version_identities=version_identities, + model_changes_complete=model_changes_complete, ) ) return frozenset(changed) @@ -186,16 +190,25 @@ def _model_own_identity_changed( changes: PlannerChangeResults, snapshot: WarehouseSnapshot, version_identities: DirectModelVersionIdentities, + model_changes_complete: bool, ) -> bool: selected_change: ChangeDetectionResult | None = changes.models.get(model_name) - if selected_change is not None: - if selected_change.change_kind in { + if selected_change is not None and ( + model_changes_complete + or selected_change.change_kind + in { ChangeKind.FIRST_RUN, ChangeKind.QUERY_CHANGED, ChangeKind.CONFIG_CHANGED, ChangeKind.SCHEMA_CHANGED, - }: - return True + } + ): + return selected_change.change_kind in { + ChangeKind.FIRST_RUN, + ChangeKind.QUERY_CHANGED, + ChangeKind.CONFIG_CHANGED, + ChangeKind.SCHEMA_CHANGED, + } model: CompiledModel | None = original_scope.models_by_name.get(model_name) if model is None: return False diff --git a/src/sqlbuild/compiler/planner/_helpers/sql_tests/assembly.py b/src/sqlbuild/compiler/planner/_helpers/sql_tests/assembly.py index 3e0ab89447..540038ec0d 100644 --- a/src/sqlbuild/compiler/planner/_helpers/sql_tests/assembly.py +++ b/src/sqlbuild/compiler/planner/_helpers/sql_tests/assembly.py @@ -50,6 +50,7 @@ SqlAnalysisResolvedTestSql, SqlTestAssertionStep, SqlTestPlanEntry, + SqlTestPlanningContext, ) from sqlbuild.compiler.planner.types import WarningSeverity from sqlbuild.compiler.references.main._quoted_reference_call_pattern import ( @@ -127,6 +128,60 @@ def resolve_all(self) -> dict[str, str]: return self.resolved +def build_sql_test_planning_context( + *, project: CompiledProject, tests: tuple[CompiledSqlTest, ...] = () +) -> SqlTestPlanningContext: + """Build project-wide indexes shared by every SQL test in one plan.""" + + models_by_name: dict[str, CompiledModel] = {model.name: model for model in project.models} + model_dependencies: dict[str, frozenset[str]] = _build_model_dependencies( + model_map=models_by_name + ) + function_locations: dict[str, CompiledRelationLocation] = { + function.name: function.destination for function in project.functions + } + chain_names_by_topology: dict[ + tuple[tuple[str, ...], tuple[tuple[str, str], ...], frozenset[str]], + tuple[str, ...], + ] = {} + chain_names_by_test_key: dict[CompiledObjectKey, tuple[str, ...]] = {} + test: CompiledSqlTest + for test in tests: + topology_inputs: tuple[tuple[str, ...], dict[str, str], frozenset[str]] | None = ( + _test_model_chain_topology_inputs(test=test) + ) + if topology_inputs is None: + continue + expected_names, model_query_overrides, mock_ref_names = topology_inputs + topology_key: tuple[tuple[str, ...], tuple[tuple[str, str], ...], frozenset[str]] = ( + tuple(sorted(set(expected_names))), + tuple(sorted(model_query_overrides.items())), + mock_ref_names, + ) + ordered_names: tuple[str, ...] | None = chain_names_by_topology.get(topology_key) + if ordered_names is None: + ordered_names = _topo_sort_model_chain( + expected_names=expected_names, + model_map=models_by_name, + model_dependencies=model_dependencies, + model_query_overrides=model_query_overrides, + mock_ref_names=mock_ref_names, + ) + chain_names_by_topology[topology_key] = ordered_names + chain_names_by_test_key[test.key] = ordered_names + return SqlTestPlanningContext( + models_by_name=models_by_name, + model_dependencies=model_dependencies, + function_locations=function_locations, + qualified_function_locations={ + name: target.qualified_name + for name, target in function_locations.items() + if target.qualified_name is not None + }, + chain_names_by_test_key=chain_names_by_test_key, + ) + + def plan_test( *, test: CompiledSqlTest, @@ -135,17 +190,19 @@ def plan_test( sql_analysis_enabled: bool = False, validate_fixtures: bool = False, fixture_planning_context: RelationFixturePlanningContext | None = None, + planning_context: SqlTestPlanningContext | None = None, ) -> tuple[SqlTestPlanEntry, tuple[PlanWarning, ...]]: """Build a test plan entry with chained resolution.""" + context: SqlTestPlanningContext = planning_context or build_sql_test_planning_context( + project=project, + tests=(test,), + ) if isinstance(test.payload, CompiledDirectLogicSqlTestPayload): - function_locations: dict[str, CompiledRelationLocation] = { - function.name: function.destination for function in project.functions - } return ( _plan_direct_logic_test( test=test, - function_locations=function_locations, + function_locations=context.function_locations, function_deps=_direct_function_deps(test=test, project=project), adapter=adapter, sql_analysis_enabled=sql_analysis_enabled, @@ -155,15 +212,9 @@ def plan_test( model_payload: CompiledModelSqlTestPayload = test.payload - model_map: dict[str, CompiledModel] = {m.name: m for m in project.models} - function_locations: dict[str, CompiledRelationLocation] = { - function.name: function.destination for function in project.functions - } - qualified_function_locations: dict[str, str] = { - name: target.qualified_name - for name, target in function_locations.items() - if target.qualified_name is not None - } + model_map: dict[str, CompiledModel] = context.models_by_name + function_locations: dict[str, CompiledRelationLocation] = context.function_locations + qualified_function_locations: dict[str, str] = context.qualified_function_locations mock_refs: dict[str, str] = _extract_mock_refs(test) mock_sources: dict[str, str] = _extract_mock_sources(test) mock_seeds: dict[str, str] = _extract_mock_seeds(test) @@ -179,12 +230,15 @@ def plan_test( expected_names: tuple[str, ...] = tuple( dict.fromkeys((*model_payload.expected_model_names, *assertion_target_names)) ) - ordered_names: tuple[str, ...] = _topo_sort_model_chain( - expected_names=expected_names, - model_map=model_map, - model_query_overrides=model_payload.model_query_overrides, - mock_ref_names=frozenset(mock_refs), - ) + ordered_names: tuple[str, ...] | None = context.chain_names_by_test_key.get(test.key) + if ordered_names is None: + ordered_names = _topo_sort_model_chain( + expected_names=expected_names, + model_map=model_map, + model_dependencies=context.model_dependencies, + model_query_overrides=model_payload.model_query_overrides, + mock_ref_names=frozenset(mock_refs), + ) if sql_analysis_enabled and validate_fixtures: mock_refs, mock_sources, mock_seeds = build_validated_test_fixtures( test=test, @@ -401,11 +455,30 @@ def resolve_test_model_chain_names( return _topo_sort_model_chain( expected_names=expected_names, model_map=effective_model_map, + model_dependencies=_build_model_dependencies(model_map=effective_model_map), model_query_overrides=test.payload.model_query_overrides, mock_ref_names=frozenset(_extract_mock_refs(test)), ) +def _test_model_chain_topology_inputs( + *, test: CompiledSqlTest +) -> tuple[tuple[str, ...], dict[str, str], frozenset[str]] | None: + if not isinstance(test.payload, CompiledModelSqlTestPayload): + return None + assertion_target_names: tuple[str, ...] = _extract_assertion_ref_targets( + assertion_map=_extract_assertion_ctes(test) + ) + expected_names: tuple[str, ...] = tuple( + dict.fromkeys((*test.payload.expected_model_names, *assertion_target_names)) + ) + return ( + expected_names, + test.payload.model_query_overrides, + frozenset(_extract_mock_refs(test)), + ) + + def _build_assertion_steps( *, assertion_map: dict[str, str], @@ -921,6 +994,7 @@ def _topo_sort_model_chain( *, expected_names: tuple[str, ...], model_map: dict[str, CompiledModel], + model_dependencies: dict[str, frozenset[str]], model_query_overrides: dict[str, str], mock_ref_names: frozenset[str], ) -> tuple[str, ...]: @@ -939,6 +1013,7 @@ def _visit(*, node: str, seen: set[str], result: list[str]) -> tuple[set[str], l model=model, query_override=model_query_overrides.get(node), model_map=model_map, + model_dependencies=model_dependencies, mock_ref_names=mock_ref_names, ) dependency_name: str @@ -957,6 +1032,7 @@ def _test_model_dependency_names( model: CompiledModel, query_override: str | None, model_map: dict[str, CompiledModel], + model_dependencies: dict[str, frozenset[str]], mock_ref_names: frozenset[str], ) -> set[str]: if query_override is not None: @@ -965,12 +1041,21 @@ def _test_model_dependency_names( for match in uncommented_pattern_matches(pattern=_REF_PATTERN, sql=query_override) } else: - candidates = { + candidates = set(model_dependencies.get(model.name, ())) + return {name for name in candidates if name not in mock_ref_names and name in model_map} + + +def _build_model_dependencies(*, model_map: dict[str, CompiledModel]) -> dict[str, frozenset[str]]: + dependencies_by_model: dict[str, frozenset[str]] = {} + model: CompiledModel + for model in model_map.values(): + dependencies_by_model[model.name] = frozenset( dependency.name for dependency in model.deps if dependency.resource_type == CompiledResourceType.MODEL - } - return {name for name in candidates if name not in mock_ref_names and name in model_map} + and dependency.name in model_map + ) + return dependencies_by_model def _extract_mock_refs(test: CompiledSqlTest) -> dict[str, str]: diff --git a/src/sqlbuild/compiler/planner/main/execution/execution.py b/src/sqlbuild/compiler/planner/main/execution/execution.py index 9863b5c6be..d5e9304ba6 100644 --- a/src/sqlbuild/compiler/planner/main/execution/execution.py +++ b/src/sqlbuild/compiler/planner/main/execution/execution.py @@ -49,6 +49,7 @@ PlannerSelection, PlannerWarehouseState, PlanOutput, + WarehouseSnapshot, ) from sqlbuild.compiler.source_freshness.models import DirectSourceFreshnessPlanningResult from sqlbuild.spec.contracts.models import LocalConfig, ProjectConfig @@ -90,12 +91,7 @@ def build_execution_plan( identities: PlannerIdentityContext = build_planner_identity_context( project=project, scopes=scopes, - ) - stale_warning_changes: PlannerChangeResults = detect_stale_warning_changes( - project=project, - scopes=scopes, - snapshot=warehouse.snapshot, - identities=identities, + include_stale_warning_identities=policies.selection_diagnostics, ) check_selected_scope_buildability( project=project, @@ -103,13 +99,15 @@ def build_execution_plan( snapshot=warehouse.snapshot, deferral=deferral, ) - changes: PlannerChangeResults = detect_changes( + changes: PlannerChangeResults + stale_warning_changes: PlannerChangeResults + changes, stale_warning_changes = _detect_planner_change_results( project=project, - scope=scopes.inspection_scope, + scopes=scopes, snapshot=warehouse.snapshot, - full_refresh=overrides.full_refresh, - expected_version_hashes=identities.version_identities.model_version_hashes, - expected_metadata_jsons=identities.version_identities.model_metadata_jsons, + identities=identities, + overrides=overrides, + policies=policies, ) resolved_actions: PlannerResolvedActions = resolve_cascades( scope=scopes.inspection_scope, @@ -183,3 +181,31 @@ def build_execution_plan( if on_progress is not None: on_progress(f"Generated plan. ({time.monotonic() - plan_start:.2f}s)") return plan_output + + +def _detect_planner_change_results( + *, + project: CompiledProject, + scopes: PlannerScopeResolution, + snapshot: WarehouseSnapshot, + identities: PlannerIdentityContext, + overrides: PlannerOverrides, + policies: PlannerPolicies, +) -> tuple[PlannerChangeResults, PlannerChangeResults]: + changes: PlannerChangeResults = detect_changes( + project=project, + scope=scopes.inspection_scope, + snapshot=snapshot, + full_refresh=overrides.full_refresh, + expected_version_hashes=identities.version_identities.model_version_hashes, + expected_metadata_jsons=identities.version_identities.model_metadata_jsons, + ) + if not policies.selection_diagnostics: + return changes, PlannerChangeResults(models={}, functions={}) + return changes, detect_stale_warning_changes( + project=project, + scopes=scopes, + snapshot=snapshot, + identities=identities, + execution_changes=changes, + ) diff --git a/src/sqlbuild/compiler/planner/models.py b/src/sqlbuild/compiler/planner/models.py index 64ccb4d751..e57289ccc6 100644 --- a/src/sqlbuild/compiler/planner/models.py +++ b/src/sqlbuild/compiler/planner/models.py @@ -132,6 +132,17 @@ class RelationFixturePlanningContext: expected_types: dict[FixtureKey, dict[str, str]] +@dataclass(frozen=True) +class SqlTestPlanningContext: + """Project-wide SQL-test indexes and reusable chain topology.""" + + models_by_name: dict[str, CompiledModel] + model_dependencies: dict[str, frozenset[str]] + function_locations: dict[str, CompiledRelationLocation] + qualified_function_locations: dict[str, str] + chain_names_by_test_key: dict[CompiledObjectKey, tuple[str, ...]] + + @dataclass(frozen=True) class GraphNodeKey: """Neutral graph key matching fingerprint identity fields.""" diff --git a/tests/integration/src/sqlbuild/compiler/planner/main/_test_types.py b/tests/integration/src/sqlbuild/compiler/planner/main/_test_types.py index b706f9bb46..fa0f96534c 100644 --- a/tests/integration/src/sqlbuild/compiler/planner/main/_test_types.py +++ b/tests/integration/src/sqlbuild/compiler/planner/main/_test_types.py @@ -56,6 +56,17 @@ class BuildExecutionPlanTestCase: expected_progress_fragments: tuple[str, ...] = field(default_factory=tuple) +@dataclass(frozen=True) +class PlannerChangeDetectionWorkTestCase: + description: str + selection_diagnostics: bool + expected_model_calls: dict[str, int] + expected_warning_count: int + query_change_tracking: bool = True + expected_query_tracking_calls: dict[tuple[str, bool], int] = field(default_factory=dict) + expected_identity_builds: int = 1 + + @dataclass(frozen=True) class SourceCursorInputPlanErrorTestCase: description: str diff --git a/tests/integration/src/sqlbuild/compiler/planner/main/test_build_execution_plan.py b/tests/integration/src/sqlbuild/compiler/planner/main/test_build_execution_plan.py index 585c10fa1c..3803447a2e 100644 --- a/tests/integration/src/sqlbuild/compiler/planner/main/test_build_execution_plan.py +++ b/tests/integration/src/sqlbuild/compiler/planner/main/test_build_execution_plan.py @@ -1,7 +1,9 @@ from __future__ import annotations +from collections import Counter from dataclasses import replace from typing import Any +from unittest.mock import Mock import pytest @@ -10,6 +12,8 @@ from sqlbuild.cli.output.main.plan import format_plan from sqlbuild.compiler.compile.models import CompiledProject from sqlbuild.compiler.compile.types import FunctionLanguage +from sqlbuild.compiler.planner._helpers.changes import detect as change_detection +from sqlbuild.compiler.planner._helpers.planning import identities as identity_planning from sqlbuild.compiler.planner.exceptions import PlannerInputError from sqlbuild.compiler.planner.models import ( CascadeResult, @@ -38,6 +42,7 @@ FormatPlanIntegrationTestCase, FutureCursorPlannerErrorTestCase, FutureCursorPlannerTestCase, + PlannerChangeDetectionWorkTestCase, SourceCursorInputPlanErrorTestCase, TableTypePlanAssemblyTestCase, ) @@ -51,6 +56,107 @@ ) +@pytest.mark.parametrize( + "test_case", + [ + PlannerChangeDetectionWorkTestCase( + description="selection diagnostics disabled", + selection_diagnostics=False, + expected_model_calls={"orders": 1}, + expected_warning_count=0, + expected_query_tracking_calls={("orders", True): 1}, + ), + PlannerChangeDetectionWorkTestCase( + description="selection diagnostics enabled", + selection_diagnostics=True, + expected_model_calls={"stg_orders": 1, "orders": 1}, + expected_warning_count=0, + expected_query_tracking_calls={ + ("stg_orders", True): 1, + ("orders", True): 1, + }, + expected_identity_builds=2, + ), + PlannerChangeDetectionWorkTestCase( + description="selection diagnostics preserve independent query tracking", + selection_diagnostics=True, + expected_model_calls={"stg_orders": 1, "orders": 2}, + expected_warning_count=0, + query_change_tracking=False, + expected_query_tracking_calls={ + ("orders", False): 1, + ("stg_orders", True): 1, + ("orders", True): 1, + }, + expected_identity_builds=2, + ), + ], + ids=lambda case: case.description, +) +def test_given_selected_model_when_planning_then_each_required_model_is_change_detected_once( + test_case: PlannerChangeDetectionWorkTestCase, + adapter: DuckDbAdapter, + connection: Any, + monkeypatch: pytest.MonkeyPatch, +) -> None: + connection.execute("CREATE TABLE staging.stg_orders AS SELECT 1 AS order_id") + project: CompiledProject = build_project_from_test_case( + BuildExecutionPlanTestCase( + description=test_case.description, + setup_sql=(), + model_locations={"stg_orders": "staging", "orders": "staging"}, + model_configs={ + "stg_orders": {"materialized": "table"}, + "orders": {"materialized": "table"}, + }, + model_queries={ + "stg_orders": "SELECT 1 AS order_id", + "orders": "SELECT * FROM staging.stg_orders", + }, + model_deps={"orders": ("stg_orders",)}, + full_refresh=False, + select=("orders",), + expected_action={}, + expected_reason={}, + ) + ) + project = replace( + project, + settings=replace( + project.settings, + query_change_tracking=test_case.query_change_tracking, + ), + ) + model_detector: Mock = Mock(wraps=change_detection.detect_model_changes) + monkeypatch.setattr(change_detection, "detect_model_changes", model_detector) + identity_builder: Mock = Mock(wraps=identity_planning.build_direct_model_version_identities) + monkeypatch.setattr( + identity_planning, + "build_direct_model_version_identities", + identity_builder, + ) + + plan: PlanOutput = build_execution_plan_from_kwargs( + project=project, + adapter=adapter, + connection=connection, + select=("orders",), + selection_diagnostics=test_case.selection_diagnostics, + ) + + model_calls: Counter[str] = Counter( + call.kwargs["model"].name for call in model_detector.call_args_list + ) + assert dict(model_calls) == test_case.expected_model_calls + query_tracking_calls: Counter[tuple[str, bool]] = Counter( + (call.kwargs["model"].name, call.kwargs["query_change_tracking"]) + for call in model_detector.call_args_list + ) + assert dict(query_tracking_calls) == test_case.expected_query_tracking_calls + assert identity_builder.call_count == test_case.expected_identity_builds + assert len(plan.warnings) == test_case.expected_warning_count + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/_test_types.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/_test_types.py index aece5714e8..505aa80a30 100644 --- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/_test_types.py @@ -52,6 +52,7 @@ class FixturePlanningWorkTestCase: test_count: int expected_metadata_builds: int expected_analysis_calls: int + expected_topology_builds: int @dataclass(frozen=True) @@ -61,6 +62,12 @@ class CompleteFixturePlanningTestCase: expected_fixture_sql: str +@dataclass(frozen=True) +class CompleteModelChangesWorkTestCase: + description: str + expected_detector_calls: int + + @dataclass(frozen=True) class BuildUpstreamDepsTestCase: description: str diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/planning/_test_types.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/planning/_test_types.py index f059e11507..60760dad2a 100644 --- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/planning/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/planning/_test_types.py @@ -36,3 +36,12 @@ class TableTypePlanningTestCase: materialized: str = "table" additional_config: tuple[tuple[str, object], ...] = () relation_type: str = "BASE TABLE" + + +@dataclass(frozen=True) +class ReusableExecutionModelNamesTestCase: + description: str + execution_hash: str + stale_warning_hash: str + query_change_tracking: bool + expected_names: frozenset[str] diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/planning/test_identities.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/planning/test_identities.py new file mode 100644 index 0000000000..b4d4755c1f --- /dev/null +++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/planning/test_identities.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import pytest + +from sqlbuild.compiler.planner._helpers.planning.identities import ( + _reusable_execution_model_names, +) +from sqlbuild.compiler.planner.models import ( + ChangeDetectionResult, + DirectModelVersionIdentities, + PlannerChangeResults, + PlannerIdentityContext, +) +from sqlbuild.compiler.planner.types import ChangeKind +from tests.unit.src.sqlbuild.compiler.planner._helpers.planning._test_types import ( + ReusableExecutionModelNamesTestCase, +) + + +@pytest.mark.parametrize( + "test_case", + [ + ReusableExecutionModelNamesTestCase( + description="different scope identity", + execution_hash="execution-hash", + stale_warning_hash="stale-warning-hash", + query_change_tracking=True, + expected_names=frozenset(), + ), + ReusableExecutionModelNamesTestCase( + description="query tracking disabled", + execution_hash="version-hash", + stale_warning_hash="version-hash", + query_change_tracking=False, + expected_names=frozenset(), + ), + ReusableExecutionModelNamesTestCase( + description="matching identity with query tracking", + execution_hash="version-hash", + stale_warning_hash="version-hash", + query_change_tracking=True, + expected_names=frozenset({"orders"}), + ), + ], + ids=lambda case: case.description, +) +def test_given_execution_change_when_checking_reuse_then_requires_equivalent_semantics( + test_case: ReusableExecutionModelNamesTestCase, +) -> None: + execution_identities: DirectModelVersionIdentities = DirectModelVersionIdentities( + function_local_hashes={}, + seed_version_hashes={}, + seed_metadata_jsons={}, + model_metadata_jsons={"orders": '{"materialized":"table"}'}, + model_local_hashes={"orders": "local-hash"}, + model_version_hashes={"orders": test_case.execution_hash}, + ) + stale_identities: DirectModelVersionIdentities = DirectModelVersionIdentities( + function_local_hashes={}, + seed_version_hashes={}, + seed_metadata_jsons={}, + model_metadata_jsons={"orders": '{"materialized":"table"}'}, + model_local_hashes={"orders": "local-hash"}, + model_version_hashes={"orders": test_case.stale_warning_hash}, + ) + + reusable_names: frozenset[str] = _reusable_execution_model_names( + identities=PlannerIdentityContext( + version_identities=execution_identities, + stale_warning_identities=stale_identities, + ), + execution_changes=PlannerChangeResults( + models={ + "orders": ChangeDetectionResult( + model_name="orders", + change_kind=ChangeKind.NO_CHANGE, + ) + }, + functions={}, + ), + query_change_tracking=test_case.query_change_tracking, + ) + + assert reusable_names == test_case.expected_names + + +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 d79f5249ff..374d6684ef 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 @@ -15,6 +15,7 @@ from sqlbuild.compiler.compile.types import CompiledResourceType 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.models import ( FixtureColumnMetadata, FixtureRelationMetadata, @@ -44,6 +45,7 @@ test_count=250, expected_metadata_builds=1, expected_analysis_calls=1, + expected_topology_builds=1, ), ), ids=lambda case: case.description, @@ -98,6 +100,8 @@ def test_given_many_tests_when_building_entries_then_project_fixture_metadata_is "analyze_resolved_column_reads", fallback_analyzer, ) + topology_builder: Mock = Mock(wraps=sql_test_assembly._topo_sort_model_chain) + monkeypatch.setattr(sql_test_assembly, "_topo_sort_model_chain", topology_builder) entries: list[SqlTestPlanEntry] warnings: list[PlanWarning] @@ -111,6 +115,7 @@ def test_given_many_tests_when_building_entries_then_project_fixture_metadata_is assert warnings == [] assert metadata_builder.call_count == test_case.expected_metadata_builds assert fallback_analyzer.call_count == test_case.expected_analysis_calls + assert topology_builder.call_count == test_case.expected_topology_builds @pytest.mark.parametrize( diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_selection_staleness.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_selection_staleness.py index c781ebbd21..fcf7f4338a 100644 --- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_selection_staleness.py +++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_selection_staleness.py @@ -2,6 +2,7 @@ from dataclasses import replace from datetime import UTC, datetime +from unittest.mock import Mock import pytest @@ -10,6 +11,7 @@ from sqlbuild.compiler.fingerprints.constants import NODE_TYPE_SEED from sqlbuild.compiler.fingerprints.models import Fingerprint from sqlbuild.compiler.planner._helpers.planning.scopes import resolve_planner_scopes +from sqlbuild.compiler.planner._helpers.pruning import selection_staleness from sqlbuild.compiler.planner._helpers.pruning.selection_classifier import ( format_stale_upstream_warning_message, ) @@ -34,6 +36,7 @@ SourceFreshnessIdentity, ) from tests.unit.src.sqlbuild.compiler.planner._helpers._test_types import ( + CompleteModelChangesWorkTestCase, PlannerStaleWarningScopeTestCase, SelectionStalenessGraphWarningTestCase, SelectionStalenessWarningTestCase, @@ -47,6 +50,70 @@ resource_type=CompiledResourceType.MODEL, name="c", ) + + +@pytest.mark.parametrize( + "test_case", + [ + CompleteModelChangesWorkTestCase( + description="complete model changes", + expected_detector_calls=0, + ) + ], + ids=lambda case: case.description, +) +def test_given_complete_change_results_when_building_stale_warnings_then_models_are_not_redetected( + test_case: CompleteModelChangesWorkTestCase, + monkeypatch: pytest.MonkeyPatch, +) -> None: + model: CompiledModel = build_run_despite_unchanged_model( + key=MODEL_KEY, + name=MODEL_KEY.name, + materialized="table", + run_despite_unchanged=None, + ) + scope: PlannerScope = PlannerScope( + upstream_deps={MODEL_KEY: ()}, + downstream_deps={}, + all_keys={MODEL_KEY.name: MODEL_KEY}, + models_by_name={MODEL_KEY.name: model}, + selected_keys=frozenset({MODEL_KEY}), + execution_order=(MODEL_KEY,), + ) + model_detector: Mock = Mock( + side_effect=AssertionError("complete change results must not trigger model detection") + ) + monkeypatch.setattr(selection_staleness, "detect_model_changes", model_detector) + + warnings: tuple[PlanWarning, ...] = build_stale_out_of_selection_warnings( + original_scope=scope, + execution_scope=scope, + changes=PlannerChangeResults( + models={ + MODEL_KEY.name: ChangeDetectionResult( + model_name=MODEL_KEY.name, + change_kind=ChangeKind.NO_CHANGE, + ) + }, + functions={}, + ), + snapshot=WarehouseSnapshot(), + version_identities=DirectModelVersionIdentities( + function_local_hashes={}, + seed_version_hashes={}, + seed_metadata_jsons={}, + model_metadata_jsons={}, + model_local_hashes={}, + model_version_hashes={}, + ), + source_freshness=None, + model_changes_complete=True, + ) + + assert warnings == () + assert model_detector.call_count == test_case.expected_detector_calls + + SEED_KEY: CompiledObjectKey = CompiledObjectKey( resource_type=CompiledResourceType.SEED, name="orders_seed",