From 49fc433d05777b3c30cff9c917838b3219ef62eb Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Tue, 6 Oct 2026 04:57:19 +0100 Subject: [PATCH 1/3] perf(compile): reuse discovered declaration files after model-only edits After an edit that touches only existing model files, reuse the non-model declaration files the stored compile parsed (sources, tests, schemas, macros, functions, seeds, scoped declarations) and discover only the model files again. Any other change discovers everything and records the declarations with the stored renders. --- .../cli/commands/_helpers/compile/pipeline.py | 2 + .../compile/classes/render_reuse_registry.py | 9 + .../compile/classes/render_reuse_session.py | 58 +++++- src/sqlbuild/compiler/compile/constants.py | 1 + .../_helpers/filesystem/aggregation.py | 176 ++++++++++-------- .../compiler/discovery/main/discover.py | 7 + src/sqlbuild/compiler/discovery/models.py | 21 +++ src/sqlbuild/compiler/discovery/types.py | 19 ++ .../compiler/compile/classes/_test_types.py | 10 + .../compiler/compile/classes/helpers.py | 78 +++++++- .../classes/test_render_reuse_session.py | 68 ++++++- 11 files changed, 367 insertions(+), 82 deletions(-) diff --git a/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py b/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py index f722548eee..b17b387940 100644 --- a/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py +++ b/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py @@ -34,6 +34,7 @@ from sqlbuild.cli.output.models import ( WrittenTarget, ) +from sqlbuild.compiler.compile.constants import COMPILE_RENDER_REUSE from sqlbuild.compiler.compile.exceptions import CompileInputError from sqlbuild.compiler.compile.models import ( CompileAnalysisSelection, @@ -123,6 +124,7 @@ def _analyze_compile_project( cache_request=DiscoveryCacheRequest( selected_target=selected_target, no_cache=analysis_selection.no_cache ), + declaration_reuse=COMPILE_RENDER_REUSE.claim_discovery(), ) discover_ms: int = elapsed_ms(discover_start) _ = complete_compile_phase( diff --git a/src/sqlbuild/compiler/compile/classes/render_reuse_registry.py b/src/sqlbuild/compiler/compile/classes/render_reuse_registry.py index c6a35d2e99..ad40b52907 100644 --- a/src/sqlbuild/compiler/compile/classes/render_reuse_registry.py +++ b/src/sqlbuild/compiler/compile/classes/render_reuse_registry.py @@ -39,3 +39,12 @@ def claim(self) -> CompileRenderReuseSession | None: if session is None or not session.claim(): return None return session + + def claim_discovery(self) -> CompileRenderReuseSession | None: + """Hand the active session to the first project discovery only, so it is used once.""" + + with self._lock: + session: CompileRenderReuseSession | None = self._session + if session is None or not session.claim_discovery(): + return None + return session diff --git a/src/sqlbuild/compiler/compile/classes/render_reuse_session.py b/src/sqlbuild/compiler/compile/classes/render_reuse_session.py index f011a76c43..868c5349b7 100644 --- a/src/sqlbuild/compiler/compile/classes/render_reuse_session.py +++ b/src/sqlbuild/compiler/compile/classes/render_reuse_session.py @@ -15,6 +15,7 @@ from sqlbuild.compiler.compile.classes.compile_input_reads import CompileInputReads from sqlbuild.compiler.compile.constants import ( COMPILE_INPUT_READS, + RENDER_REUSE_DECLARATIONS_PREFIX, RENDER_REUSE_MODEL_AUDITS_PREFIX, ) from sqlbuild.compiler.compile.models import ( @@ -24,7 +25,7 @@ RenderReuseState, StoredRender, ) -from sqlbuild.compiler.discovery.models import DiscoveredSqlModelFile +from sqlbuild.compiler.discovery.models import DiscoveredDeclarationFiles, DiscoveredSqlModelFile from sqlbuild.compiler.fact_cache.exceptions import FactCachePayloadError from sqlbuild.compiler.fact_cache.main._dump_fact_payload import dumped_fact_payload from sqlbuild.compiler.fact_cache.main._load_fact_payload import loaded_fact_payload @@ -59,6 +60,7 @@ def __init__( ) -> None: self._lock: threading.Lock = threading.Lock() self._claimed: bool = False + self._discovery_claimed: bool = False self._prior: RenderReuseState | None = prior if changed_paths is not None else None self._changed_paths: frozenset[str] = changed_paths or frozenset() self._reusable: bool = False @@ -83,6 +85,54 @@ def claim(self) -> bool: self._claimed = True return not claimed + def claim_discovery(self) -> bool: + """Return True for the first discovery claim only; later discoveries run in full.""" + + with self._lock: + claimed: bool = self._discovery_claimed + self._discovery_claimed = True + return not claimed + + def declaration_files( + self, + *, + variant: str, + discover: Callable[[], DiscoveredDeclarationFiles], + discover_models: Callable[[], tuple[DiscoveredSqlModelFile, ...]], + ) -> DiscoveredDeclarationFiles: + """Reuse stored declaration files after model-only edits, or discover and record them.""" + + name: str = f"{RENDER_REUSE_DECLARATIONS_PREFIX}{variant}" + prior: RenderReuseState | None = self._prior + payload: memoryview | None = ( + prior.group_payloads.get(name) + if prior is not None + and edits_only_models(changed_paths=self._changed_paths, model_paths=prior.model_paths) + else None + ) + stored: StoredRender | None = None if payload is None else _loaded_render(payload) + if ( + stored is not None + and payload is not None + and isinstance(stored.value, DiscoveredDeclarationFiles) + ): + model_files: tuple[DiscoveredSqlModelFile, ...] = discover_models() + _replay(stored=stored) + self._group_records[name] = payload + self._decoded_groups[name] = stored + return replace(stored.value, model_files=model_files) + with ( + COMPILE_INPUT_READS.recording() as reads, + tapped_compile_diagnostics() as reported, + ): + declarations: DiscoveredDeclarationFiles = discover() + recorded: StoredRender | None = _stored_render( + reads=reads, value=replace(declarations, model_files=()), reported=reported + ) + if recorded is not None: + self._pending_groups.append((name, recorded)) + return declarations + def plan_models(self, *, model_files: tuple[DiscoveredSqlModelFile, ...]) -> None: """Reuse stored renders only when the change set touches nothing but existing models.""" @@ -237,6 +287,12 @@ def _serialize_pending(self) -> None: pending.clear() +def edits_only_models(*, changed_paths: frozenset[str], model_paths: tuple[str, ...]) -> bool: + """Return whether every changed path is a model file of the stored compile.""" + + return changed_paths.issubset(model_paths) + + def _reserialized( *, records: dict[str, memoryview | None], decoded: dict[str, StoredRender] ) -> dict[str, memoryview | None]: diff --git a/src/sqlbuild/compiler/compile/constants.py b/src/sqlbuild/compiler/compile/constants.py index e137192e6f..4d030eda5a 100644 --- a/src/sqlbuild/compiler/compile/constants.py +++ b/src/sqlbuild/compiler/compile/constants.py @@ -180,6 +180,7 @@ RENDER_REUSE_SINGULAR_AUDITS_GROUP: str = "singular_audits" RENDER_REUSE_ATTACHED_AUDITS_GROUP: str = "source_and_seed_audits" RENDER_REUSE_MODEL_AUDITS_PREFIX: str = "model_audits:" +RENDER_REUSE_DECLARATIONS_PREFIX: str = "declarations:" SQL_TEST_FACT_CACHE_NAMESPACE: str = "sql-tests" SQL_TEST_FACT_CACHE_ALGORITHM: str = "expanded-sql-test-ctes-v2" SQL_TEST_EXPECTED_MODELS_FACT_ALGORITHM: str = "sql-test-expected-models-v2" diff --git a/src/sqlbuild/compiler/discovery/_helpers/filesystem/aggregation.py b/src/sqlbuild/compiler/discovery/_helpers/filesystem/aggregation.py index d2e37bec65..c5ad6b944b 100644 --- a/src/sqlbuild/compiler/discovery/_helpers/filesystem/aggregation.py +++ b/src/sqlbuild/compiler/discovery/_helpers/filesystem/aggregation.py @@ -3,6 +3,7 @@ from __future__ import annotations from collections.abc import Callable +from functools import partial from pathlib import Path from sqlbuild.compiler.discovery._helpers.filesystem.command_output_sinks import ( @@ -36,29 +37,21 @@ from sqlbuild.compiler.discovery._helpers.yml.project import load_local_config, load_project_config from sqlbuild.compiler.discovery.classes.directory_snapshot import DirectorySnapshot from sqlbuild.compiler.discovery.models import ( - DiscoveredAdapterFile, DiscoveredAssetFunction, DiscoveredAuditFactory, - DiscoveredAuditFile, DiscoveredCheckFunction, DiscoveredCommandOutputSink, DiscoveredConstantFile, + DiscoveredDeclarationFiles, DiscoveredEnumFile, DiscoveredEventExporter, DiscoveredHookFunction, DiscoveredLoaderFunction, DiscoveredMacroFile, DiscoveredMaterializationFile, - DiscoveredModelSchemaFile, DiscoveredProjectInputs, DiscoveredProvider, - DiscoveredPythonFunctionFile, DiscoveredPythonNodeFunctions, - DiscoveredSchemaFile, - DiscoveredSeedFile, - DiscoveredSourceFile, - DiscoveredSqlFunctionFile, - DiscoveredSqlHookFile, DiscoveredSqlModelFile, DiscoveredSqlScenarioFile, DiscoveredSqlTestFile, @@ -66,6 +59,7 @@ DiscoveryFileFault, TolerantScopeDiscovery, ) +from sqlbuild.compiler.discovery.types import DeclarationFilesReuse from sqlbuild.compiler.fact_cache.classes.fact_cache_store import FactCacheStore from sqlbuild.runtime.observability.classes.operation_lifecycle import OperationLifecycle from sqlbuild.spec.contracts.models import LocalConfig, ProjectConfig @@ -79,74 +73,56 @@ def build_discovered_project_inputs( sql_analysis_enabled: bool, extract_output_column_locations: bool = True, fact_cache: FactCacheStore | None = None, + declaration_reuse: DeclarationFilesReuse | None = None, ) -> DiscoveredProjectInputs: """Discover all project files and functions into one inputs bundle.""" - with ( - OperationLifecycle( - operation_kind="project", operation_name="discovery_declaration_parse" - ) as declaration_lifecycle, - DirectorySnapshot.scope(project_dir=project_dir), - ): - source_files: tuple[DiscoveredSourceFile, ...] = discover_source_files( - project_dir=project_dir, fact_cache=fact_cache - ) - model_files: tuple[DiscoveredSqlModelFile, ...] = discover_model_files( + with OperationLifecycle( + operation_kind="project", operation_name="discovery_declaration_parse" + ) as declaration_lifecycle: + discover: Callable[[], DiscoveredDeclarationFiles] = partial( + _discover_declaration_files, project_dir=project_dir, - extract_implicit_alias_columns=sql_analysis_enabled, + sql_analysis_enabled=sql_analysis_enabled, extract_output_column_locations=extract_output_column_locations, + fact_cache=fact_cache, ) - enum_files: tuple[DiscoveredEnumFile, ...] = discover_enum_files(project_dir=project_dir) - constant_files: tuple[DiscoveredConstantFile, ...] = discover_constant_files( - project_dir=project_dir - ) - model_schema_files: tuple[DiscoveredModelSchemaFile, ...] = discover_model_schema_files( - project_dir=project_dir - ) - sql_function_files: tuple[DiscoveredSqlFunctionFile, ...] = discover_sql_function_files( - project_dir=project_dir - ) - sql_hook_files: tuple[DiscoveredSqlHookFile, ...] = discover_sql_hook_files( - project_dir=project_dir - ) - python_function_files: tuple[DiscoveredPythonFunctionFile, ...] = ( - discover_python_function_files(project_dir=project_dir) - ) - schema_files: tuple[DiscoveredSchemaFile, ...] = discover_schema_files( - project_dir=project_dir - ) - seed_files: tuple[DiscoveredSeedFile, ...] = discover_seed_files(project_dir=project_dir) - test_files: tuple[DiscoveredSqlTestFile, ...] = discover_test_files( - project_dir=project_dir, fact_cache=fact_cache - ) - scenario_files: tuple[DiscoveredSqlScenarioFile, ...] = discover_scenario_files( - project_dir=project_dir + declarations: DiscoveredDeclarationFiles = ( + discover() + if declaration_reuse is None + else declaration_reuse.declaration_files( + variant=f"{int(sql_analysis_enabled)}{int(extract_output_column_locations)}", + discover=discover, + discover_models=partial( + _discover_model_files, + project_dir=project_dir, + sql_analysis_enabled=sql_analysis_enabled, + extract_output_column_locations=extract_output_column_locations, + ), + ) ) - audit_files: tuple[DiscoveredAuditFile, ...] = discover_audit_files(project_dir=project_dir) - macro_files: tuple[DiscoveredMacroFile, ...] = discover_macro_files(project_dir=project_dir) - adapter_file: DiscoveredAdapterFile | None = discover_adapter_file(project_dir=project_dir) declaration_lifecycle.completed( metadata={ "item_count": sum( len(files) for files in ( - source_files, - model_files, - enum_files, - constant_files, - model_schema_files, - sql_function_files, - sql_hook_files, - python_function_files, - schema_files, - seed_files, - test_files, - scenario_files, - audit_files, - macro_files, + declarations.source_files, + declarations.model_files, + declarations.enum_files, + declarations.constant_files, + declarations.model_schema_files, + declarations.sql_function_files, + declarations.sql_hook_files, + declarations.python_function_files, + declarations.schema_files, + declarations.seed_files, + declarations.test_files, + declarations.scenario_files, + declarations.audit_files, + declarations.macro_files, ) ) - + int(adapter_file is not None) + + int(declarations.adapter_file is not None) } ) with OperationLifecycle( @@ -191,7 +167,7 @@ def build_discovered_project_inputs( python_lifecycle.completed(metadata={"item_count": len(python_paths)}) loader_functions: tuple[DiscoveredLoaderFunction, ...] = tuple( python_nodes.loaders - ) + build_integration_loader_functions(source_files) + ) + build_integration_loader_functions(declarations.source_files) task_functions: tuple[DiscoveredTaskFunction, ...] = tuple(python_nodes.tasks) asset_functions: tuple[DiscoveredAssetFunction, ...] = tuple(python_nodes.assets) check_functions: tuple[DiscoveredCheckFunction, ...] = tuple(python_nodes.checks) @@ -200,20 +176,20 @@ def build_discovered_project_inputs( project_config=project_config, local_config=local_config, project_dir=project_dir, - model_files=model_files, - enum_files=enum_files, - constant_files=constant_files, - model_schema_files=model_schema_files, - sql_function_files=sql_function_files, - sql_hook_files=sql_hook_files, - python_function_files=python_function_files, - schema_files=schema_files, - source_files=source_files, - seed_files=seed_files, - test_files=test_files, - scenario_files=scenario_files, - audit_files=audit_files, - macro_files=macro_files, + model_files=declarations.model_files, + enum_files=declarations.enum_files, + constant_files=declarations.constant_files, + model_schema_files=declarations.model_schema_files, + sql_function_files=declarations.sql_function_files, + sql_hook_files=declarations.sql_hook_files, + python_function_files=declarations.python_function_files, + schema_files=declarations.schema_files, + source_files=declarations.source_files, + seed_files=declarations.seed_files, + test_files=declarations.test_files, + scenario_files=declarations.scenario_files, + audit_files=declarations.audit_files, + macro_files=declarations.macro_files, materialization_files=materialization_files, loader_functions=loader_functions, task_functions=task_functions, @@ -224,10 +200,52 @@ def build_discovered_project_inputs( event_exporters=event_exporters, command_output_sinks=command_output_sinks, providers=providers, - adapter_file=adapter_file, + adapter_file=declarations.adapter_file, ) +def _discover_declaration_files( + *, + project_dir: Path, + sql_analysis_enabled: bool, + extract_output_column_locations: bool, + fact_cache: FactCacheStore | None, +) -> DiscoveredDeclarationFiles: + with DirectorySnapshot.scope(project_dir=project_dir): + return DiscoveredDeclarationFiles( + source_files=discover_source_files(project_dir=project_dir, fact_cache=fact_cache), + model_files=discover_model_files( + project_dir=project_dir, + extract_implicit_alias_columns=sql_analysis_enabled, + extract_output_column_locations=extract_output_column_locations, + ), + enum_files=discover_enum_files(project_dir=project_dir), + constant_files=discover_constant_files(project_dir=project_dir), + model_schema_files=discover_model_schema_files(project_dir=project_dir), + sql_function_files=discover_sql_function_files(project_dir=project_dir), + sql_hook_files=discover_sql_hook_files(project_dir=project_dir), + python_function_files=discover_python_function_files(project_dir=project_dir), + schema_files=discover_schema_files(project_dir=project_dir), + seed_files=discover_seed_files(project_dir=project_dir), + test_files=discover_test_files(project_dir=project_dir, fact_cache=fact_cache), + scenario_files=discover_scenario_files(project_dir=project_dir), + audit_files=discover_audit_files(project_dir=project_dir), + macro_files=discover_macro_files(project_dir=project_dir), + adapter_file=discover_adapter_file(project_dir=project_dir), + ) + + +def _discover_model_files( + *, project_dir: Path, sql_analysis_enabled: bool, extract_output_column_locations: bool +) -> tuple[DiscoveredSqlModelFile, ...]: + with DirectorySnapshot.scope(project_dir=project_dir): + return discover_model_files( + project_dir=project_dir, + extract_implicit_alias_columns=sql_analysis_enabled, + extract_output_column_locations=extract_output_column_locations, + ) + + def build_tolerant_scope_discovery(*, project_dir: Path) -> TolerantScopeDiscovery: """Aggregate bounded scope inputs while retaining independent authored faults.""" diff --git a/src/sqlbuild/compiler/discovery/main/discover.py b/src/sqlbuild/compiler/discovery/main/discover.py index bd5915c9eb..ccf9207c03 100644 --- a/src/sqlbuild/compiler/discovery/main/discover.py +++ b/src/sqlbuild/compiler/discovery/main/discover.py @@ -24,6 +24,7 @@ SQL_ANALYSIS_SETTING_KEY, ) from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs, DiscoveryCacheRequest +from sqlbuild.compiler.discovery.types import DeclarationFilesReuse from sqlbuild.compiler.fact_cache.classes.fact_cache_store import FactCacheStore from sqlbuild.runtime.observability.classes.operation_lifecycle import OperationLifecycle from sqlbuild.spec.contracts.models import LocalConfig, ProjectConfig @@ -35,6 +36,7 @@ def discover_project_inputs( sql_analysis_enabled_override: bool | None = None, extract_output_column_locations: bool = True, cache_request: DiscoveryCacheRequest | None = None, + declaration_reuse: DeclarationFilesReuse | None = None, ) -> DiscoveredProjectInputs: """Load all raw project inputs from disk before semantic resolution.""" @@ -44,6 +46,7 @@ def discover_project_inputs( sql_analysis_enabled_override=sql_analysis_enabled_override, extract_output_column_locations=extract_output_column_locations, cache_request=cache_request, + declaration_reuse=declaration_reuse, ) @@ -53,6 +56,7 @@ def _discover_project_inputs( sql_analysis_enabled_override: bool | None, extract_output_column_locations: bool, cache_request: DiscoveryCacheRequest | None, + declaration_reuse: DeclarationFilesReuse | None, ) -> DiscoveredProjectInputs: with OperationLifecycle(operation_kind="project", operation_name="discovery_project_assembly"): return _assemble_discovered_project_inputs( @@ -60,6 +64,7 @@ def _discover_project_inputs( sql_analysis_enabled_override=sql_analysis_enabled_override, extract_output_column_locations=extract_output_column_locations, cache_request=cache_request, + declaration_reuse=declaration_reuse, ) @@ -69,6 +74,7 @@ def _assemble_discovered_project_inputs( sql_analysis_enabled_override: bool | None, extract_output_column_locations: bool, cache_request: DiscoveryCacheRequest | None, + declaration_reuse: DeclarationFilesReuse | None, ) -> DiscoveredProjectInputs: project_config: ProjectConfig = load_project_config(project_dir=project_dir) local_config: LocalConfig = load_local_config(project_dir=project_dir) @@ -105,6 +111,7 @@ def _assemble_discovered_project_inputs( sql_analysis_enabled=sql_analysis_enabled, extract_output_column_locations=extract_output_column_locations, fact_cache=fact_cache, + declaration_reuse=declaration_reuse, ) validate_discovered_inputs(discovered_inputs) from sqlbuild.runtime.event_exporting.main.configure_discovered_event_exporters import ( diff --git a/src/sqlbuild/compiler/discovery/models.py b/src/sqlbuild/compiler/discovery/models.py index 1bf81b98bf..d2d289568f 100644 --- a/src/sqlbuild/compiler/discovery/models.py +++ b/src/sqlbuild/compiler/discovery/models.py @@ -626,6 +626,27 @@ class DiscoveryCacheRequest: no_cache: bool = False +@dataclass(frozen=True) +class DiscoveredDeclarationFiles: + """Project files discovery parses without importing any project Python.""" + + source_files: tuple[DiscoveredSourceFile, ...] + model_files: tuple[DiscoveredSqlModelFile, ...] + enum_files: tuple[DiscoveredEnumFile, ...] + constant_files: tuple[DiscoveredConstantFile, ...] + model_schema_files: tuple[DiscoveredModelSchemaFile, ...] + sql_function_files: tuple[DiscoveredSqlFunctionFile, ...] + sql_hook_files: tuple[DiscoveredSqlHookFile, ...] + python_function_files: tuple[DiscoveredPythonFunctionFile, ...] + schema_files: tuple[DiscoveredSchemaFile, ...] + seed_files: tuple[DiscoveredSeedFile, ...] + test_files: tuple[DiscoveredSqlTestFile, ...] + scenario_files: tuple[DiscoveredSqlScenarioFile, ...] + audit_files: tuple[DiscoveredAuditFile, ...] + macro_files: tuple[DiscoveredMacroFile, ...] + adapter_file: DiscoveredAdapterFile | None + + @dataclass(frozen=True) class DiscoveredProjectInputs: """All raw project inputs discovered from disk before semantic resolution.""" diff --git a/src/sqlbuild/compiler/discovery/types.py b/src/sqlbuild/compiler/discovery/types.py index 916cf3e92a..f033bf0592 100644 --- a/src/sqlbuild/compiler/discovery/types.py +++ b/src/sqlbuild/compiler/discovery/types.py @@ -2,6 +2,7 @@ from __future__ import annotations +from collections.abc import Callable from dataclasses import Field from enum import StrEnum from pathlib import Path @@ -10,6 +11,10 @@ from sqlbuild.compiler.scopes.types import ScopeKind if TYPE_CHECKING: + from sqlbuild.compiler.discovery.models import ( + DiscoveredDeclarationFiles, + DiscoveredSqlModelFile, + ) from sqlbuild.provider.classes.provider import Provider type ProjectProvider = Provider @@ -38,3 +43,17 @@ class DirectorySnapshotEntry(NamedTuple): name: str is_dir: bool is_walkable_dir: bool + + +class DeclarationFilesReuse(Protocol): + """Reuse the non-model declaration files a previous compile parsed when only models changed.""" + + def declaration_files( + self, + *, + variant: str, + discover: Callable[[], DiscoveredDeclarationFiles], + discover_models: Callable[[], tuple[DiscoveredSqlModelFile, ...]], + ) -> DiscoveredDeclarationFiles: + """Return stored declaration files with freshly discovered models, or discover all.""" + ... diff --git a/tests/unit/src/sqlbuild/compiler/compile/classes/_test_types.py b/tests/unit/src/sqlbuild/compiler/compile/classes/_test_types.py index 23fea2a02f..14569d1b29 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/classes/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/compile/classes/_test_types.py @@ -46,3 +46,13 @@ class ReleasedRenderTestCase: retained_models: frozenset[str] expected_released: dict[str, bool] expected_query_sqls: dict[str, str] + + +@dataclass(frozen=True) +class DeclarationReuseTestCase: + """Paths changed since a stored compile, and whether its declaration files are reused.""" + + description: str + changed_paths: frozenset[str] | None + recorded_variant: str + expected_full_discoveries: int diff --git a/tests/unit/src/sqlbuild/compiler/compile/classes/helpers.py b/tests/unit/src/sqlbuild/compiler/compile/classes/helpers.py index 265ff3b7fc..1da570bf05 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/classes/helpers.py +++ b/tests/unit/src/sqlbuild/compiler/compile/classes/helpers.py @@ -20,7 +20,11 @@ StoredRender, ) from sqlbuild.compiler.compile.types import CompileContextKey, DiagnosticPhase, DiagnosticSeverity -from sqlbuild.compiler.discovery.models import DiscoveredSqlModelFile +from sqlbuild.compiler.discovery.models import ( + DiscoveredDeclarationFiles, + DiscoveredMacroFile, + DiscoveredSqlModelFile, +) from sqlbuild.compiler.fact_cache.main._load_fact_payload import loaded_fact_payload REGION_ENV_VAR: str = "ORDERS_REGION" @@ -156,3 +160,75 @@ def stored_query_sqls(state: RenderReuseState | None) -> dict[str, str]: def _stored_query_sql(payload: memoryview) -> str: stored: StoredRender = cast(StoredRender, loaded_fact_payload(payload)) return cast(CompileModelInput, stored.value).query_sql + + +DECLARATIONS_VARIANT: str = "10" +OTHER_DECLARATIONS_VARIANT: str = "01" + + +def declaration_files( + model_files: tuple[DiscoveredSqlModelFile, ...], +) -> DiscoveredDeclarationFiles: + """Return discovered declaration files holding one macro file and the given models.""" + + return DiscoveredDeclarationFiles( + source_files=(), + model_files=model_files, + enum_files=(), + constant_files=(), + model_schema_files=(), + sql_function_files=(), + sql_hook_files=(), + python_function_files=(), + schema_files=(), + seed_files=(), + test_files=(), + scenario_files=(), + audit_files=(), + macro_files=( + DiscoveredMacroFile( + file_path=Path("/orders/macros/currency.py"), + relative_path=Path("macros/currency.py"), + contents="def total(price: str) -> str:\n return price\n", + ), + ), + adapter_file=None, + ) + + +class CountingDiscovery: + """Discover fixed declaration files and count full and model-only discoveries.""" + + def __init__(self, model_files: tuple[DiscoveredSqlModelFile, ...]) -> None: + self.model_files: tuple[DiscoveredSqlModelFile, ...] = model_files + self.full: int = 0 + + def discover(self) -> DiscoveredDeclarationFiles: + """Discover every declaration file.""" + + self.full += 1 + return declaration_files(self.model_files) + + def discover_models(self) -> tuple[DiscoveredSqlModelFile, ...]: + """Discover model files only.""" + + return self.model_files + + +def declarations_state(*, recorded_variant: str) -> RenderReuseState: + """Return stored renders of two models, with declaration files discovered for one variant.""" + + stored: tuple[DiscoveredSqlModelFile, ...] = (model_file("orders"), model_file("customers")) + session: CompileRenderReuseSession = CompileRenderReuseSession(prior=None, changed_paths=None) + _ = session.declaration_files( + variant=recorded_variant, + discover=CountingDiscovery(stored).discover, + discover_models=CountingDiscovery(stored).discover_models, + ) + session.plan_models(model_files=stored) + with collect_compile_diagnostics(): + for item in stored: + _ = session.rendered_model(model_file=item, render=lambda item=item: _render(item)) + state: RenderReuseState | None = session.stored_state() + assert state is not None + return state diff --git a/tests/unit/src/sqlbuild/compiler/compile/classes/test_render_reuse_session.py b/tests/unit/src/sqlbuild/compiler/compile/classes/test_render_reuse_session.py index a906fdd2db..1fd8865668 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/classes/test_render_reuse_session.py +++ b/tests/unit/src/sqlbuild/compiler/compile/classes/test_render_reuse_session.py @@ -8,16 +8,22 @@ from sqlbuild.compiler.compile.classes.render_reuse_session import CompileRenderReuseSession from sqlbuild.compiler.compile.constants import COMPILE_INPUT_READS from sqlbuild.compiler.compile.models import CompileModelInput -from sqlbuild.compiler.discovery.models import DiscoveredSqlModelFile +from sqlbuild.compiler.discovery.models import DiscoveredDeclarationFiles, DiscoveredSqlModelFile from tests.unit.src.sqlbuild.compiler.compile.classes._test_types import ( CorruptRenderTestCase, + DeclarationReuseTestCase, ReleasedRenderTestCase, RenderReplayTestCase, ReusableModelsTestCase, ) from tests.unit.src.sqlbuild.compiler.compile.classes.helpers import ( + DECLARATIONS_VARIANT, + OTHER_DECLARATIONS_VARIANT, REGION_ENV_VAR, + CountingDiscovery, corrupted_session, + declaration_files, + declarations_state, edited_session, missing_description, model_file, @@ -184,5 +190,65 @@ def test_given_reused_render_when_storing_then_base_bytes_are_released_and_recov assert complete == test_case.expected_query_sqls +@pytest.mark.parametrize( + "test_case", + [ + DeclarationReuseTestCase( + description="model_edited", + changed_paths=frozenset({"models/customers.sql"}), + recorded_variant=DECLARATIONS_VARIANT, + expected_full_discoveries=0, + ), + DeclarationReuseTestCase( + description="model_added", + changed_paths=frozenset({"models/returns.sql"}), + recorded_variant=DECLARATIONS_VARIANT, + expected_full_discoveries=1, + ), + DeclarationReuseTestCase( + description="macro_edited", + changed_paths=frozenset({"macros/currency.py"}), + recorded_variant=DECLARATIONS_VARIANT, + expected_full_discoveries=1, + ), + DeclarationReuseTestCase( + description="changes_unknown", + changed_paths=None, + recorded_variant=DECLARATIONS_VARIANT, + expected_full_discoveries=1, + ), + DeclarationReuseTestCase( + description="declarations_stored_for_other_discovery", + changed_paths=frozenset({"models/customers.sql"}), + recorded_variant=OTHER_DECLARATIONS_VARIANT, + expected_full_discoveries=1, + ), + ], + ids=lambda case: case.description, +) +def test_given_stored_declarations_when_discovering_then_only_model_only_edits_reuse_them( + test_case: DeclarationReuseTestCase, +) -> None: + current: tuple[DiscoveredSqlModelFile, ...] = ( + model_file("orders"), + model_file("customers", "SELECT 2 AS customer_id"), + ) + discovery: CountingDiscovery = CountingDiscovery(current) + session: CompileRenderReuseSession = CompileRenderReuseSession( + prior=declarations_state(recorded_variant=test_case.recorded_variant), + changed_paths=test_case.changed_paths, + ) + + with collect_compile_diagnostics(): + discovered: DiscoveredDeclarationFiles = session.declaration_files( + variant=DECLARATIONS_VARIANT, + discover=discovery.discover, + discover_models=discovery.discover_models, + ) + + assert discovered == declaration_files(current) + assert discovery.full == test_case.expected_full_discoveries + + if __name__ == "__main__": raise SystemExit(pytest.main([__file__, "-n", "auto", "--dist", "loadfile"])) From d534d6c99dbdc4162390029d014d4f3f8edd465e Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Tue, 6 Oct 2026 05:13:37 +0100 Subject: [PATCH 2/3] perf(compile): serve unchanged models' analyses from the stored compile After a model-only edit, a model whose render was reused, whose recomputed analysis cache key equals the stored one and is used by no other model, and whose shared analysis inputs are unchanged may take its analysis from the stored compile instead of reading the analysis cache. The served analysis is kept only once its parents are final in this compile and each parent's final output signature equals the signature recorded with it; otherwise the model is analyzed again and its dependents are invalidated as usual. Previous signatures still come from the analysis cache, as without reuse, and an analysis is recorded for the next compile only when its recorded parent signatures match this compile's final ones, so another command advancing the analysis cache between compiles cannot leave a stale analysis served. Published shapes are computed only for models without a declared shape. --- .../_helpers/assembly/binding_waves.py | 18 +- .../compile/_helpers/assembly/project.py | 149 +++--- .../compile/classes/binding_dataflow.py | 35 +- .../compile/classes/render_reuse_session.py | 48 ++ .../compile/classes/stored_model_analyses.py | 452 ++++++++++++++++++ src/sqlbuild/compiler/compile/constants.py | 1 + .../compile/main/_build_compile_inputs.py | 1 + src/sqlbuild/compiler/compile/models.py | 40 ++ src/sqlbuild/compiler/compile/types.py | 22 +- src/sqlbuild/compiler/profiling/constants.py | 1 + src/sqlbuild/compiler/profiling/types.py | 1 + .../main/build/test_compile_json_behavior.py | 1 + .../compiler/compile/classes/_test_types.py | 23 +- .../compiler/compile/classes/helpers.py | 110 ++++- .../classes/test_stored_model_analyses.py | 166 +++++++ 15 files changed, 968 insertions(+), 100 deletions(-) create mode 100644 src/sqlbuild/compiler/compile/classes/stored_model_analyses.py create mode 100644 tests/unit/src/sqlbuild/compiler/compile/classes/test_stored_model_analyses.py diff --git a/src/sqlbuild/compiler/compile/_helpers/assembly/binding_waves.py b/src/sqlbuild/compiler/compile/_helpers/assembly/binding_waves.py index b084a8d591..337da09d48 100644 --- a/src/sqlbuild/compiler/compile/_helpers/assembly/binding_waves.py +++ b/src/sqlbuild/compiler/compile/_helpers/assembly/binding_waves.py @@ -7,7 +7,11 @@ from sqlbuild.adapter.contract.models import ExpressionInferenceProfile from sqlbuild.compiler.compile.classes.binding_dataflow import BindingDataflow -from sqlbuild.compiler.compile.models import CompileModelInput, PolyglotAnalysisResult +from sqlbuild.compiler.compile.models import ( + CompileModelInput, + DataflowReuse, + PolyglotAnalysisResult, +) from sqlbuild.compiler.compile.models import ( ModelSqlAnalysis as _ModelSqlAnalysis, ) @@ -26,8 +30,7 @@ def analyze_binding_waves( *, requests: tuple[_ModelSqlAnalysisRequest, ...], names: tuple[str, ...], - cached: dict[str, PolyglotAnalysisResult], - previous_signatures: dict[str, str], + reuse: DataflowReuse, shapes: dict[str, dict[str, str]], types: dict[str, dict[str, str]], nullability: dict[str, dict[str, InferredNullability]], @@ -40,8 +43,7 @@ def analyze_binding_waves( return BindingDataflow( requests=requests, names=names, - cached=cached, - previous_signatures=previous_signatures, + reuse=reuse, shapes=shapes, types=types, nullability=nullability, @@ -55,8 +57,7 @@ def analyze_binding_dataflow( *, requests: tuple[_ModelSqlAnalysisRequest, ...], names: tuple[str, ...], - cached: dict[str, PolyglotAnalysisResult], - previous_signatures: dict[str, str], + reuse: DataflowReuse, shapes: dict[str, dict[str, str]], types: dict[str, dict[str, str]], nullability: dict[str, dict[str, InferredNullability]], @@ -70,8 +71,7 @@ def analyze_binding_dataflow( BindingDataflow, requests=requests, names=names, - cached=cached, - previous_signatures=previous_signatures, + reuse=reuse, shapes=shapes, types=types, nullability=nullability, diff --git a/src/sqlbuild/compiler/compile/_helpers/assembly/project.py b/src/sqlbuild/compiler/compile/_helpers/assembly/project.py index 0556b04bd9..36529c2ea9 100644 --- a/src/sqlbuild/compiler/compile/_helpers/assembly/project.py +++ b/src/sqlbuild/compiler/compile/_helpers/assembly/project.py @@ -19,9 +19,7 @@ model_analysis_cache_key, model_analysis_output_signature, read_compact_analysis_cache_candidate, - read_model_analyses, record_analysis_cache_metrics, - write_model_analyses, ) from sqlbuild.compiler.compile._helpers.analysis.columns import ( substitute_placeholder_defaults, @@ -96,6 +94,7 @@ from sqlbuild.compiler.compile._helpers.render.templating import expand_template_data from sqlbuild.compiler.compile._helpers.sharing.binding import shareable_binding_prekeys from sqlbuild.compiler.compile._helpers.sql_tests.identity import build_sql_test_case_fingerprint +from sqlbuild.compiler.compile.classes.stored_model_analyses import StoredModelAnalyses from sqlbuild.compiler.compile.constants import NOT_NULL_AUDIT_NAME from sqlbuild.compiler.compile.main._scope_index_with_compile_usages import ( scope_index_with_compile_usages, @@ -132,9 +131,11 @@ CompileSqlFunctionInput, CompileSqlScenarioInput, CompileSqlTestInput, + DataflowReuse, DynamicColumnContractProof, InferredColumn, MacroContext, + ModelAnalysisCaching, PolyglotAnalysisResult, ) from sqlbuild.compiler.compile.models import ( @@ -336,7 +337,14 @@ def assemble_compiled_project( inference_profile=profile, allow_compact_analysis=allow_compact_analysis, rich_type_inference=rich_type_inference, - analysis_cache=analysis_cache, + analysis_caching=( + None + if analysis_cache is None + else ModelAnalysisCaching( + cache=analysis_cache, + reuse=inputs.analysis_reuse if analysis_model_names is None else None, + ) + ), complete_binding_schemas=complete_binding_schemas, ) scope_index: ScopeIndex = scope_index_with_compile_usages(inputs=inputs) @@ -628,11 +636,14 @@ def _analyze_model_sql_in_parallel( inference_profile: ExpressionInferenceProfile, allow_compact_analysis: bool, rich_type_inference: bool, - analysis_cache: AnalysisCacheContext | None, + analysis_caching: ModelAnalysisCaching | None, complete_binding_schemas: dict[str, dict[str, str]], ) -> dict[str, _ModelSqlAnalysis]: if not model_inputs: return {} + analysis_cache: AnalysisCacheContext | None = ( + None if analysis_caching is None else analysis_caching.cache + ) analyzed_model_names: frozenset[str] = frozenset( _model_name(model_input) for model_input in model_inputs ) @@ -646,6 +657,13 @@ def _analyze_model_sql_in_parallel( ) for model_input in model_inputs ) + stored: StoredModelAnalyses = StoredModelAnalyses( + caching=analysis_caching, + requests=requests, + column_types_by_table=column_types_by_table, + column_nullability_by_table=column_nullability_by_table, + complete_binding_schemas=complete_binding_schemas, + ) request_cache_keys: tuple[str, ...] = tuple( request.cache_key for request in requests if request.cache_key is not None ) @@ -654,7 +672,7 @@ def _analyze_model_sql_in_parallel( for request in requests ) compact_batch_plan: CompactAnalysisCachePlan | None = None - if analysis_cache is not None: + if analysis_cache is not None and not stored.served: compact_batch_plan = build_compact_analysis_cache_plan( context=analysis_cache, models=tuple( @@ -682,32 +700,24 @@ def _analyze_model_sql_in_parallel( else None ) if compact_candidate is not None: - completed_analyses: dict[str, PolyglotAnalysisResult] = {} - if analysis_cache is not None: - completed_analyses, _, _ = read_model_analyses( - context=analysis_cache, - cache_keys=request_cache_keys, - model_names=tuple(_model_name(request.model_input) for request in requests), - upstream_model_names_by_key={ - request.cache_key: _referenced_model_names( - model_input=request.model_input, - available_names=analyzed_model_names, - ) - for request in requests - if request.cache_key is not None - }, + completed_analyses: dict[str, PolyglotAnalysisResult] = ( + {} + if analysis_cache is None + else stored.read_every_model(context=analysis_cache, requests=requests) + ) + if analysis_cache is not None and all( + key in completed_analyses for key in request_cache_keys + ): + record_analysis_cache_metrics( + batch_hits=len(requests), entry_hits=0, misses=0, bypasses=0 ) - if all(key in completed_analyses for key in request_cache_keys): - record_analysis_cache_metrics( - batch_hits=len(requests), entry_hits=0, misses=0, bypasses=0 + return { + _model_name(request.model_input): _ModelSqlAnalysis( + polyglot_analysis=completed_analyses[cache_key], + placeholders=request.placeholders, ) - return { - _model_name(request.model_input): _ModelSqlAnalysis( - polyglot_analysis=completed_analyses[cache_key], - placeholders=request.placeholders, - ) - for request, cache_key in zip(requests, request_cache_keys, strict=True) - } + for request, cache_key in zip(requests, request_cache_keys, strict=True) + } try: compact_analyses: tuple[_ModelSqlAnalysis, ...] = _analyze_model_sql_requests( requests=requests, @@ -745,6 +755,7 @@ def _analyze_model_sql_in_parallel( complete_binding_schemas=complete_binding_schemas, inference_profile=inference_profile, ) + stored.record_none() return { _model_name(model_input): analysis for model_input, analysis in zip(model_inputs, compact_analyses, strict=True) @@ -756,7 +767,7 @@ def _analyze_model_sql_in_parallel( previous_signatures: dict[str, str] cached_output_signatures_by_key: dict[str, str] cached_analyses, previous_signatures, cached_output_signatures_by_key = ( - read_model_analyses( + stored.read( context=analysis_cache, cache_keys=entry_cache_keys, model_names=tuple(_model_name(request.model_input) for request in requests), @@ -798,8 +809,11 @@ def _analyze_model_sql_in_parallel( analyses, cached_analyses = analyze_binding_dataflow( requests=requests, names=tuple(_model_name(request.model_input) for request in requests), - cached=cached_analyses, - previous_signatures=previous_signatures, + reuse=DataflowReuse( + cached=cached_analyses, + previous_signatures=previous_signatures, + served={name: analysis.dependencies for name, analysis in stored.served.items()}, + ), shapes=complete_binding_schemas, types=column_types_by_table, nullability=column_nullability_by_table, @@ -898,6 +912,11 @@ def _analyze_model_sql_in_parallel( invalidated_names: set[str] = _downstream_model_names( model_inputs=model_inputs, changed_names=changed_signature_names, + ) | _with_downstream( + model_inputs=model_inputs, + names=stored.stale_names( + requests=requests, cached=cached_analyses, current_signatures=current_signatures_by_name + ), ) if invalidated_names and not dependency_ordered: invalidated_requests: tuple[_ModelSqlAnalysisRequest, ...] = tuple( @@ -959,28 +978,17 @@ def _analyze_model_sql_in_parallel( for model_name, invalidated_analysis in invalidated_analyses_by_name.items() } ) - if analysis_cache is not None: - with record_compile_timing("cache_publication_ms"): - write_model_analyses( - context=analysis_cache, - analyses_by_key={ - request.cache_key: analysis.polyglot_analysis - for request, analysis in zip(requests, analyses, strict=True) - if request.cache_key is not None - and ( - request.cache_key not in cached_analyses - or _model_name(request.model_input) in invalidated_names - ) - }, - latest_analyses_by_model=analyses_to_record_by_name, - dependency_signatures_by_key=_dependency_signatures_by_key( - requests=requests, - cached_analyses=cached_analyses, - invalidated_names=invalidated_names, - current_signatures_by_name=current_signatures_by_name, - analyzed_model_names=analyzed_model_names, - ), - ) + stored.publish( + context=analysis_cache, + requests=requests, + results=tuple(analysis.polyglot_analysis for analysis in analyses), + cached=cached_analyses, + cached_output_signatures=cached_output_signatures_by_key, + invalidated_names=invalidated_names, + latest_by_model=analyses_to_record_by_name, + previous_signatures=previous_signatures, + current_signatures=current_signatures_by_name, + ) return { _model_name(model_input): analysis for model_input, analysis in zip(model_inputs, analyses, strict=True) @@ -1197,7 +1205,7 @@ def _complete_inferred_bindings( ) results[index] = replace(results[index], polyglot_analysis=analysis) star_known: bool = not analysis.has_star or (analysis.star_resolved and inputs_known) - if analysis.columns and star_known: + if analysis.columns and star_known and name not in complete_schemas: complete_schemas.setdefault( name, published_model_shape( @@ -1257,33 +1265,6 @@ def _complete_inferred_bindings( return tuple(results) -def _dependency_signatures_by_key( - *, - requests: tuple[_ModelSqlAnalysisRequest, ...], - cached_analyses: dict[str, PolyglotAnalysisResult], - invalidated_names: set[str], - current_signatures_by_name: dict[str, str], - analyzed_model_names: frozenset[str], -) -> dict[str, dict[str, str]]: - dependencies_by_key: dict[str, dict[str, str]] = {} - for request in requests: - cache_key: str | None = request.cache_key - if cache_key is None or ( - cache_key in cached_analyses - and _model_name(request.model_input) not in invalidated_names - ): - continue - dependencies_by_key[cache_key] = { - upstream_name: current_signatures_by_name[upstream_name] - for upstream_name in _referenced_model_names( - model_input=request.model_input, - available_names=analyzed_model_names, - ) - if upstream_name in current_signatures_by_name - } - return dependencies_by_key - - def _analyze_model_sql( *, request: _ModelSqlAnalysisRequest, @@ -1377,6 +1358,10 @@ def _should_recover_cte_facts(model_input: CompileModelInput) -> bool: ) +def _with_downstream(*, model_inputs: tuple[CompileModelInput, ...], names: set[str]) -> set[str]: + return names | _downstream_model_names(model_inputs=model_inputs, changed_names=names) + + def _model_name(model_input: CompileModelInput) -> str: return model_input.model_file.file_path.stem diff --git a/src/sqlbuild/compiler/compile/classes/binding_dataflow.py b/src/sqlbuild/compiler/compile/classes/binding_dataflow.py index 0909ba58c7..638f39f370 100644 --- a/src/sqlbuild/compiler/compile/classes/binding_dataflow.py +++ b/src/sqlbuild/compiler/compile/classes/binding_dataflow.py @@ -19,7 +19,9 @@ published_model_shape, referenced_model_names, ) +from sqlbuild.compiler.compile.classes.stored_model_analyses import dependencies_current from sqlbuild.compiler.compile.models import ( + DataflowReuse, ModelSqlAnalysis, ModelSqlAnalysisRequest, PolyglotAnalysisResult, @@ -42,8 +44,7 @@ def __init__( *, requests: tuple[ModelSqlAnalysisRequest, ...], names: tuple[str, ...], - cached: dict[str, PolyglotAnalysisResult], - previous_signatures: dict[str, str], + reuse: DataflowReuse, shapes: dict[str, dict[str, str]], types: dict[str, dict[str, str]], nullability: dict[str, dict[str, InferredNullability]], @@ -53,8 +54,15 @@ def __init__( ) -> None: self._requests: tuple[ModelSqlAnalysisRequest, ...] = requests self._names: tuple[str, ...] = names - self._cached: dict[str, PolyglotAnalysisResult] = cached self._by_name: dict[str, ModelSqlAnalysisRequest] = dict(zip(names, requests, strict=True)) + served_keys: frozenset[str | None] = frozenset( + self._by_name[name].cache_key for name in reuse.served if name in self._by_name + ) + self._cached: dict[str, PolyglotAnalysisResult] = { + key: analysis for key, analysis in reuse.cached.items() if key not in served_keys + } + self._served: dict[str, dict[str, str]] = reuse.served + self._signatures: dict[str, str] = {} available_names: frozenset[str] = frozenset(names) self._dependencies: dict[str, tuple[str, ...]] = { name: referenced_model_names( @@ -62,11 +70,11 @@ def __init__( ) for name, request in self._by_name.items() } - self._previous_signatures: dict[str, str] = previous_signatures + self._previous_signatures: dict[str, str] = reuse.previous_signatures self._complete_shapes: dict[str, dict[str, str]] = dict(shapes) self._available_types: dict[str, dict[str, str]] = dict(types) self._available_nullability: dict[str, dict[str, InferredNullability]] = dict(nullability) - self._reusable: dict[str, PolyglotAnalysisResult] = dict(cached) + self._reusable: dict[str, PolyglotAnalysisResult] = dict(reuse.cached) self._profile: ExpressionInferenceProfile = profile self._analyze: Callable[..., tuple[ModelSqlAnalysis, ...]] = analyze self._complete: Callable[..., tuple[ModelSqlAnalysis, ...]] = complete @@ -251,6 +259,8 @@ def _analyze_ready(self, ready: tuple[str, ...]) -> None: self._changed.add(name) if request.cache_key is not None: self._reusable.pop(request.cache_key, None) + if not self._served_parents_final(name) and request.cache_key is not None: + self._reusable.pop(request.cache_key, None) bindings: dict[str, dict[str, str]] | None = ( None if request.binding_schema is None @@ -288,11 +298,24 @@ def _analyze_ready(self, ready: tuple[str, ...]) -> None: self._results[name] = result self._publish(name=name, request=request, result=result) + def _served_parents_final(self, name: str) -> bool: + """Return whether a stored analysis, if served, was made from the parents' final state.""" + + dependencies: dict[str, str] | None = self._served.get(name) + return dependencies is None or dependencies_current( + dependencies=dependencies, + parent_signatures={ + parent: self._signatures.get(parent) for parent in self._dependencies[name] + }, + ) + def _publish( self, *, name: str, request: ModelSqlAnalysisRequest, result: ModelSqlAnalysis ) -> None: analysis: PolyglotAnalysisResult = result.polyglot_analysis - if model_analysis_output_signature(analysis) != self._previous_signatures.get(name): + signature: str = model_analysis_output_signature(analysis) + self._signatures[name] = signature + if signature != self._previous_signatures.get(name): self._changed.add(name) required: frozenset[str] = binding_relation_names(request.model_input.references) star_known: bool = not analysis.has_star or ( diff --git a/src/sqlbuild/compiler/compile/classes/render_reuse_session.py b/src/sqlbuild/compiler/compile/classes/render_reuse_session.py index 868c5349b7..8cf0c62c8d 100644 --- a/src/sqlbuild/compiler/compile/classes/render_reuse_session.py +++ b/src/sqlbuild/compiler/compile/classes/render_reuse_session.py @@ -6,6 +6,7 @@ import threading from collections.abc import Callable from dataclasses import replace +from pathlib import Path from typing import cast from sqlbuild.compiler.compile._helpers.diagnostics.collector import ( @@ -15,6 +16,7 @@ from sqlbuild.compiler.compile.classes.compile_input_reads import CompileInputReads from sqlbuild.compiler.compile.constants import ( COMPILE_INPUT_READS, + RENDER_REUSE_ANALYSIS_PREFIX, RENDER_REUSE_DECLARATIONS_PREFIX, RENDER_REUSE_MODEL_AUDITS_PREFIX, ) @@ -23,6 +25,7 @@ CompileModelInput, CompilerDiagnostic, RenderReuseState, + StoredModelAnalysis, StoredRender, ) from sqlbuild.compiler.discovery.models import DiscoveredDeclarationFiles, DiscoveredSqlModelFile @@ -76,6 +79,8 @@ def __init__( self._serialized_models: dict[str, memoryview] = {} self._serialized_groups: dict[str, memoryview] = {} self._reused_paths: set[str] = set() + self._stored_analyses: dict[str, StoredModelAnalysis] = {} + self._stored_analysis_payloads: dict[str, memoryview] = {} def claim(self) -> bool: """Return True for the first claim only; later project renders neither reuse nor record.""" @@ -243,9 +248,36 @@ def reused_any(self) -> bool: return bool(self._decoded_models or self._decoded_groups) + def reused_model_names(self) -> frozenset[str]: + """Return the names of models whose stored renders this compile reused.""" + + return frozenset(Path(path).stem for path in self._reused_paths) + + def stored_analyses(self) -> dict[str, StoredModelAnalysis]: + """Return stored analyses of reused models by model name.""" + + return self._stored_analyses + + def record_analyses(self, *, analyses: dict[str, StoredModelAnalysis | None]) -> None: + """Record every model's analysis for the next compile; absent or None drops it.""" + + for path in self._model_paths or (): + name: str = f"{RENDER_REUSE_ANALYSIS_PREFIX}{path}" + model_name: str = Path(path).stem + analysis: StoredModelAnalysis | None = analyses.get(model_name) + stored: StoredRender = StoredRender( + value=analysis, diagnostics=(), environment_names=() + ) + if analysis is not None and analysis is self._stored_analyses.get(model_name): + self._group_records[name] = self._stored_analysis_payloads.get(name) + self._decoded_groups[name] = stored + else: + self._pending_groups.append((name, stored)) + def release_stored(self) -> None: """Drop stored base bytes once every render is decoded, recording reused renders as None.""" + self._keep_stored_analyses() self._prior = None self._reusable = False for records, retained in ( @@ -255,6 +287,22 @@ def release_stored(self) -> None: for key in records.keys() - retained: records[key] = None + def _keep_stored_analyses(self) -> None: + prior: RenderReuseState | None = self._prior + if prior is None or not self._reusable: + return + for path in self._reused_paths: + name: str = f"{RENDER_REUSE_ANALYSIS_PREFIX}{path}" + payload: memoryview | None = prior.group_payloads.get(name) + stored: StoredRender | None = None if payload is None else _loaded_render(payload) + if payload is None or stored is None: + continue + if not isinstance(stored.value, StoredModelAnalysis): + continue + self._stored_analyses[Path(path).stem] = stored.value + if name in self._retained_groups: + self._stored_analysis_payloads[name] = memoryview(bytes(payload)) + def stored_state(self, *, complete: bool = False) -> RenderReuseState | None: """Serialize this compile's renders; with complete, re-serialize released reused ones.""" diff --git a/src/sqlbuild/compiler/compile/classes/stored_model_analyses.py b/src/sqlbuild/compiler/compile/classes/stored_model_analyses.py new file mode 100644 index 0000000000..1204f1e697 --- /dev/null +++ b/src/sqlbuild/compiler/compile/classes/stored_model_analyses.py @@ -0,0 +1,452 @@ +"""Serve unchanged models' analyses from the previous compile instead of the analysis cache.""" + +from __future__ import annotations + +import hashlib +from collections import Counter + +import orjson + +from sqlbuild.compiler.compile._helpers.analysis.cache import ( + model_analysis_output_signature, + read_model_analyses, + write_model_analyses, +) +from sqlbuild.compiler.compile._helpers.assembly.semantic_shapes import referenced_model_names +from sqlbuild.compiler.compile.models import ( + AnalysisCacheContext, + CompileModelInput, + ModelAnalysisCaching, + ModelSqlAnalysisRequest, + PolyglotAnalysisResult, + StoredAnalysisScope, + StoredModelAnalysis, +) +from sqlbuild.compiler.compile.types import ModelAnalysisReuse +from sqlbuild.compiler.lineage.types import InferredNullability +from sqlbuild.compiler.profiling.main._metric import record_compile_metric +from sqlbuild.compiler.profiling.main.record import record_compile_timing + + +class StoredModelAnalyses: + """Analyses a compile may serve from the previous compile, and what it records for the next.""" + + def __init__( + self, + *, + caching: ModelAnalysisCaching | None, + requests: tuple[ModelSqlAnalysisRequest, ...], + column_types_by_table: dict[str, dict[str, str]], + column_nullability_by_table: dict[str, dict[str, InferredNullability]], + complete_binding_schemas: dict[str, dict[str, str]], + ) -> None: + self._scope: StoredAnalysisScope | None = _stored_analysis_scope( + caching=caching, + requests=requests, + column_types_by_table=column_types_by_table, + column_nullability_by_table=column_nullability_by_table, + complete_binding_schemas=complete_binding_schemas, + ) + + @property + def served(self) -> dict[str, StoredModelAnalysis]: + """Return stored analyses that may serve their own model, whose cache key no other uses.""" + + return {} if self._scope is None else self._scope.stored + + def stale_names( + self, + *, + requests: tuple[ModelSqlAnalysisRequest, ...], + cached: dict[str, PolyglotAnalysisResult], + current_signatures: dict[str, str], + ) -> set[str]: + """Return served models whose parents' final signatures differ from their stored ones.""" + + analyzed_names: frozenset[str] = frozenset( + _name(request.model_input) for request in requests + ) + served: dict[str, StoredModelAnalysis] = self.served + return { + name + for request in requests + if (name := _name(request.model_input)) in served + and request.cache_key is not None + and cached.get(request.cache_key) is served[name].analysis + and not dependencies_current( + dependencies=served[name].dependencies, + parent_signatures=_parent_signatures( + request=request, + signatures=current_signatures, + analyzed_names=analyzed_names, + ), + ) + } + + def read( + self, + *, + context: AnalysisCacheContext, + cache_keys: tuple[str, ...], + model_names: tuple[str, ...], + upstream_model_names_by_key: dict[str, tuple[str, ...]], + ) -> tuple[dict[str, PolyglotAnalysisResult], dict[str, str], dict[str, str]]: + """Read the analysis cache for keys without a stored analysis; serve the rest.""" + + stored: dict[str, StoredModelAnalysis] = self.served + stored_by_key: dict[str, StoredModelAnalysis] = { + analysis.cache_key: analysis for analysis in stored.values() + } + analyses, signatures, output_signatures = read_model_analyses( + context=context, + cache_keys=tuple(key for key in cache_keys if key not in stored_by_key), + model_names=model_names, + upstream_model_names_by_key={ + key: names + for key, names in upstream_model_names_by_key.items() + if key not in stored_by_key + }, + ) + for key in cache_keys: + served: StoredModelAnalysis | None = stored_by_key.get(key) + if served is not None: + analyses[key] = served.analysis + output_signatures[key] = served.output_signature + return analyses, signatures, output_signatures + + def read_every_model( + self, *, context: AnalysisCacheContext, requests: tuple[ModelSqlAnalysisRequest, ...] + ) -> dict[str, PolyglotAnalysisResult]: + """Read every model's analysis from the cache; record them when the cache held them all.""" + + analyzed_names: frozenset[str] = frozenset( + _name(request.model_input) for request in requests + ) + cache_keys: tuple[str, ...] = tuple( + request.cache_key for request in requests if request.cache_key is not None + ) + analyses, signatures, output_signatures = read_model_analyses( + context=context, + cache_keys=cache_keys, + model_names=tuple(_name(request.model_input) for request in requests), + upstream_model_names_by_key={ + request.cache_key: referenced_model_names( + model_input=request.model_input, available_names=analyzed_names + ) + for request in requests + if request.cache_key is not None + }, + ) + if all(key in analyses for key in cache_keys): + self._record( + requests=requests, + cached=analyses, + cached_output_signatures=output_signatures, + written={}, + written_dependencies={}, + previous_signatures=signatures, + current_signatures={ + _name(request.model_input): output_signatures[request.cache_key] + for request in requests + if request.cache_key is not None + }, + ) + return analyses + + def record_none(self) -> None: + """Record no analyses, so the next compile analyzes or reads every model from the cache.""" + + if self._scope is not None: + self._scope.reuse.record_analyses(analyses={}) + + def publish( + self, + *, + context: AnalysisCacheContext, + requests: tuple[ModelSqlAnalysisRequest, ...], + results: tuple[PolyglotAnalysisResult, ...], + cached: dict[str, PolyglotAnalysisResult], + cached_output_signatures: dict[str, str], + invalidated_names: set[str], + latest_by_model: dict[str, PolyglotAnalysisResult], + previous_signatures: dict[str, str], + current_signatures: dict[str, str], + ) -> None: + """Write new and invalidated analyses to the cache, then record them for the next run.""" + + analyzed_names: frozenset[str] = frozenset( + _name(request.model_input) for request in requests + ) + written: dict[str, PolyglotAnalysisResult] = { + request.cache_key: result + for request, result in zip(requests, results, strict=True) + if request.cache_key is not None + and (request.cache_key not in cached or _name(request.model_input) in invalidated_names) + } + written_dependencies: dict[str, dict[str, str]] = { + key: _dependency_signatures( + request=request, + current_signatures=current_signatures, + analyzed_names=analyzed_names, + ) + for request in requests + if (key := request.cache_key) is not None and key in written + } + record_compile_metric( + metric="analysis_reuse_hits", + value=sum( + request.cache_key is not None + and (served := self.served.get(_name(request.model_input))) is not None + and cached.get(request.cache_key) is served.analysis + and _name(request.model_input) not in invalidated_names + for request in requests + ), + ) + with record_compile_timing("cache_publication_ms"): + write_model_analyses( + context=context, + analyses_by_key=written, + latest_analyses_by_model=latest_by_model, + dependency_signatures_by_key=written_dependencies, + ) + self._record( + requests=requests, + cached=cached, + cached_output_signatures=cached_output_signatures, + written=written, + written_dependencies=written_dependencies, + previous_signatures=previous_signatures, + current_signatures=current_signatures, + ) + + def _record( + self, + *, + requests: tuple[ModelSqlAnalysisRequest, ...], + cached: dict[str, PolyglotAnalysisResult], + cached_output_signatures: dict[str, str], + written: dict[str, PolyglotAnalysisResult], + written_dependencies: dict[str, dict[str, str]], + previous_signatures: dict[str, str], + current_signatures: dict[str, str], + ) -> None: + if self._scope is None: + return + self._scope.reuse.record_analyses( + analyses=_recorded_analyses( + scope=self._scope, + requests=requests, + cached=cached, + cached_output_signatures=cached_output_signatures, + written=written, + written_dependencies=written_dependencies, + previous_signatures=previous_signatures, + current_signatures=current_signatures, + ) + ) + + +def dependencies_current( + *, dependencies: dict[str, str], parent_signatures: dict[str, str | None] +) -> bool: + """Return whether a stored analysis was made from parents with these final signatures.""" + + return dependencies == parent_signatures + + +def _stored_analysis_scope( + *, + caching: ModelAnalysisCaching | None, + requests: tuple[ModelSqlAnalysisRequest, ...], + column_types_by_table: dict[str, dict[str, str]], + column_nullability_by_table: dict[str, dict[str, InferredNullability]], + complete_binding_schemas: dict[str, dict[str, str]], +) -> StoredAnalysisScope | None: + if caching is None or caching.reuse is None: + return None + context: str = analysis_reuse_context( + cache=caching.cache, + model_names=frozenset(_name(request.model_input) for request in requests), + column_types_by_table=column_types_by_table, + column_nullability_by_table=column_nullability_by_table, + complete_binding_schemas=complete_binding_schemas, + ) + stored: dict[str, StoredModelAnalysis] = _reusable_stored_analyses( + reuse=caching.reuse, context=context, requests=requests + ) + record_compile_metric(metric="analysis_reuse_hits", value=0) + return StoredAnalysisScope(reuse=caching.reuse, context=context, stored=stored) + + +def analysis_reuse_context( + *, + cache: AnalysisCacheContext, + model_names: frozenset[str], + column_types_by_table: dict[str, dict[str, str]], + column_nullability_by_table: dict[str, dict[str, InferredNullability]], + complete_binding_schemas: dict[str, dict[str, str]], +) -> str: + """Digest every analysis input shared by all models, including non-model relations.""" + + relations: dict[str, list[object]] = { + name: _relation_inputs( + name=name, + column_types_by_table=column_types_by_table, + column_nullability_by_table=column_nullability_by_table, + complete_binding_schemas=complete_binding_schemas, + ) + for name in ( + column_types_by_table.keys() + | column_nullability_by_table.keys() + | complete_binding_schemas.keys() + ) + - model_names + } + return hashlib.sha256( + orjson.dumps( + { + "shared_fingerprint": cache.shared_fingerprint, + "signature_namespace": cache.signature_namespace, + "relations": relations, + }, + option=orjson.OPT_SORT_KEYS, + ) + ).hexdigest() + + +def _relation_inputs( + *, + name: str, + column_types_by_table: dict[str, dict[str, str]], + column_nullability_by_table: dict[str, dict[str, InferredNullability]], + complete_binding_schemas: dict[str, dict[str, str]], +) -> list[object]: + nullability: dict[str, InferredNullability] = column_nullability_by_table.get(name, {}) + return [ + column_types_by_table.get(name), + {column: value.value for column, value in nullability.items()}, + complete_binding_schemas.get(name), + ] + + +def _reusable_stored_analyses( + *, + reuse: ModelAnalysisReuse, + context: str, + requests: tuple[ModelSqlAnalysisRequest, ...], +) -> dict[str, StoredModelAnalysis]: + reused_names: frozenset[str] = reuse.reused_model_names() + stored: dict[str, StoredModelAnalysis] = reuse.stored_analyses() + key_counts: Counter[str | None] = Counter(request.cache_key for request in requests) + reusable: dict[str, StoredModelAnalysis] = {} + for request in requests: + name: str = _name(request.model_input) + analysis: StoredModelAnalysis | None = stored.get(name) + if ( + analysis is not None + and analysis.context == context + and name in reused_names + and request.cache_key is not None + and analysis.cache_key == request.cache_key + and key_counts[request.cache_key] == 1 + ): + reusable[name] = analysis + return reusable + + +def _parent_signatures( + *, + request: ModelSqlAnalysisRequest, + signatures: dict[str, str], + analyzed_names: frozenset[str], +) -> dict[str, str | None]: + return { + upstream_name: signatures.get(upstream_name) + for upstream_name in referenced_model_names( + model_input=request.model_input, available_names=analyzed_names + ) + } + + +def _dependency_signatures( + *, + request: ModelSqlAnalysisRequest, + current_signatures: dict[str, str], + analyzed_names: frozenset[str], +) -> dict[str, str]: + return { + upstream_name: current_signatures[upstream_name] + for upstream_name in referenced_model_names( + model_input=request.model_input, available_names=analyzed_names + ) + if upstream_name in current_signatures + } + + +def _recorded_analyses( + *, + scope: StoredAnalysisScope, + requests: tuple[ModelSqlAnalysisRequest, ...], + cached: dict[str, PolyglotAnalysisResult], + cached_output_signatures: dict[str, str], + written: dict[str, PolyglotAnalysisResult], + written_dependencies: dict[str, dict[str, str]], + previous_signatures: dict[str, str], + current_signatures: dict[str, str], +) -> dict[str, StoredModelAnalysis | None]: + analyzed_names: frozenset[str] = frozenset(_name(request.model_input) for request in requests) + recorded: dict[str, StoredModelAnalysis | None] = {} + for request in requests: + name: str = _name(request.model_input) + key: str | None = request.cache_key + written_analysis: PolyglotAnalysisResult | None = None if key is None else written.get(key) + cached_analysis: PolyglotAnalysisResult | None = None if key is None else cached.get(key) + previous: StoredModelAnalysis | None = scope.stored.get(name) + upstream: tuple[str, ...] = referenced_model_names( + model_input=request.model_input, available_names=analyzed_names + ) + parents: dict[str, str | None] = _parent_signatures( + request=request, signatures=current_signatures, analyzed_names=analyzed_names + ) + if key is None or (written_analysis is None and cached_analysis is None): + recorded[name] = None + elif written_analysis is not None: + recorded[name] = StoredModelAnalysis( + context=scope.context, + cache_key=key, + analysis=written_analysis, + output_signature=model_analysis_output_signature(written_analysis), + dependencies=written_dependencies.get(key, {}), + signature=current_signatures.get(name), + ) + elif ( + previous is not None + and previous.cache_key == key + and previous.analysis is cached_analysis + and previous.signature == current_signatures.get(name) + and dependencies_current(dependencies=previous.dependencies, parent_signatures=parents) + ): + recorded[name] = previous + elif cached_analysis is not None and all( + upstream_name in previous_signatures + and previous_signatures[upstream_name] == parents[upstream_name] + for upstream_name in upstream + ): + recorded[name] = StoredModelAnalysis( + context=scope.context, + cache_key=key, + analysis=cached_analysis, + output_signature=cached_output_signatures.get(key) + or model_analysis_output_signature(cached_analysis), + dependencies={ + upstream_name: previous_signatures[upstream_name] for upstream_name in upstream + }, + signature=current_signatures.get(name), + ) + else: + recorded[name] = None + return recorded + + +def _name(model_input: CompileModelInput) -> str: + return model_input.model_file.file_path.stem diff --git a/src/sqlbuild/compiler/compile/constants.py b/src/sqlbuild/compiler/compile/constants.py index 4d030eda5a..01ab10b9f3 100644 --- a/src/sqlbuild/compiler/compile/constants.py +++ b/src/sqlbuild/compiler/compile/constants.py @@ -181,6 +181,7 @@ RENDER_REUSE_ATTACHED_AUDITS_GROUP: str = "source_and_seed_audits" RENDER_REUSE_MODEL_AUDITS_PREFIX: str = "model_audits:" RENDER_REUSE_DECLARATIONS_PREFIX: str = "declarations:" +RENDER_REUSE_ANALYSIS_PREFIX: str = "analysis:" SQL_TEST_FACT_CACHE_NAMESPACE: str = "sql-tests" SQL_TEST_FACT_CACHE_ALGORITHM: str = "expanded-sql-test-ctes-v2" SQL_TEST_EXPECTED_MODELS_FACT_ALGORITHM: str = "sql-test-expected-models-v2" diff --git a/src/sqlbuild/compiler/compile/main/_build_compile_inputs.py b/src/sqlbuild/compiler/compile/main/_build_compile_inputs.py index 4d49ec40b0..ebee19a4b6 100644 --- a/src/sqlbuild/compiler/compile/main/_build_compile_inputs.py +++ b/src/sqlbuild/compiler/compile/main/_build_compile_inputs.py @@ -239,6 +239,7 @@ def build_compile_inputs( external_sql_reference_resolver=external_sql_reference_resolver, scope_index=declaration_scope.index, declaration_scope=declaration_scope, + analysis_reuse=render_reuse, ) diff --git a/src/sqlbuild/compiler/compile/models.py b/src/sqlbuild/compiler/compile/models.py index e07afcffe8..17de6fa5f4 100644 --- a/src/sqlbuild/compiler/compile/models.py +++ b/src/sqlbuild/compiler/compile/models.py @@ -21,6 +21,7 @@ DiagnosticPhase, DiagnosticSeverity, FunctionLanguage, + ModelAnalysisReuse, SqlTestMode, TypedSqlValueRenderer, ) @@ -775,6 +776,44 @@ class ModelSqlAnalysis: validated_schema: dict[str, dict[str, str]] | None = None +@dataclass(frozen=True) +class DataflowReuse: + """Analyses a binding dataflow may reuse, and the signatures that decide whether it may.""" + + cached: dict[str, PolyglotAnalysisResult] + previous_signatures: dict[str, str] + served: dict[str, dict[str, str]] + + +@dataclass(frozen=True) +class StoredModelAnalysis: + """One model's analysis as the analysis cache held it after a compile, kept for reuse.""" + + context: str + cache_key: str + analysis: PolyglotAnalysisResult + output_signature: str + dependencies: dict[str, str] + signature: str | None + + +@dataclass(frozen=True) +class StoredAnalysisScope: + """Stored analyses one compile may serve, and the session recording its own analyses.""" + + reuse: ModelAnalysisReuse + context: str + stored: dict[str, StoredModelAnalysis] + + +@dataclass(frozen=True) +class ModelAnalysisCaching: + """The analysis cache one compile reads and writes, and its stored analysis reuse.""" + + cache: AnalysisCacheContext + reuse: ModelAnalysisReuse | None = None + + @dataclass(frozen=True) class ModelSqlAnalysisRequest: """Inputs to one model's SQL analysis.""" @@ -889,6 +928,7 @@ class CompileProjectInputs: external_sql_reference_resolver: ExternalSqlReferenceResolver | None = None scope_index: ScopeIndex = field(default_factory=ScopeIndex) declaration_scope: DeclarationScopeBuild | None = field(default=None, repr=False, compare=False) + analysis_reuse: ModelAnalysisReuse | None = field(default=None, repr=False, compare=False) @dataclass(frozen=True) diff --git a/src/sqlbuild/compiler/compile/types.py b/src/sqlbuild/compiler/compile/types.py index 9820257fd2..e049aee727 100644 --- a/src/sqlbuild/compiler/compile/types.py +++ b/src/sqlbuild/compiler/compile/types.py @@ -8,7 +8,11 @@ from sqlbuild.sql_values.models import SqlValue if TYPE_CHECKING: - from sqlbuild.compiler.compile.models import CompactBatchPreparation, CompileProjectInputs + from sqlbuild.compiler.compile.models import ( + CompactBatchPreparation, + CompileProjectInputs, + StoredModelAnalysis, + ) from sqlbuild.compiler.sql_analysis.models import SqlLexicalSyntax @@ -39,6 +43,22 @@ class CompileInputsReadyCallback(Protocol): def __call__(self, *, inputs: CompileProjectInputs) -> None: ... +class ModelAnalysisReuse(Protocol): + """Analyses a previous compile stored for models whose renders this compile reused.""" + + def reused_model_names(self) -> frozenset[str]: + """Return the names of models whose stored renders this compile reused.""" + ... + + def stored_analyses(self) -> dict[str, StoredModelAnalysis]: + """Return stored analyses of reused models by model name.""" + ... + + def record_analyses(self, *, analyses: dict[str, StoredModelAnalysis | None]) -> None: + """Record every model's analysis for the next compile; absent or None drops it.""" + ... + + class AttachedAuditTargetKind(StrEnum): MODEL = "model" SOURCE = "source" diff --git a/src/sqlbuild/compiler/profiling/constants.py b/src/sqlbuild/compiler/profiling/constants.py index b969bb04cf..8d8a0a4009 100644 --- a/src/sqlbuild/compiler/profiling/constants.py +++ b/src/sqlbuild/compiler/profiling/constants.py @@ -22,6 +22,7 @@ "analysis_entry_cache_hits", "analysis_cache_misses", "analysis_cache_bypasses", + "analysis_reuse_hits", "fact_cache_hits", "fact_cache_misses", "render_reuse_hits", diff --git a/src/sqlbuild/compiler/profiling/types.py b/src/sqlbuild/compiler/profiling/types.py index f3e9a1c2b4..a1bddfd128 100644 --- a/src/sqlbuild/compiler/profiling/types.py +++ b/src/sqlbuild/compiler/profiling/types.py @@ -22,6 +22,7 @@ "analysis_entry_cache_hits", "analysis_cache_misses", "analysis_cache_bypasses", + "analysis_reuse_hits", "fact_cache_hits", "fact_cache_misses", "render_reuse_hits", diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py index 7701ffa63f..9e8f5cfd02 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py @@ -88,6 +88,7 @@ def test_given_waffle_shop_when_running_compile_json_then_it_reports_offline_que "analysis_entry_cache_hits", "analysis_cache_misses", "analysis_cache_bypasses", + "analysis_reuse_hits", "fact_cache_hits", "fact_cache_misses", "render_reuse_hits", diff --git a/tests/unit/src/sqlbuild/compiler/compile/classes/_test_types.py b/tests/unit/src/sqlbuild/compiler/compile/classes/_test_types.py index 14569d1b29..82cd55965b 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/classes/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/compile/classes/_test_types.py @@ -1,4 +1,4 @@ -"""Test case types for the compile render reuse session.""" +"""Test case types for the compile render reuse session and stored model analyses.""" from __future__ import annotations @@ -56,3 +56,24 @@ class DeclarationReuseTestCase: changed_paths: frozenset[str] | None recorded_variant: str expected_full_discoveries: int + + +@dataclass(frozen=True) +class StoredAnalysisScopeTestCase: + """Which renders a compile reused, how analyses were stored, and which may be served.""" + + description: str + reused_models: frozenset[str] + rekeyed_models: frozenset[str] + twinned_models: frozenset[str] + stored_column_types: dict[str, dict[str, str]] + expected_served: frozenset[str] + + +@dataclass(frozen=True) +class StaleServedAnalysisTestCase: + """Final parent signatures in this compile, and which served analyses they make stale.""" + + description: str + current_signatures: dict[str, str] + expected_stale: set[str] diff --git a/tests/unit/src/sqlbuild/compiler/compile/classes/helpers.py b/tests/unit/src/sqlbuild/compiler/compile/classes/helpers.py index 1da570bf05..9482548c3a 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/classes/helpers.py +++ b/tests/unit/src/sqlbuild/compiler/compile/classes/helpers.py @@ -1,4 +1,4 @@ -"""Shared builders for render reuse session unit tests.""" +"""Shared builders for render reuse session and stored model analysis unit tests.""" from __future__ import annotations @@ -12,11 +12,21 @@ report_compile_diagnostic, ) from sqlbuild.compiler.compile.classes.render_reuse_session import CompileRenderReuseSession +from sqlbuild.compiler.compile.classes.stored_model_analyses import ( + StoredModelAnalyses, + analysis_reuse_context, +) from sqlbuild.compiler.compile.constants import COMPILE_INPUT_READS from sqlbuild.compiler.compile.models import ( + AnalysisCacheContext, CompileModelInput, CompilerDiagnostic, + CompileSqlReference, + ModelAnalysisCaching, + ModelSqlAnalysisRequest, + PolyglotAnalysisResult, RenderReuseState, + StoredModelAnalysis, StoredRender, ) from sqlbuild.compiler.compile.types import CompileContextKey, DiagnosticPhase, DiagnosticSeverity @@ -26,8 +36,13 @@ DiscoveredSqlModelFile, ) from sqlbuild.compiler.fact_cache.main._load_fact_payload import loaded_fact_payload +from sqlbuild.compiler.references.types import SqlReferenceKind REGION_ENV_VAR: str = "ORDERS_REGION" +ORDERS_CACHE: AnalysisCacheContext = AnalysisCacheContext( + root=Path("/orders/target/cache"), shared_fingerprint="orders-fingerprint" +) +RAW_ORDERS_TYPES: dict[str, dict[str, str]] = {"raw_orders": {"order_id": "INTEGER"}} def model_file(name: str, sql: str = "SELECT 1 AS order_id") -> DiscoveredSqlModelFile: @@ -232,3 +247,96 @@ def declarations_state(*, recorded_variant: str) -> RenderReuseState: state: RenderReuseState | None = session.stored_state() assert state is not None return state + + +def analyzed_model(name: str, *upstream: str) -> CompileModelInput: + """Return one model input reading the given upstream models.""" + + model: DiscoveredSqlModelFile = model_file(name) + return CompileModelInput( + model_file=model, + query_sql=model.contents, + references=tuple( + CompileSqlReference(ref_kind=SqlReferenceKind.REF, ref_name=upstream_name) + for upstream_name in upstream + ), + ) + + +def stored_analysis( + *, name: str, context: str, dependencies: dict[str, str] +) -> StoredModelAnalysis: + """Return the analysis a previous compile stored for one model.""" + + return StoredModelAnalysis( + context=context, + cache_key=f"{name}-key", + analysis=PolyglotAnalysisResult(analysis_succeeded=True), + output_signature=f"{name}-signature", + dependencies=dependencies, + signature=f"{name}-signature", + ) + + +def analysis_request(model_input: CompileModelInput, cache_key: str) -> ModelSqlAnalysisRequest: + """Return the analysis request of one model with the given cache key.""" + + return ModelSqlAnalysisRequest( + model_input=model_input, + query_sql=model_input.query_sql, + placeholders=None, + cache_key=cache_key, + binding_schema=None, + ) + + +def orders_analysis_context( + *, requests: tuple[ModelSqlAnalysisRequest, ...], column_types: dict[str, dict[str, str]] +) -> str: + """Return the shared analysis inputs digest of the orders project with these source types.""" + + return analysis_reuse_context( + cache=ORDERS_CACHE, + model_names=frozenset(item.model_input.model_file.file_path.stem for item in requests), + column_types_by_table=column_types, + column_nullability_by_table={}, + complete_binding_schemas={}, + ) + + +def orders_stored_analyses( + *, requests: tuple[ModelSqlAnalysisRequest, ...], reuse: RecordingAnalysisReuse +) -> StoredModelAnalyses: + """Return the stored analyses a compile of the orders project may serve.""" + + return StoredModelAnalyses( + caching=ModelAnalysisCaching(cache=ORDERS_CACHE, reuse=reuse), + requests=requests, + column_types_by_table=RAW_ORDERS_TYPES, + column_nullability_by_table={}, + complete_binding_schemas={}, + ) + + +class RecordingAnalysisReuse: + """Stored analyses and reused renders of a previous compile, recording new analyses.""" + + def __init__(self, *, reused: frozenset[str], stored: dict[str, StoredModelAnalysis]) -> None: + self._reused: frozenset[str] = reused + self._stored: dict[str, StoredModelAnalysis] = stored + self.recorded: dict[str, StoredModelAnalysis | None] | None = None + + def reused_model_names(self) -> frozenset[str]: + """Return the names of models whose renders were reused.""" + + return self._reused + + def stored_analyses(self) -> dict[str, StoredModelAnalysis]: + """Return the stored analyses by model name.""" + + return self._stored + + def record_analyses(self, *, analyses: dict[str, StoredModelAnalysis | None]) -> None: + """Keep what the compile records.""" + + self.recorded = analyses diff --git a/tests/unit/src/sqlbuild/compiler/compile/classes/test_stored_model_analyses.py b/tests/unit/src/sqlbuild/compiler/compile/classes/test_stored_model_analyses.py new file mode 100644 index 0000000000..79d647a281 --- /dev/null +++ b/tests/unit/src/sqlbuild/compiler/compile/classes/test_stored_model_analyses.py @@ -0,0 +1,166 @@ +"""A stored analysis is served only while its inputs and its parents' final signatures match.""" + +from __future__ import annotations + +import pytest + +from sqlbuild.compiler.compile.classes.stored_model_analyses import StoredModelAnalyses +from sqlbuild.compiler.compile.models import ( + ModelSqlAnalysisRequest, + PolyglotAnalysisResult, + StoredModelAnalysis, +) +from tests.unit.src.sqlbuild.compiler.compile.classes._test_types import ( + StaleServedAnalysisTestCase, + StoredAnalysisScopeTestCase, +) +from tests.unit.src.sqlbuild.compiler.compile.classes.helpers import ( + RAW_ORDERS_TYPES, + RecordingAnalysisReuse, + analysis_request, + analyzed_model, + orders_analysis_context, + orders_stored_analyses, + stored_analysis, +) + +_CHAIN: tuple[ModelSqlAnalysisRequest, ...] = ( + analysis_request(analyzed_model("stg_orders"), "stg_orders-key"), + analysis_request(analyzed_model("int_orders", "stg_orders"), "int_orders-key"), + analysis_request(analyzed_model("fct_orders", "int_orders"), "fct_orders-key"), +) +_CHAIN_DEPENDENCIES: dict[str, dict[str, str]] = { + "stg_orders": {}, + "int_orders": {"stg_orders": "stg_orders-signature"}, + "fct_orders": {"int_orders": "int_orders-signature"}, +} +_STORED_SIGNATURES: dict[str, str] = { + "stg_orders": "stg_orders-signature", + "int_orders": "int_orders-signature", + "fct_orders": "fct_orders-signature", +} +_ALL_MODELS: frozenset[str] = frozenset(_CHAIN_DEPENDENCIES) + + +@pytest.mark.parametrize( + "test_case", + [ + StoredAnalysisScopeTestCase( + description="nothing_changed", + reused_models=_ALL_MODELS, + rekeyed_models=frozenset(), + twinned_models=frozenset(), + stored_column_types=RAW_ORDERS_TYPES, + expected_served=_ALL_MODELS, + ), + StoredAnalysisScopeTestCase( + description="root_edited", + reused_models=frozenset({"int_orders", "fct_orders"}), + rekeyed_models=frozenset(), + twinned_models=frozenset(), + stored_column_types=RAW_ORDERS_TYPES, + expected_served=frozenset({"int_orders", "fct_orders"}), + ), + StoredAnalysisScopeTestCase( + description="cache_key_changed", + reused_models=_ALL_MODELS, + rekeyed_models=frozenset({"int_orders"}), + twinned_models=frozenset(), + stored_column_types=RAW_ORDERS_TYPES, + expected_served=frozenset({"stg_orders", "fct_orders"}), + ), + StoredAnalysisScopeTestCase( + description="cache_key_shared_with_another_model", + reused_models=_ALL_MODELS, + rekeyed_models=frozenset(), + twinned_models=frozenset({"fct_orders"}), + stored_column_types=RAW_ORDERS_TYPES, + expected_served=frozenset({"stg_orders", "int_orders"}), + ), + StoredAnalysisScopeTestCase( + description="source_types_changed", + reused_models=_ALL_MODELS, + rekeyed_models=frozenset(), + twinned_models=frozenset(), + stored_column_types={"raw_orders": {"order_id": "BIGINT"}}, + expected_served=frozenset(), + ), + ], + ids=lambda case: case.description, +) +def test_given_stored_analyses_when_scoping_reuse_then_only_identical_inputs_are_served( + test_case: StoredAnalysisScopeTestCase, +) -> None: + context: str = orders_analysis_context( + requests=_CHAIN, column_types=test_case.stored_column_types + ) + stored: dict[str, StoredModelAnalysis] = { + name: stored_analysis(name=name, context=context, dependencies=dependencies) + for name, dependencies in _CHAIN_DEPENDENCIES.items() + } + requests: tuple[ModelSqlAnalysisRequest, ...] = tuple( + analysis_request( + request.model_input, + f"{request.model_input.model_file.file_path.stem}-key" + + "-edited" + * (request.model_input.model_file.file_path.stem in test_case.rekeyed_models), + ) + for request in _CHAIN + ) + tuple( + analysis_request(analyzed_model(f"{name}_twin", "int_orders"), f"{name}-key") + for name in sorted(test_case.twinned_models) + ) + + served: StoredModelAnalyses = orders_stored_analyses( + requests=requests, + reuse=RecordingAnalysisReuse(reused=test_case.reused_models, stored=stored), + ) + + assert frozenset(served.served) == test_case.expected_served + + +@pytest.mark.parametrize( + "test_case", + [ + StaleServedAnalysisTestCase( + description="parents_unchanged", + current_signatures=_STORED_SIGNATURES, + expected_stale=set(), + ), + StaleServedAnalysisTestCase( + description="root_signature_changed", + current_signatures={**_STORED_SIGNATURES, "stg_orders": "stg_orders-widened"}, + expected_stale={"int_orders"}, + ), + StaleServedAnalysisTestCase( + description="middle_signature_changed", + current_signatures={**_STORED_SIGNATURES, "int_orders": "int_orders-widened"}, + expected_stale={"fct_orders"}, + ), + ], + ids=lambda case: case.description, +) +def test_given_served_analyses_when_parents_are_final_then_changed_parents_make_children_stale( + test_case: StaleServedAnalysisTestCase, +) -> None: + context: str = orders_analysis_context(requests=_CHAIN, column_types=RAW_ORDERS_TYPES) + stored: dict[str, StoredModelAnalysis] = { + name: stored_analysis(name=name, context=context, dependencies=dependencies) + for name, dependencies in _CHAIN_DEPENDENCIES.items() + } + served: StoredModelAnalyses = orders_stored_analyses( + requests=_CHAIN, reuse=RecordingAnalysisReuse(reused=_ALL_MODELS, stored=stored) + ) + cached: dict[str, PolyglotAnalysisResult] = { + analysis.cache_key: analysis.analysis for analysis in stored.values() + } + + stale: set[str] = served.stale_names( + requests=_CHAIN, cached=cached, current_signatures=test_case.current_signatures + ) + + assert stale == test_case.expected_stale + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-n", "auto", "--dist", "loadfile"])) From d4ea211c5239122f5c40687c42a98f235a63451f Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Tue, 6 Oct 2026 05:25:53 +0100 Subject: [PATCH 3/3] perf(compile): skip reading artifacts a stored compile left unchanged Store each compiled artifact's digest beside its stat stamp in the compile reuse entry. When a later full compile writes an artifact whose stat identity still matches the stored stamp and whose new contents have the stored digest, it neither reads nor rewrites the file. Any stat or digest difference falls back to comparing the file's bytes. --- .../sqlbuild/references/docs/cli/compile.md | 13 +- .../cli/commands/_helpers/compile/pipeline.py | 2 +- .../_helpers/compile/target_writer.py | 6 +- .../cli/commands/main/project/_compile.py | 7 +- .../cli/compile_reuse/_helpers/attempt.py | 2 + .../cli/compile_reuse/_helpers/entry_file.py | 5 + .../cli/compile_reuse/_helpers/store.py | 5 + .../compile_reuse/_helpers/target_files.py | 10 +- .../compile_artifact_write_registry.py | 26 +- .../compile_reuse/classes/stored_artifacts.py | 29 ++ src/sqlbuild/cli/compile_reuse/models.py | 3 + .../cli/commands/main/compile/_test_types.py | 24 ++ .../cli/commands/main/compile/helpers.py | 345 +++++++++++++++++- .../test_compile_incremental_edit_phases.py | 305 ++++++++++++++++ .../compile/test_compile_incremental_edits.py | 5 + .../cli/compile_reuse/classes/__init__.py | 0 .../cli/compile_reuse/classes/_test_types.py | 17 + .../cli/compile_reuse/classes/helpers.py | 32 ++ .../classes/test_stored_artifacts.py | 73 ++++ website/src/content/docs/docs/cli/compile.mdx | 13 +- 20 files changed, 883 insertions(+), 39 deletions(-) create mode 100644 src/sqlbuild/cli/compile_reuse/classes/stored_artifacts.py create mode 100644 tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_incremental_edit_phases.py create mode 100644 tests/unit/src/sqlbuild/cli/compile_reuse/classes/__init__.py create mode 100644 tests/unit/src/sqlbuild/cli/compile_reuse/classes/_test_types.py create mode 100644 tests/unit/src/sqlbuild/cli/compile_reuse/classes/helpers.py create mode 100644 tests/unit/src/sqlbuild/cli/compile_reuse/classes/test_stored_artifacts.py diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/compile.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/compile.md index bc37009e7e..5761be4184 100644 --- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/compile.md +++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/compile.md @@ -125,9 +125,14 @@ line to stderr before the report. When the only changes since the stored compile are edits to existing model files, compile runs in full but reuses the stored render of every unchanged model, together with the diagnostics and -environment reads that rendering reported. Only the edited models are rendered again. Analysis, -contracts, rules, and artifact writes still run for the whole project, so downstream effects of an -edit, such as a changed column type, are reported exactly as in an uncached compile. The output is +environment reads that rendering reported. Only the edited models are rendered again. Declaration +files other than models are not read again, and an unchanged model reuses its stored column +analysis only when the analyses of its upstream models in this compile match the ones it was stored +with. Contracts, semantic validation, and rules still run for the whole project, and a model whose +upstream analyses changed is analyzed again, so downstream effects of an edit, such as a changed +column type, are reported exactly as in an uncached compile, even when another command such as +`sqb build` ran in between. Compiled +files that still hold the bytes the stored compile wrote are not read or rewritten. The output is byte-identical to `sqb compile --no-cache`. Every other change renders the whole project: adding, removing, or renaming a model, and editing @@ -162,7 +167,7 @@ A replayed JSON report is byte-identical to the previous one except for `compile holds `project_reuse_hits`, `project_reuse_misses`, `project_reuse_bypasses`, `project_reuse_check_ms`, and `total_ms`. Full compiles report the same reuse counters alongside the phase timings, plus `render_reuse_hits` and `render_reuse_misses` for model renders reused or -rendered again after model-only edits. In text mode, a replayed compile does not print the per-phase progress lines. +rendered again after model-only edits and `analysis_reuse_hits` for stored model analyses reused. In text mode, a replayed compile does not print the per-phase progress lines. Lifecycle sinks receive only the invocation events for a replayed compile; per-phase `operation` events are not emitted because no compile phase runs. diff --git a/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py b/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py index b17b387940..18aca51c83 100644 --- a/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py +++ b/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py @@ -329,7 +329,7 @@ def write_compile_dag_artifact( python_graph=python_graph, ) _ = resolved_dag_path.write_text(dag_json, encoding="utf-8") - COMPILE_ARTIFACT_WRITES.written( + _ = COMPILE_ARTIFACT_WRITES.written( path=resolved_dag_path, contents=dag_json.replace("\n", os.linesep).encode("utf-8") ) _ = complete_compile_phase( diff --git a/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py b/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py index 6be15c5435..7c2a99bf20 100644 --- a/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py +++ b/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py @@ -672,7 +672,7 @@ def _write_sql(*, path: Path, sql: str, check_existing: bool = True) -> None: def _write_text_if_changed(*, path: Path, contents: str, check_existing: bool = True) -> None: with record_compile_timing("physical_write_ms"): - COMPILE_ARTIFACT_WRITES.written( + _ = COMPILE_ARTIFACT_WRITES.written( path=path, contents=contents.replace(_POSIX_LINE_SEPARATOR, os.linesep).encode("utf-8"), ) @@ -687,7 +687,9 @@ def _write_text_if_changed(*, path: Path, contents: str, check_existing: bool = def _write_bytes_if_changed(*, path: Path, contents: bytes, check_existing: bool = True) -> None: with record_compile_timing("physical_write_ms"): - COMPILE_ARTIFACT_WRITES.written(path=path, contents=contents) + unchanged: bool = COMPILE_ARTIFACT_WRITES.written(path=path, contents=contents) + if check_existing and unchanged: + return if check_existing: existing: bytes | None = _read_existing_file(path=path) if existing == contents: diff --git a/src/sqlbuild/cli/commands/main/project/_compile.py b/src/sqlbuild/cli/commands/main/project/_compile.py index 466362096a..3902e30069 100644 --- a/src/sqlbuild/cli/commands/main/project/_compile.py +++ b/src/sqlbuild/cli/commands/main/project/_compile.py @@ -32,6 +32,7 @@ from sqlbuild.cli.compile_render_reuse.main._render_reuse_session import render_reuse_session from sqlbuild.cli.compile_render_reuse.models import CompileRenderReuse from sqlbuild.cli.compile_reuse.classes.recorded_compile_output import RecordedCompileOutput +from sqlbuild.cli.compile_reuse.classes.stored_artifacts import StoredArtifacts from sqlbuild.cli.compile_reuse.constants import COMPILE_ARTIFACT_WRITES from sqlbuild.cli.compile_reuse.main._compile_reuse_timings import compile_reuse_timings from sqlbuild.cli.compile_reuse.main._write_reusable_compile import write_reusable_compile @@ -77,7 +78,11 @@ def run_compile( with ( collect_compile_timings() as detailed_timings, COMPILE_INPUT_READS.recording() as input_reads, - COMPILE_ARTIFACT_WRITES.recording() as artifact_writes, + COMPILE_ARTIFACT_WRITES.recording( + stored=StoredArtifacts( + stamps=attempt.prior_target_files, digests=attempt.prior_target_digests + ) + ) as artifact_writes, activated_render_reuse(render_reuse=render_reuse), PreparedCompileArtifacts( enabled=not request.profile_flags.skip_write diff --git a/src/sqlbuild/cli/compile_reuse/_helpers/attempt.py b/src/sqlbuild/cli/compile_reuse/_helpers/attempt.py index e963bbc831..d7a0e729b3 100644 --- a/src/sqlbuild/cli/compile_reuse/_helpers/attempt.py +++ b/src/sqlbuild/cli/compile_reuse/_helpers/attempt.py @@ -106,6 +106,8 @@ def attempt_reuse(*, request: CompileReuseRequest) -> CompileReuseAttempt: attempt, prior_entry=True, prior_modules=header.inputs.modules, + prior_target_files=header.inputs.target_files, + prior_target_digests=header.inputs.target_digests, restamped=restamped_paths(stored=header.inputs.project_files, current=attempt.snapshot), render_state_path=( None diff --git a/src/sqlbuild/cli/compile_reuse/_helpers/entry_file.py b/src/sqlbuild/cli/compile_reuse/_helpers/entry_file.py index c7453b448c..aceecdc384 100644 --- a/src/sqlbuild/cli/compile_reuse/_helpers/entry_file.py +++ b/src/sqlbuild/cli/compile_reuse/_helpers/entry_file.py @@ -254,6 +254,7 @@ def _payload(*, inputs: StoredCompileInputs, output: StoredCompileOutput) -> dic for path, item in inputs.project_files.items() }, "target_files": {path: list(stamp) for path, stamp in inputs.target_files.items()}, + "target_digests": inputs.target_digests, "stderr_lines": list(output.stderr_lines), "exit_code": output.exit_code, "timings_span": None if output.timings_span is None else list(output.timings_span), @@ -309,6 +310,10 @@ def _header(*, payload: object) -> StoredCompileHeader: _settings_inputs(item) for item in _list(fields["settings_inputs"]) ), settings_digest=_string(fields["settings_digest"]), + target_digests={ + _string(path): _string(digest) + for path, digest in _mapping(fields.get("target_digests", {})).items() + }, ), output=StoredCompileOutput( stderr_lines=tuple(_string(line) for line in _list(fields["stderr_lines"])), diff --git a/src/sqlbuild/cli/compile_reuse/_helpers/store.py b/src/sqlbuild/cli/compile_reuse/_helpers/store.py index 7ead497bbc..853013c52f 100644 --- a/src/sqlbuild/cli/compile_reuse/_helpers/store.py +++ b/src/sqlbuild/cli/compile_reuse/_helpers/store.py @@ -164,6 +164,11 @@ def _write_compile_entry( target_tree=artifacts_written, settings_inputs=settings.inputs, settings_digest=settings_inputs_digest(inputs=settings.inputs), + target_digests={ + path: artifact.digest + for path, artifact in artifact_writes.artifacts.items() + if artifact.digest is not None and path in target_files + }, ), output=StoredCompileOutput( stderr_lines=output.stderr_lines, diff --git a/src/sqlbuild/cli/compile_reuse/_helpers/target_files.py b/src/sqlbuild/cli/compile_reuse/_helpers/target_files.py index 72d1fd4ecf..4cbddde38a 100644 --- a/src/sqlbuild/cli/compile_reuse/_helpers/target_files.py +++ b/src/sqlbuild/cli/compile_reuse/_helpers/target_files.py @@ -32,11 +32,11 @@ def snapshot_target_files( if entry.is_dir(follow_symlinks=False): pending.append(entry.path) continue - stamp: FileStamp | None = _path_stamp(path=entry.path) + stamp: FileStamp | None = path_stamp(path=entry.path) if stamp is not None: stamps[entry.path] = stamp for path in extra_paths: - extra: FileStamp | None = _path_stamp(path=path) + extra: FileStamp | None = path_stamp(path=path) if extra is not None: stamps[path] = extra return stamps @@ -81,7 +81,7 @@ def _target_stamps(*, project_dir: str, paths: tuple[str, ...], tree: bool) -> d """Stamp the whole compiled tree plus extra paths, or only the given paths.""" if not tree: - return {path: stamp for path in paths if (stamp := _path_stamp(path=path)) is not None} + return {path: stamp for path in paths if (stamp := path_stamp(path=path)) is not None} compiled_prefix: str = _compiled_root(project_dir=project_dir) + os.sep extra_paths: tuple[str, ...] = tuple( path for path in paths if not path.startswith(compiled_prefix) @@ -103,7 +103,9 @@ def _compiled_root(*, project_dir: str) -> str: return os.path.join(project_dir, TARGET_DIRECTORY_NAME, COMPILED_DIRECTORY_NAME) -def _path_stamp(*, path: str) -> FileStamp | None: +def path_stamp(*, path: str) -> FileStamp | None: + """Return the stat identity of one path, or None when it cannot be stated.""" + try: status: os.stat_result = os.lstat(path) except OSError: diff --git a/src/sqlbuild/cli/compile_reuse/classes/compile_artifact_write_registry.py b/src/sqlbuild/cli/compile_reuse/classes/compile_artifact_write_registry.py index 069c0f24a9..cba8640bd6 100644 --- a/src/sqlbuild/cli/compile_reuse/classes/compile_artifact_write_registry.py +++ b/src/sqlbuild/cli/compile_reuse/classes/compile_artifact_write_registry.py @@ -7,10 +7,14 @@ import threading from collections.abc import Iterator from contextlib import contextmanager +from typing import TYPE_CHECKING from sqlbuild.cli.compile_reuse.classes.compile_artifact_writes import CompileArtifactWrites from sqlbuild.cli.compile_reuse.models import RecordedArtifact +if TYPE_CHECKING: + from sqlbuild.cli.compile_reuse.classes.stored_artifacts import StoredArtifacts + class CompileArtifactWriteRegistry: """Deliver artifact writes from every thread, including background staging threads.""" @@ -19,31 +23,41 @@ def __init__(self, *, digest_size: int) -> None: self._digest_size: int = digest_size self._lock: threading.Lock = threading.Lock() self._recorders: tuple[CompileArtifactWrites, ...] = () + self._stored: StoredArtifacts | None = None @contextmanager - def recording(self) -> Iterator[CompileArtifactWrites]: - """Record artifact writes until the block exits.""" + def recording( + self, *, stored: StoredArtifacts | None = None + ) -> Iterator[CompileArtifactWrites]: + """Record artifact writes until the block exits, trusting unchanged stored artifacts.""" recorder: CompileArtifactWrites = CompileArtifactWrites() with self._lock: self._recorders = (*self._recorders, recorder) + if stored is not None: + self._stored = stored try: yield recorder finally: with self._lock: self._recorders = tuple(item for item in self._recorders if item is not recorder) + if stored is not None: + self._stored = None - def written(self, *, path: str | os.PathLike[str], contents: bytes) -> None: - """Note the exact bytes an artifact holds after a write or an unchanged-file check.""" + def written(self, *, path: str | os.PathLike[str], contents: bytes) -> bool: + """Note the exact bytes an artifact holds; return whether it already holds them.""" recorders: tuple[CompileArtifactWrites, ...] = self._recorders if not recorders: - return + return False + absolute_path: str = os.path.abspath(path) artifact: RecordedArtifact = RecordedArtifact( digest=hashlib.blake2b(contents, digest_size=self._digest_size).hexdigest() ) for recorder in recorders: - recorder.record(path=os.path.abspath(path), artifact=artifact) + recorder.record(path=absolute_path, artifact=artifact) + stored: StoredArtifacts | None = self._stored + return stored is not None and stored.holds(path=absolute_path, digest=artifact.digest) def kept(self, *, path: str | os.PathLike[str], size: int, mtime_ns: int) -> None: """Note an earlier artifact kept as-is after its size and mtime were validated.""" diff --git a/src/sqlbuild/cli/compile_reuse/classes/stored_artifacts.py b/src/sqlbuild/cli/compile_reuse/classes/stored_artifacts.py new file mode 100644 index 0000000000..0fc105cfbb --- /dev/null +++ b/src/sqlbuild/cli/compile_reuse/classes/stored_artifacts.py @@ -0,0 +1,29 @@ +"""Artifacts a stored compile left in target/, trusted while their stat identity is unchanged.""" + +from __future__ import annotations + +from sqlbuild.cli.compile_reuse._helpers.target_files import path_stamp +from sqlbuild.cli.compile_reuse.types import FileStamp + + +class StoredArtifacts: + """Answer whether an artifact still holds the bytes a stored compile recorded for it.""" + + def __init__(self, *, stamps: dict[str, FileStamp], digests: dict[str, str]) -> None: + self._stamps: dict[str, FileStamp] = stamps + self._digests: dict[str, str] = digests + + def holds(self, *, path: str, digest: str | None) -> bool: + """Return whether the file at path is unchanged since it was stored with this digest.""" + + return ( + digest is not None + and self._digests.get(path) == digest + and self.unchanged_since_stored(path=path) + ) + + def unchanged_since_stored(self, *, path: str) -> bool: + """Return whether the file at path still has the stat identity it was stored with.""" + + stamp: FileStamp | None = self._stamps.get(path) + return stamp is not None and path_stamp(path=path) == stamp diff --git a/src/sqlbuild/cli/compile_reuse/models.py b/src/sqlbuild/cli/compile_reuse/models.py index 12d6acfc49..16f143dfb5 100644 --- a/src/sqlbuild/cli/compile_reuse/models.py +++ b/src/sqlbuild/cli/compile_reuse/models.py @@ -81,6 +81,7 @@ class StoredCompileInputs: target_tree: bool settings_inputs: tuple[SettingsEnvironmentInputs, ...] settings_digest: str + target_digests: dict[str, str] = field(default_factory=dict) @dataclass(frozen=True) @@ -133,6 +134,8 @@ class CompileReuseAttempt: render_state_path: Path | None = None prior_entry: bool = False prior_modules: tuple[tuple[str, int, int], ...] = () + prior_target_files: dict[str, FileStamp] = field(default_factory=dict) + prior_target_digests: dict[str, str] = field(default_factory=dict) @dataclass(frozen=True) 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 b399e498dd..df9edf5609 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 @@ -6,6 +6,8 @@ from dataclasses import dataclass, field from pathlib import Path +import pytest + from tests.integration.src.sqlbuild.compiler.pipeline._test_types import PreparedCompile @@ -692,6 +694,7 @@ class IncrementalEditStep: description: str edit: Callable[[Path], None] expected_render_reuse: bool + between: Callable[[Path], None] @dataclass(frozen=True) @@ -714,6 +717,18 @@ class RandomEditChainTestCase: expected_matches_uncached: bool = True +@dataclass(frozen=True) +class BrokenEditInvalidationTestCase: + """An incremental invalidation deliberately broken, and edits whose output it must change.""" + + description: str + compiled_edits: tuple[Callable[[Path], None], ...] + edit: Callable[[Path], None] + sabotage: Callable[[pytest.MonkeyPatch], None] + expected_matches_uncached: bool + after_edit: Callable[[Path, pytest.MonkeyPatch], None] + + @dataclass(frozen=True) class BrokenChangeDetectionTestCase: """A model edit compiled while change detection reports nothing changed.""" @@ -771,3 +786,12 @@ class DenseBatchedPreparationTestCase: type DensePreparedCompile = PreparedCompile + + +@dataclass(frozen=True) +class SharedCacheKeyTestCase: + """A model analysis schedule under which a model shares a served model's cache key.""" + + description: str + schedule: Callable[[pytest.MonkeyPatch], None] + expected_matches_uncached: bool = True 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 a5709dcc3f..32277c374e 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py @@ -27,9 +27,15 @@ import sqlbuild.cli.commands.main.project._compile as compile_command_module import sqlbuild.cli.compile_render_reuse._helpers.load_notice as render_load_notice +import sqlbuild.cli.compile_render_reuse.main._render_reuse_session as render_reuse_main import sqlbuild.cli.compile_reuse._helpers.attempt as reuse_attempt import sqlbuild.cli.compile_reuse._helpers.project_files as reuse_project_files import sqlbuild.cli.compile_reuse._helpers.store as reuse_store +import sqlbuild.compiler.compile._helpers.assembly.binding_waves as binding_waves +import sqlbuild.compiler.compile._helpers.assembly.project as project_assembly +import sqlbuild.compiler.compile.classes.binding_dataflow as binding_dataflow +import sqlbuild.compiler.compile.classes.render_reuse_session as render_reuse_session +import sqlbuild.compiler.compile.classes.stored_model_analyses as stored_model_analyses from scripts.cold_compile_performance.main.read_compile_measurement import read_compile_measurement from scripts.cold_compile_performance.main.semantic_compile_fingerprint import ( semantic_compile_fingerprint, @@ -40,6 +46,7 @@ read_entry_stdout, write_entry, ) +from sqlbuild.cli.compile_reuse.classes.stored_artifacts import StoredArtifacts from sqlbuild.cli.compile_reuse.constants import ( REUSE_DISABLE_ENV_VAR, REUSE_ENTRY_DIRECTORY_PARTS, @@ -3332,6 +3339,18 @@ def compile_in_process(*, project_dir: Path) -> int: return main(["--project-dir", str(project_dir), "--no-color", "compile", "--json"]) +def compile_edit_without_reuse(project_dir: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Compile an edit with compile reuse disabled, so only the analysis cache sees it.""" + + monkeypatch.setenv(REUSE_DISABLE_ENV_VAR, "1") + _ = compile_in_process(project_dir=project_dir) + monkeypatch.setenv(REUSE_DISABLE_ENV_VAR, "0") + + +def leave_edit_uncompiled(_project_dir: Path, _monkeypatch: pytest.MonkeyPatch) -> None: + """Leave the edit for the compared compile to see first.""" + + def write_before_reuse_store( *, monkeypatch: pytest.MonkeyPatch, project_dir: Path, write: Callable[[Path], None] ) -> None: @@ -3434,12 +3453,12 @@ def raise_error(**_kwargs: Any) -> None: def compile_in_process_output( - *, project_dir: Path, capsys: pytest.CaptureFixture[str] + *, project_dir: Path, capsys: pytest.CaptureFixture[str], args: tuple[str, ...] = () ) -> tuple[int, str, str]: """Run one JSON compile in this process and return its exit code, stdout, and stderr.""" _ = capsys.readouterr() - code: int = main(["--project-dir", str(project_dir), "--no-color", "compile", "--json"]) + code: int = main(["--project-dir", str(project_dir), "--no-color", "compile", "--json", *args]) out, err = capsys.readouterr() return code, out, err @@ -3489,11 +3508,11 @@ def compare_incremental_compile(*, project_dir: Path) -> IncrementalEditComparis def in_process_reuse_run( - *, project_dir: Path, capsys: pytest.CaptureFixture[str] + *, project_dir: Path, capsys: pytest.CaptureFixture[str], args: tuple[str, ...] = () ) -> CompileReuseRun: """Compile with --json in this process and capture output comparable with fresh processes.""" - code, out, err = compile_in_process_output(project_dir=project_dir, capsys=capsys) + code, out, err = compile_in_process_output(project_dir=project_dir, capsys=capsys, args=args) payload: dict[str, object] = cast(dict[str, object], json.loads(out)) return CompileReuseRun( returncode=code, @@ -3505,6 +3524,8 @@ def in_process_reuse_run( GENERATED_EDIT_MODEL_PREFIX: str = "orders_step_" +_STAR_MODEL_SHARE: float = 0.4 +_TWIN_MODEL_SHARE: float = 0.25 _GENERATED_EDIT_COLUMNS: tuple[str, ...] = ("order_id", "customer_id", "quantity", "status") _PLAIN_QUANTITY: str = " quantity,\n" _CAST_QUANTITY: str = " CAST(quantity AS BIGINT) AS quantity,\n" @@ -3516,22 +3537,37 @@ def in_process_reuse_run( "type_change", "header_change", "introduce_error", + "drop_column", "add_model", "macro_edit", "test_edit", ) _RANDOM_EDIT_FOLLOW_UPS: dict[str, tuple[str, ...]] = { "introduce_error": ("introduce_error", "fix_error"), + "drop_column": ("drop_column", "comment", "restore_column"), } _MODEL_ONLY_EDIT_KINDS: frozenset[str] = frozenset( - {"comment", "add_column", "type_change", "header_change", "introduce_error", "fix_error"} + { + "comment", + "add_column", + "type_change", + "header_change", + "introduce_error", + "fix_error", + "drop_column", + "restore_column", + } ) +_DROPPED_COLUMN: str = " customer_id,\n" +_KEPT_COLUMN: str = " order_id,\n" +_EDIT_ANCHORS: dict[str, str] = {"type_change": " quantity", "drop_column": _DROPPED_COLUMN} def write_generated_edit_models(*, project_dir: Path, model_count: int, seed: int) -> None: - """Add a seeded chain of models over the staging orders, each reading an earlier one.""" + """Add seeded models over the staging orders reading earlier ones, some with * or twins.""" chooser: random.Random = random.Random(seed) + star_chooser: random.Random = random.Random(seed + 2) upstreams: tuple[str, ...] = ( "stg_orders", *( @@ -3539,12 +3575,27 @@ def write_generated_edit_models(*, project_dir: Path, model_count: int, seed: in for index in range(1, model_count) ), ) + twin_chooser: random.Random = random.Random(seed + 3) for index, upstream in enumerate(upstreams): - write_project_file( - project_dir, - f"models/generated/{GENERATED_EDIT_MODEL_PREFIX}{index:03d}.sql", - _generated_edit_model_sql(upstream=upstream), - ) + name: str = f"models/generated/{GENERATED_EDIT_MODEL_PREFIX}{index:03d}" + contents: str = (_generated_edit_model_sql, _generated_star_model_sql)[ + star_chooser.random() < _STAR_MODEL_SHARE + ](upstream=upstream) + write_project_file(project_dir, f"{name}.sql", contents) + twins: tuple[str, ...] = (contents,) * (twin_chooser.random() < _TWIN_MODEL_SHARE) + for twin in twins: + write_project_file( + project_dir, + f"{name}_twin.sql", + twin.replace("MODEL (description '", "MODEL (description 'Twin. ", 1), + ) + + +def _generated_star_model_sql(*, upstream: str) -> str: + return ( + "MODEL (description 'Generated order pass-through.', materialized view);\n\n" + f'SELECT\n *\nFROM __ref("{upstream}")\n' + ) def _generated_edit_model_sql(*, upstream: str) -> str: @@ -3590,12 +3641,29 @@ def __init__(self, *, project_dir: Path, seed: int) -> None: self._project_dir: Path = project_dir self._random: random.Random = random.Random(seed) self._broken: Path = project_dir / "models" / "generated" / "unbroken.sql" + self._dropped: Path = project_dir / "models" / "generated" / "undropped.sql" self._added: int = 0 + self._interventions: random.Random = random.Random(seed + 1) + + def intervene(self) -> None: + """Sometimes run another command over the edited project before it is compiled.""" + + commands: tuple[Callable[[Path], None], ...] = ( + *INTERVENING_COMMANDS, + *(no_intervening_command,) * 3, + ) + self._interventions.choice(commands)(self._project_dir) def apply(self, kind: str) -> None: """Apply one edit of the given kind to a randomly chosen generated model.""" - models: list[Path] = sorted((self._project_dir / "models" / "generated").glob("*.sql")) + anchor: str = _EDIT_ANCHORS.get(kind, "") + models: list[Path] = list( + filter( + lambda model: anchor in model.read_text(encoding="utf-8"), + sorted((self._project_dir / "models" / "generated").glob("*.sql")), + ) + ) edit: Callable[[Path], None] = getattr(self, f"_{kind}") edit(self._random.choice(models)) @@ -3603,8 +3671,8 @@ def _rewrite(self, model: Path, transform: Callable[[str], str]) -> None: model.write_text(transform(model.read_text(encoding="utf-8")), encoding="utf-8") def _comment(self, model: Path) -> None: - note: str = f"-- note {self._random.random()}\nFROM __ref" - self._rewrite(model, lambda contents: contents.replace("FROM __ref", note)) + note: str = f"-- note {self._random.random()}\nSELECT\n" + self._rewrite(model, lambda contents: contents.replace("SELECT\n", note, 1)) def _add_column(self, model: Path) -> None: column: str = f",\n status AS status_{self._random.randrange(1000)}\nFROM __ref" @@ -3628,6 +3696,16 @@ def _introduce_error(self, model: Path) -> None: def _fix_error(self, _model: Path) -> None: self._rewrite(self._broken, lambda contents: contents.replace(" unknown_column,\n", "")) + def _drop_column(self, model: Path) -> None: + self._rewrite(model, lambda contents: contents.replace(_DROPPED_COLUMN, "", 1)) + self._dropped = model + + def _restore_column(self, _model: Path) -> None: + self._rewrite( + self._dropped, + lambda contents: contents.replace(_KEPT_COLUMN, _KEPT_COLUMN + _DROPPED_COLUMN, 1), + ) + def _add_model(self, model: Path) -> None: self._added += 1 write_project_file( @@ -3652,19 +3730,144 @@ def _test_edit(self, _model: Path) -> None: STG_ORDERS_MODEL: str = "models/staging/stg_orders.sql" +STG_PAYMENTS_MODEL: str = "models/staging/stg_payments.sql" FACT_ORDERS_MODEL: str = "models/marts/fact_orders.sql" +DIM_CUSTOMERS_ARTIFACT: str = "target/compiled/models/marts/dim_customers.sql" +_FACT_QUANTITY_TYPE: str = " quantity (type INTEGER),\n" +_FACT_ORDER_ID_COLUMN: str = " order_id (nullable false, audits [not_null]),\n" + +def no_intervening_command(_root: Path) -> None: + """Compile the edit directly after making it.""" -def model_edit_step(description: str, edit: Callable[[Path], None]) -> IncrementalEditStep: + +def model_edit_step( + description: str, + edit: Callable[[Path], None], + between: Callable[[Path], None] = no_intervening_command, +) -> IncrementalEditStep: """Return an edit to model files only, whose compile must reuse unaffected renders.""" - return IncrementalEditStep(description=description, edit=edit, expected_render_reuse=True) + return IncrementalEditStep( + description=description, edit=edit, expected_render_reuse=True, between=between + ) def full_edit_step(description: str, edit: Callable[[Path], None]) -> IncrementalEditStep: """Return an edit beyond model contents, whose compile must render everything again.""" - return IncrementalEditStep(description=description, edit=edit, expected_render_reuse=False) + return IncrementalEditStep( + description=description, + edit=edit, + expected_render_reuse=False, + between=no_intervening_command, + ) + + +def plan_between(root: Path) -> None: + """Plan the edited project before the compared compile.""" + + _ = run_installed_sqb(project_dir=root, args=("plan",), env=COMPILE_REUSE_ENV) + + +def build_between(root: Path) -> None: + """Build the edited project before the compared compile.""" + + _ = run_installed_sqb(project_dir=root, args=("build",), env=COMPILE_REUSE_ENV) + + +def other_target_build_between(root: Path) -> None: + """Build the edited project for another target before the compared default-target compile.""" + + _ = run_installed_sqb( + project_dir=root, args=("build", "--target", "prod"), env=COMPILE_REUSE_ENV + ) + + +def compile_without_reuse_between(root: Path) -> None: + """Compile the edited project with compile reuse disabled before the compared compile.""" + + _ = run_installed_sqb( + project_dir=root, + args=("compile", "--json"), + env={**COMPILE_REUSE_ENV, REUSE_DISABLE_ENV_VAR: "1"}, + ) + + +INTERVENING_COMMANDS: tuple[Callable[[Path], None], ...] = ( + plan_between, + build_between, + other_target_build_between, + compile_without_reuse_between, +) +_STAR_CHAIN_MODELS: dict[str, str] = { + "models/marts/chain_mid.sql": ( + "MODEL (description 'Every staged order column.', materialized view);\n\n" + 'SELECT * FROM __ref("stg_orders")\n' + ), + "models/marts/chain_leaf.sql": ( + "MODEL (description 'Every chained order column.', materialized view);\n\n" + 'SELECT * FROM __ref("chain_mid")\n' + ), +} +_STAGING_LAST_COLUMN: str = " status\nFROM __source" + + +_TWIN_MODEL: str = "models/marts/chain_a_twin.sql" + + +def star_chain_with_twin_added(root: Path) -> None: + """Add the star chain and a twin of its leaf with the same SQL, so both share a cache key.""" + + star_chain_added(root) + write_project_file(root, _TWIN_MODEL, _STAR_CHAIN_MODELS["models/marts/chain_leaf.sql"]) + + +def twin_header_changed(root: Path) -> None: + """Change the twin's header only, so its render is not reused while its SQL is unchanged.""" + + replace_project_text( + root, + _TWIN_MODEL, + "'Every chained order column.'", + "'Every chained order column, twice.'", + ) + + +def analyze_one_model_at_a_time(monkeypatch: pytest.MonkeyPatch) -> None: + """Run the binding dataflow with one worker taking one model at a time.""" + + for name in ("_DATAFLOW_WORKERS", "_DATAFLOW_BATCH_MIN", "_DATAFLOW_BATCH_LIMIT"): + monkeypatch.setattr(binding_waves, name, 1) + + +def analyze_in_one_batch(monkeypatch: pytest.MonkeyPatch) -> None: + """Analyze every model in one batch, without dependency-ordered binding.""" + + monkeypatch.setattr(project_assembly, "binding_schema_for_model", lambda **_kwargs: None) + + +def star_chain_added(root: Path) -> None: + """Add two models that each select every column of the one before, below staging orders.""" + + for relative_path, contents in _STAR_CHAIN_MODELS.items(): + write_project_file(root, relative_path, contents) + + +def staging_extra_flag(root: Path) -> None: + """Add an output column to the staging orders, which every star model passes on.""" + + replace_project_text( + root, STG_ORDERS_MODEL, _STAGING_LAST_COLUMN, " status,\n 1 AS extra_flag\nFROM __source" + ) + + +def staging_extra_flag_removed(root: Path) -> None: + """Remove the added staging orders output column again.""" + + replace_project_text( + root, STG_ORDERS_MODEL, " status,\n 1 AS extra_flag\nFROM __source", _STAGING_LAST_COLUMN + ) def staging_comment(root: Path) -> None: @@ -3740,6 +3943,114 @@ def fact_comment(root: Path) -> None: replace_project_text(root, FACT_ORDERS_MODEL, "FROM __ref", "-- leaf\nFROM __ref") +def staging_column_removed(root: Path) -> None: + """Drop the quantity column from staging orders, which downstream models and tests read.""" + + replace_project_text(root, STG_ORDERS_MODEL, " quantity,\n", "") + + +def staging_column_restored(root: Path) -> None: + """Restore the staging orders quantity column.""" + + replace_project_text( + root, STG_ORDERS_MODEL, " waffle_type_id,\n", " waffle_type_id,\n quantity,\n" + ) + + +def staging_column_renamed(root: Path) -> None: + """Rename the staging orders status column that audits, tests, and models read.""" + + replace_project_text(root, STG_ORDERS_MODEL, " status\nFROM", " status AS order_state\nFROM") + + +def staging_rename_reverted(root: Path) -> None: + """Restore the staging orders status column name.""" + + replace_project_text(root, STG_ORDERS_MODEL, " status AS order_state\nFROM", " status\nFROM") + + +def staging_quantity_text(root: Path) -> None: + """Turn the staging orders quantity into text, which violates a downstream declared type.""" + + replace_project_text( + root, STG_ORDERS_MODEL, " quantity,\n", " CAST(quantity AS VARCHAR) AS quantity,\n" + ) + + +def staging_quantity_restored(root: Path) -> None: + """Restore the staging orders quantity type.""" + + replace_project_text( + root, STG_ORDERS_MODEL, " CAST(quantity AS VARCHAR) AS quantity,\n", " quantity,\n" + ) + + +def fact_quantity_type_declared(root: Path) -> None: + """Declare the fact orders quantity type, which enforces it.""" + + replace_project_text( + root, FACT_ORDERS_MODEL, _FACT_ORDER_ID_COLUMN, _FACT_ORDER_ID_COLUMN + _FACT_QUANTITY_TYPE + ) + + +def fact_quantity_type_removed(root: Path) -> None: + """Remove the declared fact orders quantity type.""" + + replace_project_text(root, FACT_ORDERS_MODEL, _FACT_QUANTITY_TYPE, "") + + +def payments_comment(root: Path) -> None: + """Add a comment to the staging payments model, unrelated to staging orders.""" + + anchor: str = 'FROM __source("raw__payments")' + replace_project_text(root, STG_PAYMENTS_MODEL, anchor, f"-- paid\n{anchor}") + + +def compiled_artifact_tampered(root: Path) -> None: + """Overwrite an unchanged model's compiled artifact, then edit another model.""" + + artifact: Path = root / DIM_CUSTOMERS_ARTIFACT + artifact.write_bytes(artifact.read_bytes() + b"-- tampered\n") + staging_comment(root) + + +def stg_orders_test_edit(root: Path) -> None: + """Change one expected value in the staging orders SQL test.""" + + replace_project_text( + root, "tests/unit/test_stg_orders.sql", "100 AS customer_id", "101 AS customer_id" + ) + + +def keep_invalidation(_monkeypatch: pytest.MonkeyPatch) -> None: + """Leave every incremental invalidation intact.""" + + +def skip_upstream_analysis_checks(monkeypatch: pytest.MonkeyPatch) -> None: + """Serve stored analyses even when an upstream model's final signature changed.""" + + for module in (stored_model_analyses, binding_dataflow): + monkeypatch.setattr(module, "dependencies_current", lambda **_kwargs: True) + + +def reuse_declarations_after_any_edit(monkeypatch: pytest.MonkeyPatch) -> None: + """Load stored renders and reuse stored declaration files even after a non-model edit.""" + + read_render_state: Callable[..., object] = render_reuse_main.read_render_state + monkeypatch.setattr( + render_reuse_main, + "read_render_state", + lambda *, path, changed_paths: read_render_state(path=path, changed_paths=frozenset()), + ) + monkeypatch.setattr(render_reuse_session, "edits_only_models", lambda **_kwargs: True) + + +def trust_changed_artifacts(monkeypatch: pytest.MonkeyPatch) -> None: + """Trust stored artifact digests even when the artifact changed since it was stored.""" + + monkeypatch.setattr(StoredArtifacts, "unchanged_since_stored", lambda _self, **_kwargs: True) + + def disable_change_detection(monkeypatch: pytest.MonkeyPatch) -> None: """Make compile reuse believe no project file changed since the stored compile.""" diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_incremental_edit_phases.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_incremental_edit_phases.py new file mode 100644 index 0000000000..f2520c73c2 --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_incremental_edit_phases.py @@ -0,0 +1,305 @@ +"""Edits that reuse stored declarations, analyses, and artifacts must match an uncached compile.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import ( + BrokenEditInvalidationTestCase, + IncrementalEditSequenceTestCase, + SharedCacheKeyTestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import ( + CompileReuseRun, + IncrementalEditComparison, + analyze_in_one_batch, + analyze_one_model_at_a_time, + build_between, + compare_incremental_compile, + compile_edit_without_reuse, + compile_in_process, + compile_without_reuse_between, + compiled_artifact_tampered, + enable_compile_reuse, + fact_comment, + fact_quantity_type_declared, + fact_quantity_type_removed, + full_edit_step, + in_process_reuse_run, + keep_invalidation, + leave_edit_uncompiled, + model_edit_step, + other_target_build_between, + payments_comment, + plan_between, + prime_render_store, + prime_render_store_in_process, + replace_project_text, + reuse_declarations_after_any_edit, + run_reuse_compile, + skip_upstream_analysis_checks, + staging_column_removed, + staging_column_renamed, + staging_column_restored, + staging_comment, + staging_extra_flag, + staging_extra_flag_removed, + staging_quantity_restored, + staging_quantity_text, + staging_rename_reverted, + star_chain_added, + star_chain_with_twin_added, + stg_orders_test_edit, + trust_changed_artifacts, + twin_header_changed, +) + + +@pytest.mark.parametrize( + "test_case", + [ + IncrementalEditSequenceTestCase( + description="upstream_shape_changes", + steps=( + model_edit_step("upstream_column_removed", staging_column_removed), + model_edit_step("edit_while_downstream_broken", payments_comment), + model_edit_step("upstream_column_restored", staging_column_restored), + model_edit_step("upstream_column_renamed", staging_column_renamed), + model_edit_step("upstream_rename_reverted", staging_rename_reverted), + ), + ), + IncrementalEditSequenceTestCase( + description="contract_changes", + steps=( + model_edit_step("downstream_type_declared", fact_quantity_type_declared), + model_edit_step("upstream_type_breaks_contract", staging_quantity_text), + model_edit_step("edit_after_contract_error", fact_comment), + model_edit_step("upstream_type_restored", staging_quantity_restored), + model_edit_step("downstream_type_removed", fact_quantity_type_removed), + ), + ), + IncrementalEditSequenceTestCase( + description="star_chain_after_plan", + steps=( + full_edit_step("star_chain_added", star_chain_added), + model_edit_step("upstream_column_added", staging_extra_flag, plan_between), + model_edit_step("edit_after_added_column", fact_comment), + model_edit_step( + "upstream_column_removed", staging_extra_flag_removed, plan_between + ), + model_edit_step("edit_after_removed_column", payments_comment), + ), + ), + IncrementalEditSequenceTestCase( + description="star_chain_after_build", + steps=( + full_edit_step("star_chain_added", star_chain_added), + model_edit_step("upstream_column_added", staging_extra_flag, build_between), + model_edit_step("edit_after_added_column", fact_comment), + model_edit_step( + "upstream_column_removed", staging_extra_flag_removed, build_between + ), + model_edit_step("edit_after_removed_column", payments_comment), + ), + ), + IncrementalEditSequenceTestCase( + description="star_chain_after_uncached_compile", + steps=( + full_edit_step("star_chain_added", star_chain_added), + model_edit_step( + "upstream_column_added", staging_extra_flag, compile_without_reuse_between + ), + model_edit_step("edit_after_added_column", fact_comment), + model_edit_step( + "upstream_column_removed", + staging_extra_flag_removed, + compile_without_reuse_between, + ), + model_edit_step("edit_after_removed_column", payments_comment), + ), + ), + IncrementalEditSequenceTestCase( + description="star_chain_after_other_target_build", + steps=( + full_edit_step("star_chain_added", star_chain_added), + model_edit_step( + "upstream_column_added", staging_extra_flag, other_target_build_between + ), + model_edit_step("edit_after_added_column", fact_comment), + model_edit_step( + "upstream_column_removed", + staging_extra_flag_removed, + other_target_build_between, + ), + model_edit_step("edit_after_removed_column", payments_comment), + ), + ), + IncrementalEditSequenceTestCase( + description="declarations_and_artifacts", + steps=( + full_edit_step( + "source_type_edit", + lambda root: replace_project_text( + root, + "sources/raw.yml", + " - name: quantity\n type: INTEGER", + " - name: quantity\n type: BIGINT", + ), + ), + model_edit_step("edit_after_source", staging_comment), + model_edit_step("compiled_artifact_tampered", compiled_artifact_tampered), + full_edit_step("test_edit", stg_orders_test_edit), + model_edit_step("edit_after_test", fact_comment), + ), + ), + ], + ids=lambda case: case.description, +) +def test_given_edit_sequence_when_reusing_stored_phases_then_each_step_matches_uncached_compile( + compile_reuse_project: Path, test_case: IncrementalEditSequenceTestCase +) -> None: + project_dir: Path = compile_reuse_project + cold: CompileReuseRun = run_reuse_compile(project_dir=project_dir) + primed: CompileReuseRun = prime_render_store(project_dir) + assert (cold.returncode, primed.returncode) == (0, 0), primed.stderr + + for step in test_case.steps: + step.edit(project_dir) + step.between(project_dir) + comparison: IncrementalEditComparison = compare_incremental_compile(project_dir=project_dir) + + assert comparison.matches is test_case.expected_matches_uncached, ( + step.description, + comparison.mismatched_artifacts, + comparison.incremental.stderr, + ) + assert (comparison.reused_renders > 0) is step.expected_render_reuse, step.description + assert ( + comparison.incremental.timings.get("analysis_reuse_hits", 0) > 0 + ) is step.expected_render_reuse, step.description + + +@pytest.mark.parametrize( + "test_case", + [ + BrokenEditInvalidationTestCase( + description="intact_invalidation_after_upstream_shape_change", + compiled_edits=(), + edit=staging_column_removed, + sabotage=keep_invalidation, + expected_matches_uncached=True, + after_edit=compile_edit_without_reuse, + ), + BrokenEditInvalidationTestCase( + description="analysis_served_despite_upstream_shape_change", + compiled_edits=(), + edit=staging_column_removed, + sabotage=skip_upstream_analysis_checks, + expected_matches_uncached=False, + after_edit=compile_edit_without_reuse, + ), + BrokenEditInvalidationTestCase( + description="intact_invalidation_after_test_edit", + compiled_edits=(), + edit=stg_orders_test_edit, + sabotage=keep_invalidation, + expected_matches_uncached=True, + after_edit=leave_edit_uncompiled, + ), + BrokenEditInvalidationTestCase( + description="declarations_reused_despite_test_edit", + compiled_edits=(), + edit=stg_orders_test_edit, + sabotage=reuse_declarations_after_any_edit, + expected_matches_uncached=False, + after_edit=leave_edit_uncompiled, + ), + BrokenEditInvalidationTestCase( + description="intact_invalidation_after_tampering", + compiled_edits=(), + edit=compiled_artifact_tampered, + sabotage=keep_invalidation, + expected_matches_uncached=True, + after_edit=leave_edit_uncompiled, + ), + BrokenEditInvalidationTestCase( + description="artifact_trusted_despite_tampering", + compiled_edits=(), + edit=compiled_artifact_tampered, + sabotage=trust_changed_artifacts, + expected_matches_uncached=False, + after_edit=leave_edit_uncompiled, + ), + ], + ids=lambda case: case.description, +) +def test_given_broken_edit_invalidation_when_compiling_an_edit_then_the_oracle_reports_a_mismatch( + compile_reuse_project: Path, + test_case: BrokenEditInvalidationTestCase, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + project_dir: Path = compile_reuse_project + enable_compile_reuse(monkeypatch) + assert prime_render_store_in_process(project_dir) == 0 + for edit in test_case.compiled_edits: + edit(project_dir) + assert compile_in_process(project_dir=project_dir) == 0 + test_case.edit(project_dir) + test_case.after_edit(project_dir, monkeypatch) + test_case.sabotage(monkeypatch) + + broken: CompileReuseRun = in_process_reuse_run(project_dir=project_dir, capsys=capsys) + reference: CompileReuseRun = run_reuse_compile(project_dir=project_dir, args=("--no-cache",)) + comparison: IncrementalEditComparison = IncrementalEditComparison( + incremental=broken, reference=reference + ) + + assert comparison.matches is test_case.expected_matches_uncached, ( + comparison.mismatched_artifacts + ) + + +@pytest.mark.parametrize( + "test_case", + [ + SharedCacheKeyTestCase( + description="dataflow_one_model_at_a_time", schedule=analyze_one_model_at_a_time + ), + SharedCacheKeyTestCase(description="one_batch", schedule=analyze_in_one_batch), + ], + ids=lambda case: case.description, +) +def test_given_model_sharing_a_served_cache_key_when_upstream_changes_then_it_matches_uncached( + compile_reuse_project: Path, + test_case: SharedCacheKeyTestCase, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + project_dir: Path = compile_reuse_project + enable_compile_reuse(monkeypatch) + test_case.schedule(monkeypatch) + star_chain_with_twin_added(project_dir) + assert prime_render_store_in_process(project_dir) == 0 + staging_extra_flag(project_dir) + twin_header_changed(project_dir) + compile_edit_without_reuse(project_dir, monkeypatch) + + incremental: CompileReuseRun = in_process_reuse_run(project_dir=project_dir, capsys=capsys) + reference: CompileReuseRun = in_process_reuse_run( + project_dir=project_dir, capsys=capsys, args=("--no-cache",) + ) + comparison: IncrementalEditComparison = IncrementalEditComparison( + incremental=incremental, reference=reference + ) + + assert comparison.matches is test_case.expected_matches_uncached, ( + comparison.mismatched_artifacts + ) + assert incremental.timings.get("analysis_reuse_hits", 0) > 0 + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-n", "auto", "--dist", "loadfile"])) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_incremental_edits.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_incremental_edits.py index fc0550161c..b4100aaac1 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_incremental_edits.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_incremental_edits.py @@ -221,6 +221,7 @@ def test_given_edit_sequence_when_compiling_incrementally_then_each_step_matches for step in test_case.steps: step.edit(project_dir) + step.between(project_dir) comparison: IncrementalEditComparison = compare_incremental_compile(project_dir=project_dir) assert comparison.incremental.reused is False, step.description @@ -256,6 +257,7 @@ def test_given_random_edit_chain_when_compiling_incrementally_then_each_step_mat random_edit_plan(seed=test_case.seed, step_count=test_case.step_count) ): chain.apply(kind) + chain.intervene() comparison: IncrementalEditComparison = compare_incremental_compile(project_dir=project_dir) assert comparison.matches is test_case.expected_matches_uncached, ( @@ -265,6 +267,9 @@ def test_given_random_edit_chain_when_compiling_incrementally_then_each_step_mat comparison.incremental.stderr, ) assert (comparison.reused_renders > 0) is is_model_only_edit(kind), (step, kind) + assert (comparison.incremental.timings.get("analysis_reuse_hits", 0) > 0) is ( + is_model_only_edit(kind) + ), (step, kind) @pytest.mark.parametrize( diff --git a/tests/unit/src/sqlbuild/cli/compile_reuse/classes/__init__.py b/tests/unit/src/sqlbuild/cli/compile_reuse/classes/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/unit/src/sqlbuild/cli/compile_reuse/classes/_test_types.py b/tests/unit/src/sqlbuild/cli/compile_reuse/classes/_test_types.py new file mode 100644 index 0000000000..7ed01cf136 --- /dev/null +++ b/tests/unit/src/sqlbuild/cli/compile_reuse/classes/_test_types.py @@ -0,0 +1,17 @@ +"""Test case types for trusting artifacts a stored compile left.""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path + + +@dataclass(frozen=True) +class StoredArtifactTestCase: + """A change to a stored artifact and whether its stored digest may still be trusted.""" + + description: str + change: Callable[[Path], None] + written_contents: bytes + expected_holds: bool diff --git a/tests/unit/src/sqlbuild/cli/compile_reuse/classes/helpers.py b/tests/unit/src/sqlbuild/cli/compile_reuse/classes/helpers.py new file mode 100644 index 0000000000..d7998b411c --- /dev/null +++ b/tests/unit/src/sqlbuild/cli/compile_reuse/classes/helpers.py @@ -0,0 +1,32 @@ +"""Builders for stored compile artifact unit tests.""" + +from __future__ import annotations + +import hashlib +from pathlib import Path + +from sqlbuild.cli.compile_reuse.constants import DIGEST_SIZE_BYTES + +STORED_SQL: bytes = b"SELECT order_id FROM orders\n" + + +def artifact_unchanged(_path: Path) -> None: + """Leave the stored artifact as it is.""" + + +def artifact_rewritten(path: Path) -> None: + """Rewrite the stored artifact with other contents.""" + + path.write_bytes(b"SELECT order_id FROM returns\n") + + +def artifact_removed(path: Path) -> None: + """Remove the stored artifact.""" + + path.unlink() + + +def artifact_digest(contents: bytes) -> str: + """Return the digest compile reuse records for artifact contents.""" + + return hashlib.blake2b(contents, digest_size=DIGEST_SIZE_BYTES).hexdigest() diff --git a/tests/unit/src/sqlbuild/cli/compile_reuse/classes/test_stored_artifacts.py b/tests/unit/src/sqlbuild/cli/compile_reuse/classes/test_stored_artifacts.py new file mode 100644 index 0000000000..cac7162a49 --- /dev/null +++ b/tests/unit/src/sqlbuild/cli/compile_reuse/classes/test_stored_artifacts.py @@ -0,0 +1,73 @@ +"""A stored artifact is trusted only while its stat identity and digest both still match.""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from sqlbuild.cli.compile_reuse._helpers.target_files import path_stamp +from sqlbuild.cli.compile_reuse.classes.stored_artifacts import StoredArtifacts +from sqlbuild.cli.compile_reuse.types import FileStamp +from tests.unit.src.sqlbuild.cli.compile_reuse.classes._test_types import StoredArtifactTestCase +from tests.unit.src.sqlbuild.cli.compile_reuse.classes.helpers import ( + STORED_SQL, + artifact_digest, + artifact_removed, + artifact_rewritten, + artifact_unchanged, +) + + +@pytest.mark.parametrize( + "test_case", + [ + StoredArtifactTestCase( + description="unchanged_same_contents", + change=artifact_unchanged, + written_contents=STORED_SQL, + expected_holds=True, + ), + StoredArtifactTestCase( + description="unchanged_new_contents", + change=artifact_unchanged, + written_contents=b"SELECT 1 AS order_id\n", + expected_holds=False, + ), + StoredArtifactTestCase( + description="rewritten_since_stored", + change=artifact_rewritten, + written_contents=STORED_SQL, + expected_holds=False, + ), + StoredArtifactTestCase( + description="removed_since_stored", + change=artifact_removed, + written_contents=STORED_SQL, + expected_holds=False, + ), + ], + ids=lambda case: case.description, +) +def test_given_stored_artifact_when_writing_then_trusted_only_while_stamp_and_digest_match( + tmp_path: Path, test_case: StoredArtifactTestCase +) -> None: + artifact: Path = tmp_path / "compiled" / "orders.sql" + artifact.parent.mkdir() + artifact.write_bytes(STORED_SQL) + path: str = os.fspath(artifact) + stamp: FileStamp | None = path_stamp(path=path) + assert stamp is not None + stored: StoredArtifacts = StoredArtifacts( + stamps={path: stamp}, digests={path: artifact_digest(STORED_SQL)} + ) + test_case.change(artifact) + + holds: bool = stored.holds(path=path, digest=artifact_digest(test_case.written_contents)) + + assert holds is test_case.expected_holds + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-n", "auto", "--dist", "loadfile"])) diff --git a/website/src/content/docs/docs/cli/compile.mdx b/website/src/content/docs/docs/cli/compile.mdx index affbda9659..2650ac6231 100644 --- a/website/src/content/docs/docs/cli/compile.mdx +++ b/website/src/content/docs/docs/cli/compile.mdx @@ -116,9 +116,14 @@ line to stderr before the report. When the only changes since the stored compile are edits to existing model files, compile runs in full but reuses the stored render of every unchanged model, together with the diagnostics and -environment reads that rendering reported. Only the edited models are rendered again. Analysis, -contracts, rules, and artifact writes still run for the whole project, so downstream effects of an -edit, such as a changed column type, are reported exactly as in an uncached compile. The output is +environment reads that rendering reported. Only the edited models are rendered again. Declaration +files other than models are not read again, and an unchanged model reuses its stored column +analysis only when the analyses of its upstream models in this compile match the ones it was stored +with. Contracts, semantic validation, and rules still run for the whole project, and a model whose +upstream analyses changed is analyzed again, so downstream effects of an edit, such as a changed +column type, are reported exactly as in an uncached compile, even when another command such as +`sqb build` ran in between. Compiled +files that still hold the bytes the stored compile wrote are not read or rewritten. The output is byte-identical to `sqb compile --no-cache`. Every other change renders the whole project: adding, removing, or renaming a model, and editing @@ -154,7 +159,7 @@ A replayed JSON report is byte-identical to the previous one except for `compile holds `project_reuse_hits`, `project_reuse_misses`, `project_reuse_bypasses`, `project_reuse_check_ms`, and `total_ms`. Full compiles report the same reuse counters alongside the phase timings, plus `render_reuse_hits` and `render_reuse_misses` for model renders reused or -rendered again after model-only edits. In text mode, a replayed compile does not print the per-phase progress lines. +rendered again after model-only edits and `analysis_reuse_hits` for stored model analyses reused. In text mode, a replayed compile does not print the per-phase progress lines. Lifecycle sinks receive only the invocation events for a replayed compile; per-phase `operation` events are not emitted because no compile phase runs.