diff --git a/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py b/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py index cede009293..5654563290 100644 --- a/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py +++ b/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py @@ -53,6 +53,7 @@ from sqlbuild.presentation.classes.transient_status_reporter import TransientStatusReporter from sqlbuild.rule_engine.classes.early_sql_lint import EarlySqlLint from sqlbuild.rule_engine.main.load_config import load_rules_config +from sqlbuild.rule_engine.main.render_rules_progress import format_rules_progress from sqlbuild.rule_engine.main.run_rules import run_rules from sqlbuild.rule_engine.models import RulesConfig, RulesRunResult from sqlbuild.runtime.observability.classes.operation_lifecycle import OperationLifecycle @@ -204,6 +205,7 @@ def _analyze_compile_project( ): prepared_artifacts.start(project=graph.project, adapter=adapter) _ = start_compile_phase(status=status, message="Evaluating built-in and custom rules...") + rules_start: float = time.monotonic() rules_result = run_rules( graph=graph, discovered_inputs=discovered_inputs, @@ -217,9 +219,8 @@ def _analyze_compile_project( ) _ = complete_compile_phase( status=status, - message=( - f"Evaluated rules. (built-in {rules_result.built_in_ms / 1000:.2f}s, " - f"custom {rules_result.custom_ms / 1000:.2f}s)" + message=format_rules_progress( + elapsed_seconds=time.monotonic() - rules_start, result=rules_result ), ) rule_diagnostics: tuple[CompilerDiagnostic, ...] = tuple( diff --git a/src/sqlbuild/compiler/fingerprints/_helpers/latest.py b/src/sqlbuild/compiler/fingerprints/_helpers/latest.py new file mode 100644 index 0000000000..7a360e9f9e --- /dev/null +++ b/src/sqlbuild/compiler/fingerprints/_helpers/latest.py @@ -0,0 +1,25 @@ +"""Latest fingerprint set assembly shared by warehouse reads and in-memory selection.""" + +from __future__ import annotations + +from collections.abc import Iterable + +from sqlbuild.compiler.fingerprints.constants import NODE_TYPE_MODEL +from sqlbuild.compiler.fingerprints.models import Fingerprint, FingerprintSet + + +def build_latest_fingerprint_set(*, schema: str, latest: Iterable[Fingerprint]) -> FingerprintSet: + """Index latest rows by identity and by node name, preferring models on name collisions.""" + + fingerprints: dict[str, Fingerprint] = {} + fingerprints_by_identity: dict[tuple[str, str], Fingerprint] = {} + fingerprint: Fingerprint + for fingerprint in latest: + fingerprints_by_identity[(fingerprint.node_type, fingerprint.node_name)] = fingerprint + if fingerprint.node_type == NODE_TYPE_MODEL or fingerprint.node_name not in fingerprints: + fingerprints[fingerprint.node_name] = fingerprint + return FingerprintSet( + schema=schema, + fingerprints=fingerprints, + fingerprints_by_identity=fingerprints_by_identity, + ) diff --git a/src/sqlbuild/compiler/fingerprints/main/_select_latest.py b/src/sqlbuild/compiler/fingerprints/main/_select_latest.py new file mode 100644 index 0000000000..656c02090b --- /dev/null +++ b/src/sqlbuild/compiler/fingerprints/main/_select_latest.py @@ -0,0 +1,28 @@ +"""In-memory node selection over an unfiltered latest fingerprint read.""" + +from __future__ import annotations + +from sqlbuild.compiler.fingerprints._helpers.latest import build_latest_fingerprint_set +from sqlbuild.compiler.fingerprints.models import FingerprintSet + + +def select_latest_fingerprints( + *, + fingerprint_set: FingerprintSet, + node_names: tuple[str, ...], + filtered_node_types: tuple[str, ...] = (), +) -> FingerprintSet: + """Apply the ``read_latest_fingerprints`` node filter to an unfiltered latest read.""" + + if not node_names and not filtered_node_types: + return FingerprintSet(schema=fingerprint_set.schema, fingerprints={}) + names: frozenset[str] = frozenset(node_names) + types: frozenset[str] = frozenset(filtered_node_types) + return build_latest_fingerprint_set( + schema=fingerprint_set.schema, + latest=( + fingerprint + for fingerprint in (fingerprint_set.fingerprints_by_identity or {}).values() + if fingerprint.node_name in names or (types and fingerprint.node_type not in types) + ), + ) diff --git a/src/sqlbuild/compiler/fingerprints/main/read.py b/src/sqlbuild/compiler/fingerprints/main/read.py index f22200c28e..df1a170199 100644 --- a/src/sqlbuild/compiler/fingerprints/main/read.py +++ b/src/sqlbuild/compiler/fingerprints/main/read.py @@ -9,10 +9,10 @@ from typing import Any from sqlbuild.adapter.contract.types import AdapterExecute +from sqlbuild.compiler.fingerprints._helpers.latest import build_latest_fingerprint_set from sqlbuild.compiler.fingerprints._helpers.sql import ( build_qualified_table_name, ) -from sqlbuild.compiler.fingerprints.constants import NODE_TYPE_MODEL from sqlbuild.compiler.fingerprints.exceptions import FingerprintInputError from sqlbuild.compiler.fingerprints.models import Fingerprint, FingerprintSet @@ -75,18 +75,9 @@ def read_latest_fingerprints( "fingerprint table to regenerate fingerprints." ) from error rows: list[tuple[Any, ...]] = result.fetchall() - fingerprints: dict[str, Fingerprint] = {} - fingerprints_by_identity: dict[tuple[str, str], Fingerprint] = {} - row: tuple[Any, ...] - for row in rows: - fingerprint: Fingerprint = _row_to_fingerprint(row=row, qualified_name=qualified_name) - fingerprints_by_identity[(fingerprint.node_type, fingerprint.node_name)] = fingerprint - if fingerprint.node_type == NODE_TYPE_MODEL or fingerprint.node_name not in fingerprints: - fingerprints[fingerprint.node_name] = fingerprint - return FingerprintSet( + return build_latest_fingerprint_set( schema=schema, - fingerprints=fingerprints, - fingerprints_by_identity=fingerprints_by_identity, + latest=(_row_to_fingerprint(row=row, qualified_name=qualified_name) for row in rows), ) diff --git a/src/sqlbuild/compiler/pipeline/main/compile.py b/src/sqlbuild/compiler/pipeline/main/compile.py index 1a4f08017c..b4a7800262 100644 --- a/src/sqlbuild/compiler/pipeline/main/compile.py +++ b/src/sqlbuild/compiler/pipeline/main/compile.py @@ -66,7 +66,7 @@ from sqlbuild.rule_engine.exceptions import RulesError from sqlbuild.rule_engine.main.load_config import load_rules_config from sqlbuild.rule_engine.main.render_result import format_result -from sqlbuild.rule_engine.main.render_skipped_rules import format_skipped_type_proof_rules +from sqlbuild.rule_engine.main.render_rules_progress import format_rules_progress from sqlbuild.rule_engine.main.run_rules import run_rules from sqlbuild.rule_engine.models import RulesCacheConfig, RulesConfig, RulesResult, RulesRunResult from sqlbuild.runtime.contracts.main.open_connection import open_connection_with_hooks @@ -106,6 +106,7 @@ def run_compile_pipeline( rules_config = replace(rules_config, cache=RulesCacheConfig(enabled=False)) if on_progress is not None: on_progress("Evaluating configured rules...") + rules_start: float = time.monotonic() rules_result: RulesRunResult = run_rules( graph=graph, discovered_inputs=discovered_inputs, @@ -120,13 +121,12 @@ def run_compile_pipeline( no_sql_analysis=resolved_options.no_sql_validation, ) if on_progress is not None: - skipped_rules_note: str | None = format_skipped_type_proof_rules( - codes=rules_result.skipped_type_proof_rules - ) on_progress( - f"Evaluated rules. (built-in {rules_result.built_in_ms / 1000:.2f}s, " - f"custom {rules_result.custom_ms / 1000:.2f}s)" - + ("" if skipped_rules_note is None else f"; {skipped_rules_note}") + format_rules_progress( + elapsed_seconds=time.monotonic() - rules_start, + result=rules_result, + note_skipped_rules=True, + ) ) if rules_result.findings: rendered: str = format_result( diff --git a/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py b/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py index d95c9230d8..bdd11253a8 100644 --- a/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py +++ b/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py @@ -3,7 +3,8 @@ from __future__ import annotations import re -from collections.abc import Sequence +from collections import ChainMap +from collections.abc import Mapping, Sequence from functools import lru_cache from typing import Any @@ -91,7 +92,7 @@ def build_relation_fixture_completion( authoritative_columns=context.authoritative_columns, adapter=adapter, ) - expected_types: dict[FixtureKey, dict[str, str]] = _expected_fixture_types( + expected_types: Mapping[FixtureKey, dict[str, str]] = _expected_fixture_types( base_expected_types=context.expected_types, ordered_model_names=ordered_model_names, model_map=model_map, @@ -327,7 +328,7 @@ def complete_empty_fixture_sql( def _typed_null_fixture_types( *, null_literal_names_by_fixture: dict[FixtureKey, frozenset[str]], - expected_types: dict[FixtureKey, dict[str, str]], + expected_types: Mapping[FixtureKey, dict[str, str]], ) -> dict[FixtureKey, dict[str, str]]: result: dict[FixtureKey, dict[str, str]] = {} for key, null_literal_names in null_literal_names_by_fixture.items(): @@ -665,8 +666,8 @@ def _expected_fixture_types( base_expected_types: dict[FixtureKey, dict[str, str]], ordered_model_names: tuple[str, ...], model_map: dict[str, CompiledModel], -) -> dict[FixtureKey, dict[str, str]]: - expected_types: dict[FixtureKey, dict[str, str]] = dict(base_expected_types) +) -> Mapping[FixtureKey, dict[str, str]]: + expected_types: ChainMap[FixtureKey, dict[str, str]] = ChainMap({}, base_expected_types) for model_name in ordered_model_names: model: CompiledModel | None = model_map.get(model_name) if model is None: diff --git a/src/sqlbuild/compiler/planner/_helpers/identity/graph.py b/src/sqlbuild/compiler/planner/_helpers/identity/graph.py index 1b7c8b26dd..be520d01e0 100644 --- a/src/sqlbuild/compiler/planner/_helpers/identity/graph.py +++ b/src/sqlbuild/compiler/planner/_helpers/identity/graph.py @@ -4,6 +4,7 @@ from collections.abc import Mapping +from sqlbuild.compiler.planner.classes.graph_identity_resolver import GraphIdentityResolver from sqlbuild.compiler.planner.models import GraphIdentityNode, GraphNodeKey from sqlbuild.compiler.planner.types import GraphIdentityComposer @@ -14,37 +15,13 @@ def build_expected_graph_identity_hashes( execution_order: tuple[GraphNodeKey, ...], compose_identity: GraphIdentityComposer, ) -> dict[GraphNodeKey, str | None]: - hashes: dict[GraphNodeKey, str | None] = {} - - def resolve( - *, - key: GraphNodeKey, - visiting: frozenset[GraphNodeKey], - cache: dict[GraphNodeKey, str | None], - ) -> tuple[str | None, dict[GraphNodeKey, str | None]]: - if key in cache: - return cache[key], cache - node: GraphIdentityNode | None = nodes.get(key) - if node is None or node.local_hash is None: - return None, {**cache, key: None} - if key in visiting: - return node.local_hash, {**cache, key: node.local_hash} - upstream_hashes: list[tuple[GraphNodeKey, str]] = [] - upstream_key: GraphNodeKey - for upstream_key in node.upstream_keys: - upstream_hash, cache = resolve(key=upstream_key, visiting=visiting | {key}, cache=cache) - if upstream_hash is not None: - upstream_hashes.append((upstream_key, upstream_hash)) - composed: str = compose_identity( - local_hash=node.local_hash, - upstream_hashes=tuple(upstream_hashes), - ) - return composed, {**cache, key: composed} - + resolver: GraphIdentityResolver = GraphIdentityResolver( + nodes=nodes, compose_identity=compose_identity, base_hashes={}, selected_keys=None + ) key: GraphNodeKey for key in execution_order: - _, hashes = resolve(key=key, visiting=frozenset(), cache=hashes) - return hashes + _ = resolver.resolve_expected(key) + return resolver.hashes def build_graph_write_identity_hashes( @@ -55,48 +32,13 @@ def build_graph_write_identity_hashes( base_identity_hashes: Mapping[GraphNodeKey, str], compose_identity: GraphIdentityComposer, ) -> dict[GraphNodeKey, str]: - hashes: dict[GraphNodeKey, str] = dict(base_identity_hashes) - resolved_selected: dict[GraphNodeKey, str] = {} - - def resolve( - *, - key: GraphNodeKey, - visiting: frozenset[GraphNodeKey], - cache: dict[GraphNodeKey, str], - selected_cache: dict[GraphNodeKey, str], - ) -> tuple[str | None, dict[GraphNodeKey, str], dict[GraphNodeKey, str]]: - if key not in selected_keys: - return cache.get(key), cache, selected_cache - if key in selected_cache: - return selected_cache[key], cache, selected_cache - node: GraphIdentityNode | None = nodes.get(key) - if node is None or node.local_hash is None: - return cache.get(key), cache, selected_cache - if key in visiting: - return node.local_hash, cache, selected_cache - upstream_hashes: list[tuple[GraphNodeKey, str]] = [] - upstream_key: GraphNodeKey - for upstream_key in node.upstream_keys: - upstream_hash, cache, selected_cache = resolve( - key=upstream_key, - visiting=visiting | {key}, - cache=cache, - selected_cache=selected_cache, - ) - if upstream_hash is not None: - upstream_hashes.append((upstream_key, upstream_hash)) - composed: str = compose_identity( - local_hash=node.local_hash, - upstream_hashes=tuple(upstream_hashes), - ) - return composed, {**cache, key: composed}, {**selected_cache, key: composed} - + resolver: GraphIdentityResolver = GraphIdentityResolver( + nodes=nodes, + compose_identity=compose_identity, + base_hashes=base_identity_hashes, + selected_keys=selected_keys, + ) key: GraphNodeKey for key in execution_order: - _, hashes, resolved_selected = resolve( - key=key, - visiting=frozenset(), - cache=hashes, - selected_cache=resolved_selected, - ) - return hashes + _ = resolver.resolve_write(key) + return {hashed_key: value for hashed_key, value in resolver.hashes.items() if value is not None} diff --git a/src/sqlbuild/compiler/planner/_helpers/migrations/planning.py b/src/sqlbuild/compiler/planner/_helpers/migrations/planning.py index df0491ea04..4b1638a0cc 100644 --- a/src/sqlbuild/compiler/planner/_helpers/migrations/planning.py +++ b/src/sqlbuild/compiler/planner/_helpers/migrations/planning.py @@ -147,6 +147,7 @@ def plan_model_migrations( adapter=runtime.adapter, connection=runtime.connection, database=planning_database(runtime=runtime), + known_fingerprints=snapshot.fingerprints, ) declares_migrations: bool = any( _declared_request(model) is not None for model in runtime.project.models diff --git a/src/sqlbuild/compiler/planner/_helpers/warehouse/snapshot.py b/src/sqlbuild/compiler/planner/_helpers/warehouse/snapshot.py index 1bb31e2133..aebb030fd9 100644 --- a/src/sqlbuild/compiler/planner/_helpers/warehouse/snapshot.py +++ b/src/sqlbuild/compiler/planner/_helpers/warehouse/snapshot.py @@ -24,6 +24,7 @@ run_recorded_inspection_query, ) from sqlbuild.adapter.relations.models import InspectionQueryRecord +from sqlbuild.compiler.compile.constants import MIGRATE_FROM_CONFIG_KEY from sqlbuild.compiler.compile.main._cursor_roles import resolve_cursor_input_roles from sqlbuild.compiler.compile.models import ( CompiledFunction, @@ -42,6 +43,7 @@ NODE_TYPE_MODEL, NODE_TYPE_SEED, ) +from sqlbuild.compiler.fingerprints.main._select_latest import select_latest_fingerprints from sqlbuild.compiler.fingerprints.main.read import read_latest_fingerprints from sqlbuild.compiler.fingerprints.models import Fingerprint, FingerprintSet from sqlbuild.compiler.graph.main.transitive_closure_many import transitive_closure_many @@ -297,6 +299,9 @@ def gather_warehouse_snapshot( schemas=query_schemas, fingerprint_state_schemas=fingerprint_state_schemas, node_names=_selected_node_names(relevant_keys), + read_unfiltered=_shares_unfiltered_fingerprints( + project=project, selected_keys=relevant_keys + ), ) effective_full_refresh_names: frozenset[str] = ( @@ -695,8 +700,9 @@ def _gather_fingerprints( schemas: tuple[str, ...] | None, fingerprint_state_schemas: frozenset[str], node_names: tuple[str, ...] | None, + read_unfiltered: bool, ) -> WarehouseFingerprints: - """Read latest fingerprints across all target schemas grouped by node type.""" + """Read latest fingerprints per target schema, keeping unfiltered reads for migrations.""" if schemas is None: return WarehouseFingerprints() @@ -704,6 +710,8 @@ def _gather_fingerprints( function_fingerprints: dict[str, Fingerprint] = {} seed_fingerprints: dict[str, Fingerprint] = {} python_fingerprints: dict[tuple[str, str], Fingerprint] = {} + unfiltered_schemas: dict[str, tuple[Fingerprint, ...]] = {} + filtered_node_types: tuple[str, ...] = (NODE_TYPE_MODEL, *FUNCTION_NODE_TYPES, NODE_TYPE_SEED) schema: str for schema in schemas: fingerprint_set: FingerprintSet = read_latest_fingerprints( @@ -714,9 +722,19 @@ def _gather_fingerprints( schema=schema, render_qualified_name=adapter.render_qualified_name, render_read_latest_sql=adapter.render_read_latest_fingerprints_sql, - node_names=node_names, - filtered_node_types=(NODE_TYPE_MODEL, *FUNCTION_NODE_TYPES, NODE_TYPE_SEED), + node_names=None if read_unfiltered else node_names, + filtered_node_types=filtered_node_types, ) + if read_unfiltered: + unfiltered_schemas[schema.lower()] = tuple( + (fingerprint_set.fingerprints_by_identity or {}).values() + ) + if node_names is not None: + fingerprint_set = select_latest_fingerprints( + fingerprint_set=fingerprint_set, + node_names=node_names, + filtered_node_types=filtered_node_types, + ) node_name: str fingerprint: Fingerprint for node_name, fingerprint in fingerprint_set.fingerprints.items(): @@ -740,6 +758,8 @@ def _gather_fingerprints( functions=function_fingerprints, seeds=seed_fingerprints, python_nodes=python_fingerprints, + unfiltered_schemas=unfiltered_schemas, + unfiltered_database=database, ) @@ -751,6 +771,22 @@ def _selected_node_names( return tuple(sorted({key.name for key in selected_keys})) +def _shares_unfiltered_fingerprints( + *, + project: CompiledProject, + selected_keys: frozenset[CompiledObjectKey] | None, +) -> bool: + """Return whether one unfiltered read can serve this selection and migration inspection.""" + + if selected_keys is None: + return True + selected: frozenset[str] = frozenset(key.name for key in selected_keys) + return all(model.name in selected for model in project.models) or any( + isinstance(model.config.values.get(MIGRATE_FROM_CONFIG_KEY), str) + for model in project.models + ) + + def _gather_cursor_snapshots( *, project: CompiledProject, diff --git a/src/sqlbuild/compiler/planner/classes/graph_identity_resolver.py b/src/sqlbuild/compiler/planner/classes/graph_identity_resolver.py new file mode 100644 index 0000000000..7d2f6e1fcf --- /dev/null +++ b/src/sqlbuild/compiler/planner/classes/graph_identity_resolver.py @@ -0,0 +1,85 @@ +"""Memoized graph identity resolution over one dependency graph.""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping + +from sqlbuild.compiler.planner.models import GraphIdentityNode, GraphNodeKey +from sqlbuild.compiler.planner.types import GraphIdentityComposer + +type _Resolve = Callable[[GraphNodeKey], str | None] + + +class GraphIdentityResolver: + """Resolve expected or selected-write identity hashes, each node once, in linear time.""" + + def __init__( + self, + *, + nodes: Mapping[GraphNodeKey, GraphIdentityNode], + compose_identity: GraphIdentityComposer, + base_hashes: Mapping[GraphNodeKey, str | None], + selected_keys: frozenset[GraphNodeKey] | None, + ) -> None: + self._nodes: Mapping[GraphNodeKey, GraphIdentityNode] = nodes + self._compose_identity: GraphIdentityComposer = compose_identity + self._selected_keys: frozenset[GraphNodeKey] | None = selected_keys + self._hashes: dict[GraphNodeKey, str | None] = dict(base_hashes) + self._resolved_selected: dict[GraphNodeKey, str] = {} + self._visiting: set[GraphNodeKey] = set() + + @property + def hashes(self) -> dict[GraphNodeKey, str | None]: + """Return every hash recorded so far, in first-recorded order.""" + + return self._hashes + + def resolve_expected(self, key: GraphNodeKey) -> str | None: + """Compose a node's expected hash from its upstream expected hashes.""" + + if key in self._hashes: + return self._hashes[key] + node: GraphIdentityNode | None = self._nodes.get(key) + if node is None or node.local_hash is None: + self._hashes[key] = None + return None + if key in self._visiting: + self._hashes[key] = node.local_hash + return node.local_hash + composed: str = self._compose( + key=key, local_hash=node.local_hash, node=node, resolve=self.resolve_expected + ) + self._hashes[key] = composed + return composed + + def resolve_write(self, key: GraphNodeKey) -> str | None: + """Recompute a selected node's hash from upstream write hashes; others keep theirs.""" + + if self._selected_keys is None or key not in self._selected_keys: + return self._hashes.get(key) + if key in self._resolved_selected: + return self._resolved_selected[key] + node: GraphIdentityNode | None = self._nodes.get(key) + if node is None or node.local_hash is None: + return self._hashes.get(key) + if key in self._visiting: + return node.local_hash + composed: str = self._compose( + key=key, local_hash=node.local_hash, node=node, resolve=self.resolve_write + ) + self._hashes[key] = composed + self._resolved_selected[key] = composed + return composed + + def _compose( + self, *, key: GraphNodeKey, local_hash: str, node: GraphIdentityNode, resolve: _Resolve + ) -> str: + self._visiting.add(key) + upstream_hashes: list[tuple[GraphNodeKey, str]] = [] + upstream_key: GraphNodeKey + for upstream_key in node.upstream_keys: + upstream_hash: str | None = resolve(upstream_key) + if upstream_hash is not None: + upstream_hashes.append((upstream_key, upstream_hash)) + self._visiting.discard(key) + return self._compose_identity(local_hash=local_hash, upstream_hashes=tuple(upstream_hashes)) diff --git a/src/sqlbuild/compiler/planner/classes/migration_state_inspection.py b/src/sqlbuild/compiler/planner/classes/migration_state_inspection.py index a612b03206..0a63ed1c45 100644 --- a/src/sqlbuild/compiler/planner/classes/migration_state_inspection.py +++ b/src/sqlbuild/compiler/planner/classes/migration_state_inspection.py @@ -35,15 +35,28 @@ OldNameViewEvent, OldNameViewHistory, ) +from sqlbuild.compiler.planner.models import WarehouseFingerprints class MigrationStateInspection: """Relations, fingerprints, and migration events for one planning database.""" - def __init__(self, *, adapter: BaseAdapter, connection: Any, database: str | None) -> None: + def __init__( + self, + *, + adapter: BaseAdapter, + connection: Any, + database: str | None, + known_fingerprints: WarehouseFingerprints | None = None, + ) -> None: self._adapter: BaseAdapter = adapter self._connection: Any = connection self._database: str | None = database + self._known_fingerprints: dict[str, tuple[Fingerprint, ...]] = ( + known_fingerprints.unfiltered_schemas + if known_fingerprints is not None and known_fingerprints.unfiltered_database == database + else {} + ) self._read_schemas: set[str] = set() self._fingerprints: dict[str, tuple[Fingerprint, ...]] = {} self._events: list[MigrationEvent] = [] @@ -210,19 +223,20 @@ def inspect_schemas(self, *, schemas: set[str]) -> None: self._fingerprints[schema.lower()] = self._read_model_fingerprints(schema=schema) def _read_model_fingerprints(self, *, schema: str) -> tuple[Fingerprint, ...]: - fingerprint_set: FingerprintSet = read_latest_fingerprints( - connection=self._connection, - execute=self._adapter.execute, - table_exists=True, - database=self._database, - schema=schema, - render_qualified_name=self._adapter.render_qualified_name, - render_read_latest_sql=self._adapter.render_read_latest_fingerprints_sql, - ) + latest: tuple[Fingerprint, ...] | None = self._known_fingerprints.get(schema.lower()) + if latest is None: + fingerprint_set: FingerprintSet = read_latest_fingerprints( + connection=self._connection, + execute=self._adapter.execute, + table_exists=True, + database=self._database, + schema=schema, + render_qualified_name=self._adapter.render_qualified_name, + render_read_latest_sql=self._adapter.render_read_latest_fingerprints_sql, + ) + latest = tuple((fingerprint_set.fingerprints_by_identity or {}).values()) return tuple( - fingerprint - for fingerprint in (fingerprint_set.fingerprints_by_identity or {}).values() - if fingerprint.node_type == NODE_TYPE_MODEL + fingerprint for fingerprint in latest if fingerprint.node_type == NODE_TYPE_MODEL ) def inspect_relations(self, *, locations: tuple[CompiledRelationLocation, ...]) -> None: diff --git a/src/sqlbuild/compiler/planner/models.py b/src/sqlbuild/compiler/planner/models.py index 2c2c510c49..9acbb34cc1 100644 --- a/src/sqlbuild/compiler/planner/models.py +++ b/src/sqlbuild/compiler/planner/models.py @@ -2,7 +2,7 @@ from __future__ import annotations -from collections.abc import Callable +from collections.abc import Callable, Mapping from dataclasses import dataclass, field from datetime import UTC, datetime from decimal import Decimal, InvalidOperation @@ -120,7 +120,7 @@ class RelationFixtureCompletion: fixture_sql_by_key: dict[FixtureKey, str] inferred_by_fixture: dict[FixtureKey, tuple[InferredColumn, ...]] - expected_types: dict[FixtureKey, dict[str, str]] + expected_types: Mapping[FixtureKey, dict[str, str]] diagnostics: tuple[RelationFixtureDiagnostic, ...] @@ -601,6 +601,10 @@ class WarehouseFingerprints: functions: dict[str, Fingerprint] = field(default_factory=dict) seeds: dict[str, Fingerprint] = field(default_factory=dict) python_nodes: dict[tuple[str, str], Fingerprint] = field(default_factory=dict) + unfiltered_schemas: dict[str, tuple[Fingerprint, ...]] = field( + default_factory=dict, compare=False, repr=False + ) + unfiltered_database: str | None = field(default=None, compare=False, repr=False) @dataclass(frozen=True) diff --git a/src/sqlbuild/rule_engine/main/render_rules_progress.py b/src/sqlbuild/rule_engine/main/render_rules_progress.py new file mode 100644 index 0000000000..11f4bc5531 --- /dev/null +++ b/src/sqlbuild/rule_engine/main/render_rules_progress.py @@ -0,0 +1,21 @@ +"""Rules phase completion line entrypoint.""" + +from sqlbuild.rule_engine.main.render_skipped_rules import format_skipped_type_proof_rules +from sqlbuild.rule_engine.models import RulesRunResult + + +def format_rules_progress( + *, elapsed_seconds: float, result: RulesRunResult, note_skipped_rules: bool = False +) -> str: + """Report the rules phase wall time first, then built-in and custom rule time.""" + + message: str = ( + f"Evaluated rules. ({elapsed_seconds:.2f}s; built-in {result.built_in_ms / 1000:.2f}s, " + f"custom {result.custom_ms / 1000:.2f}s)" + ) + skipped_rules_note: str | None = ( + format_skipped_type_proof_rules(codes=result.skipped_type_proof_rules) + if note_skipped_rules + else None + ) + return message if skipped_rules_note is None else f"{message}; {skipped_rules_note}" diff --git a/src/sqlbuild/runtime/observability/_helpers/identity.py b/src/sqlbuild/runtime/observability/_helpers/identity.py index 84d8b3f36b..599e68933e 100644 --- a/src/sqlbuild/runtime/observability/_helpers/identity.py +++ b/src/sqlbuild/runtime/observability/_helpers/identity.py @@ -5,7 +5,7 @@ from collections.abc import Iterator, Mapping from contextlib import contextmanager from contextvars import ContextVar, Token -from dataclasses import asdict, replace +from dataclasses import fields, replace from types import MappingProxyType from typing import cast from uuid import uuid4 @@ -20,6 +20,9 @@ ) +_IDENTITY_FIELD_NAMES: tuple[str, ...] = tuple(field.name for field in fields(ExecutionIdentity)) + + _CURRENT_INVOCATION_EXTERNAL_CONTEXT: ContextVar[Mapping[str, JSONValue]] = ContextVar( "sqlbuild_invocation_external_context", default=MappingProxyType({}) ) @@ -39,7 +42,7 @@ def identity_scope(identity: ExecutionIdentity) -> Iterator[ExecutionIdentity]: def execution_identity_to_dict(identity: ExecutionIdentity) -> dict[str, str | None]: - return asdict(identity) + return {name: getattr(identity, name) for name in _IDENTITY_FIELD_NAMES} def _required_current_identity() -> ExecutionIdentity: diff --git a/src/sqlbuild/runtime/observability/_helpers/validation.py b/src/sqlbuild/runtime/observability/_helpers/validation.py index 6e752841a3..1dac7d85e3 100644 --- a/src/sqlbuild/runtime/observability/_helpers/validation.py +++ b/src/sqlbuild/runtime/observability/_helpers/validation.py @@ -65,25 +65,42 @@ def validate_schema_version(*, value: object) -> None: def freeze_json(*, value: object, path: str) -> JSONValue: """Validate and recursively freeze a JSON-compatible value.""" + return _freeze_json(value=value, path=(path,)) + + +def _freeze_json(*, value: object, path: tuple[str | int, ...]) -> JSONValue: if value is None or isinstance(value, (str, bool, int)): return value if isinstance(value, float): if not math.isfinite(value): - raise ObservabilityValidationError(f"{path} must not contain NaN or infinity") + raise ObservabilityValidationError( + f"{_render_path(path)} must not contain NaN or infinity" + ) return value if isinstance(value, Mapping): frozen: dict[str, JSONValue] = {} for key, item in value.items(): if not isinstance(key, str): - raise ObservabilityValidationError(f"{path} keys must be strings") - frozen[key] = freeze_json(value=item, path=f"{path}.{key}") + raise ObservabilityValidationError(f"{_render_path(path)} keys must be strings") + frozen[key] = ( + item + if item is None or isinstance(item, (str, bool, int)) + else _freeze_json(value=item, path=(*path, key)) + ) return MappingProxyType(frozen) if isinstance(value, (list, tuple)): return tuple( - freeze_json(value=item, path=f"{path}[{index}]") for index, item in enumerate(value) + _freeze_json(value=item, path=(*path, index)) for index, item in enumerate(value) ) raise ObservabilityValidationError( - f"{path} contains non-JSON value of type {type(value).__name__}" + f"{_render_path(path)} contains non-JSON value of type {type(value).__name__}" + ) + + +def _render_path(path: tuple[str | int, ...]) -> str: + return "".join( + f"[{segment}]" if isinstance(segment, int) else f".{segment}" if index else segment + for index, segment in enumerate(path) ) diff --git a/src/sqlbuild/runtime/observability/classes/deferred_thread_starts.py b/src/sqlbuild/runtime/observability/classes/deferred_thread_starts.py new file mode 100644 index 0000000000..62fbb013db --- /dev/null +++ b/src/sqlbuild/runtime/observability/classes/deferred_thread_starts.py @@ -0,0 +1,83 @@ +"""One shared daemon that runs deferred thread starts once their deadline passes.""" + +from __future__ import annotations + +import heapq +import itertools +import threading +import time +from collections.abc import Callable + +type _DeferredDeadline = tuple[float, int] + +_COMPACTION_SLACK: int = 64 + + +class DeferredThreadStarts: + """Run each scheduled start callback at its monotonic deadline on one shared thread.""" + + def __init__(self, *, name: str) -> None: + self._name: str = name + self._condition: threading.Condition = threading.Condition() + self._deadlines: list[_DeferredDeadline] = [] + self._starts: dict[int, Callable[[], None]] = {} + self._order: itertools.count[int] = itertools.count() + self._thread: threading.Thread | None = None + + @property + def pending_count(self) -> int: + """Return how many scheduled starts have neither run nor been cancelled.""" + + with self._condition: + return len(self._starts) + + def schedule(self, *, delay_seconds: float, start: Callable[[], None]) -> int: + """Call ``start`` after ``delay_seconds`` unless cancelled; return its cancel token.""" + + token: int = next(self._order) + deadline: _DeferredDeadline = (time.monotonic() + delay_seconds, token) + with self._condition: + self._starts[token] = start + heapq.heappush(self._deadlines, deadline) + if self._thread is None: + self._thread = threading.Thread(target=self._run, name=self._name, daemon=True) + self._thread.start() + elif self._deadlines[0] is deadline: + self._condition.notify() + return token + + def cancel(self, token: int) -> None: + """Drop a scheduled start so nothing it references outlives its owner.""" + + with self._condition: + if self._starts.pop(token, None) is None: + return + if len(self._deadlines) > 2 * len(self._starts) + _COMPACTION_SLACK: + self._deadlines = [ + deadline for deadline in self._deadlines if deadline[1] in self._starts + ] + heapq.heapify(self._deadlines) + + def _run(self) -> None: + while True: + start: Callable[[], None] = self._next_due() + try: + start() + except BaseException: + pass + + def _next_due(self) -> Callable[[], None]: + with self._condition: + while True: + if not self._deadlines: + _ = self._condition.wait() + continue + due_at, token = self._deadlines[0] + remaining: float = due_at - time.monotonic() + if remaining > 0: + _ = self._condition.wait(remaining) + continue + heapq.heappop(self._deadlines) + start: Callable[[], None] | None = self._starts.pop(token, None) + if start is not None: + return start diff --git a/src/sqlbuild/runtime/observability/classes/event_dispatcher.py b/src/sqlbuild/runtime/observability/classes/event_dispatcher.py index b1742c8446..4bffd0c4ff 100644 --- a/src/sqlbuild/runtime/observability/classes/event_dispatcher.py +++ b/src/sqlbuild/runtime/observability/classes/event_dispatcher.py @@ -4,7 +4,6 @@ from collections import deque from contextvars import ContextVar, Token -from dataclasses import replace from functools import partial from threading import RLock from typing import Literal, cast, overload @@ -14,7 +13,6 @@ _safe_error_type, _safe_subscriber_name, ) -from sqlbuild.runtime.observability._helpers.validation import validate_known_lifecycle_event from sqlbuild.runtime.observability.constants import ( LIFECYCLE_INVOCATION_METADATA_SCHEMA_VERSION, ) @@ -98,8 +96,6 @@ def publish_lifecycle(self, event: LifecycleEvent | OpaqueLifecycleEvent) -> Non raise ObservabilityValidationError( "lifecycle publication requires LifecycleEvent or OpaqueLifecycleEvent" ) - if isinstance(event, LifecycleEvent): - validate_known_lifecycle_event(event=event) self._publish(channel=_LIFECYCLE_CHANNEL, value=event) def publish_diagnostic(self, log: DiagnosticLog) -> None: @@ -130,7 +126,7 @@ def _publish( ): sequence: int = self._lifecycle_sequences.get(lifecycle_value.invocation_id, 0) self._lifecycle_sequences[lifecycle_value.invocation_id] = sequence + 1 - lifecycle_value = replace(lifecycle_value, invocation_sequence=sequence) + lifecycle_value = lifecycle_value.with_invocation_sequence(sequence) self._pending_publications.append((channel, lifecycle_value, suppress_health)) else: diagnostic_value: DiagnosticLog = cast(DiagnosticLog, value) diff --git a/src/sqlbuild/runtime/observability/classes/statement_lifecycle.py b/src/sqlbuild/runtime/observability/classes/statement_lifecycle.py index b4e034e447..10dc1dbb38 100644 --- a/src/sqlbuild/runtime/observability/classes/statement_lifecycle.py +++ b/src/sqlbuild/runtime/observability/classes/statement_lifecycle.py @@ -81,6 +81,7 @@ def __init__( None ) self._pending_failure: tuple[BaseException, str | None, str | None] | None = None + self._statement_facts: tuple[str, str] | None = None @staticmethod @contextmanager @@ -268,11 +269,17 @@ def __exit__( notify_statement_finished(sql=self._sql) def _base_payload(self) -> dict[str, JSONValue]: + if self._statement_facts is None: + self._statement_facts = ( + hashlib.sha256(self._sql.encode("utf-8")).hexdigest(), + _statement_kind(sql=self._sql), + ) + sql_digest, statement_kind = self._statement_facts payload: dict[str, JSONValue] = { "adapter": self._adapter, "intent": self._intent, - "sql_digest": hashlib.sha256(self._sql.encode("utf-8")).hexdigest(), - "statement_kind": _statement_kind(sql=self._sql), + "sql_digest": sql_digest, + "statement_kind": statement_kind, } if self._batch_size is not None: payload["batch_size"] = self._batch_size diff --git a/src/sqlbuild/runtime/observability/classes/statement_monitor.py b/src/sqlbuild/runtime/observability/classes/statement_monitor.py index 438024d213..d662dd2000 100644 --- a/src/sqlbuild/runtime/observability/classes/statement_monitor.py +++ b/src/sqlbuild/runtime/observability/classes/statement_monitor.py @@ -7,9 +7,13 @@ from collections.abc import Callable from contextvars import Context, copy_context +from sqlbuild.runtime.observability.classes.deferred_thread_starts import DeferredThreadStarts from sqlbuild.runtime.observability.constants import STATEMENT_HEARTBEAT_THRESHOLD_SECONDS _STATEMENT_QUERY_ID_POLL_SECONDS: float = 0.1 +_DEFERRED_MONITOR_STARTS: DeferredThreadStarts = DeferredThreadStarts( + name="sqlbuild-statement-monitor-starter" +) class StatementMonitor: @@ -38,6 +42,10 @@ def __init__( self._query_id_provider: Callable[[], str | None] | None = None self._query_id: str | None = None self._thread: threading.Thread | None = None + self._thread_lock: threading.Lock = threading.Lock() + self._context: Context | None = None + self._stopped: bool = False + self._deferred_start: int | None = None @property def query_id(self) -> str | None: @@ -46,22 +54,23 @@ def query_id(self) -> str | None: return self._query_id def start(self) -> None: - """Start monitoring in a daemon thread with the current execution context.""" - - context: Context = copy_context() - self._thread = threading.Thread( - target=context.run, - args=(self._run,), - name="sqlbuild-statement-monitor", - daemon=True, + """Monitor in a daemon thread, deferred to the first heartbeat without a query-ID poll.""" + + self._context = copy_context() + if self._has_query_id_provider() or self._threshold_seconds <= 0: + self._start_thread() + return + self._deferred_start = _DEFERRED_MONITOR_STARTS.schedule( + delay_seconds=self._threshold_seconds, start=self._start_thread ) - self._thread.start() def set_query_id_provider(self, provider: Callable[[], str | None]) -> None: """Install a non-blocking adapter query-ID reader.""" with self._provider_lock: self._query_id_provider = provider + if self._context is not None: + self._start_thread() self._wake_event.set() def stop(self) -> str | None: @@ -70,11 +79,41 @@ def stop(self) -> str | None: self._capture_query_id() self._stop_event.set() self._wake_event.set() - thread: threading.Thread | None = self._thread + with self._thread_lock: + self._stopped = True + self._context = None + thread: threading.Thread | None = self._thread + deferred_start: int | None = self._deferred_start + self._deferred_start = None + if deferred_start is not None: + _DEFERRED_MONITOR_STARTS.cancel(deferred_start) if thread is not None and thread is not threading.current_thread(): thread.join() + self._release_callbacks() return self._query_id + def _release_callbacks(self) -> None: + """Drop references back to the statement so a stopped monitor retains nothing.""" + + with self._provider_lock: + self._query_id_provider = None + self._on_submitted = lambda query_id: None + self._on_heartbeat = lambda elapsed_seconds, query_id: None + + def _start_thread(self) -> None: + with self._thread_lock: + context: Context | None = self._context + if self._stopped or self._thread is not None or context is None: + return + thread: threading.Thread = threading.Thread( + target=context.run, + args=(self._run,), + name="sqlbuild-statement-monitor", + daemon=True, + ) + thread.start() + self._thread = thread + def _run(self) -> None: next_heartbeat: float = self._started_at + self._threshold_seconds while not self._stop_event.is_set(): diff --git a/src/sqlbuild/runtime/observability/models.py b/src/sqlbuild/runtime/observability/models.py index 0e937361d4..1efbb621d0 100644 --- a/src/sqlbuild/runtime/observability/models.py +++ b/src/sqlbuild/runtime/observability/models.py @@ -2,6 +2,7 @@ from __future__ import annotations +import copy from collections.abc import Mapping from dataclasses import dataclass, field from datetime import datetime @@ -190,6 +191,17 @@ def __post_init__(self) -> None: object.__setattr__(self, "payload", frozen_payload) validate_known_lifecycle_event(event=self) + def with_invocation_sequence(self, sequence: int) -> LifecycleEvent: + """Return this validated event with a dispatcher-assigned invocation sequence.""" + + if isinstance(sequence, bool) or not isinstance(sequence, int) or sequence < 0: + raise ObservabilityValidationError( + "invocation_sequence must be a non-negative integer excluding bool" + ) + sequenced: LifecycleEvent = copy.copy(self) + object.__setattr__(sequenced, "invocation_sequence", sequence) + return sequenced + @dataclass(frozen=True) class OpaqueLifecycleEvent: 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 9f3215492c..604db55122 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 @@ -331,6 +331,16 @@ class PlanScalingGuardTestCase: expected_max_linear_factor: float +@dataclass(frozen=True) +class PlanPhaseScalingGuardTestCase: + description: str + small_model_count: int + runs: int + expected_phases: tuple[str, ...] + expected_max_linear_factor: float + expected_max_large_phase_seconds: float + + @dataclass(frozen=True) class BuildPerformanceGuardTestCase: description: str 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 6d30815bdf..237943741e 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py @@ -643,6 +643,7 @@ class _TimedSqbRun(NamedTuple): minor_page_faults: int payload: object output_path: Path + stderr_path: Path def _run_timed_sqb( @@ -703,6 +704,7 @@ def _run_timed_sqb( minor_page_faults=int(minor_text), payload=payload, output_path=output_path, + stderr_path=stderr_path, ) @@ -712,6 +714,7 @@ class InspectionCommandMeasurement(NamedTuple): cpu_seconds: float payload: object output: str + stderr: str = "" def run_fresh_process_inspection_command( @@ -735,9 +738,34 @@ def run_fresh_process_inspection_command( cpu_seconds=run.cpu_seconds, payload=run.payload, output=run.output_path.read_text(encoding="utf-8"), + stderr=run.stderr_path.read_text(encoding="utf-8"), ) +_PLAN_PHASE_LINE: re.Pattern[str] = re.compile( + r"^(Inspected warehouse state|Generated plan)\. \((\d+\.\d+)s\)$", re.MULTILINE +) + + +def fastest_plan_phase_seconds(*, project_dir: Path, label: str, runs: int) -> dict[str, float]: + """Plan without the compile cache several times and keep each phase's fastest time.""" + + fastest: dict[str, float] = {} + index: int + for index in range(runs): + measurement: InspectionCommandMeasurement = run_fresh_process_inspection_command( + project_dir=project_dir, + label=f"{label}-{index}", + sqb_args=("plan", "--json", "--no-cache"), + expected_max_wall_seconds=120.0, + ) + phase: str + seconds: str + for phase, seconds in _PLAN_PHASE_LINE.findall(measurement.stderr): + fastest[phase] = min(float(seconds), fastest.get(phase, float(seconds))) + return fastest + + INSPECTION_DIAMOND_LAYERS: int = 16 INSPECTION_DIAMOND_WIDTH: int = 8 INSPECTION_BENCHMARK_MODEL_COUNT: int = ( diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py index a7aae1f4a8..c7e463afc8 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py @@ -18,15 +18,20 @@ from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import ( InspectionCommandPerformanceGuardTestCase, PlanComparedToCompileGuardTestCase, + PlanPhaseScalingGuardTestCase, PlanScalingGuardTestCase, ) from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import ( INSPECTION_BENCHMARK_MODEL_COUNT, + INSPECTION_DIAMOND_LAYERS, + INSPECTION_DIAMOND_WIDTH, SHARED_DIAMOND_HUB, SHARED_DIAMOND_ROLLUP, InspectionCommandMeasurement, + fastest_plan_phase_seconds, prepare_small_inspection_project, run_fresh_process_inspection_command, + write_inspection_benchmark_project, ) _LOGGER: logging.Logger = logging.getLogger(__name__) @@ -285,3 +290,58 @@ def test_given_benchmarks_of_two_sizes_when_planning_then_planning_work_scales_l linear_cpu, ) assert planning_cpu["large"] < test_case.expected_max_linear_factor * linear_cpu + + +@pytest.mark.performance +@pytest.mark.cold_compile_performance +@pytest.mark.parametrize( + "test_case", + [ + PlanPhaseScalingGuardTestCase( + description="models_3000_plan_phases_scale_linearly_from_1000_models", + small_model_count=1_000, + runs=2, + expected_phases=("Inspected warehouse state", "Generated plan"), + expected_max_linear_factor=1.5, + expected_max_large_phase_seconds=20.0, + ) + ], + ids=lambda case: case.description, +) +def test_given_benchmarks_of_two_sizes_when_planning_then_each_plan_phase_scales_linearly( + inspection_benchmark_project: Path, + test_case: PlanPhaseScalingGuardTestCase, + tmp_path: Path, +) -> None: + small_project: Path = tmp_path / "semantic_small" + write_inspection_benchmark_project( + project_dir=small_project, model_count=test_case.small_model_count + ) + lattice_models: int = INSPECTION_DIAMOND_LAYERS * INSPECTION_DIAMOND_WIDTH + 2 + size_ratio: float = _MODEL_COUNT / (test_case.small_model_count + lattice_models) + fastest: dict[str, dict[str, float]] = { + label: fastest_plan_phase_seconds( + project_dir=project_dir, + label=f"{test_case.description}-{label}-plan", + runs=test_case.runs, + ) + for label, project_dir in ( + ("small", small_project), + ("large", inspection_benchmark_project), + ) + } + + _LOGGER.info( + "plan phase scaling %s size_ratio=%.2f small=%s large=%s", + test_case.description, + size_ratio, + fastest["small"], + fastest["large"], + ) + ratios: dict[str, float] = { + phase: seconds / max(fastest["small"].get(phase, 0.0), 0.01) + for phase, seconds in fastest["large"].items() + } + assert set(fastest["large"]) == set(fastest["small"]) == set(test_case.expected_phases) + assert max(ratios.values()) < test_case.expected_max_linear_factor * size_ratio, ratios + assert max(fastest["large"].values()) < test_case.expected_max_large_phase_seconds diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/plan/test_terminal_progress.py b/tests/e2e/src/sqlbuild/cli/commands/main/plan/test_terminal_progress.py index d89032a16c..4850001e11 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/plan/test_terminal_progress.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/plan/test_terminal_progress.py @@ -32,7 +32,8 @@ columns=100, expected_screen_fragments=( "Compiled project.", - "Evaluated rules.", + "Evaluated rules. (", + "s; built-in ", "Connecting to duckdb...", "✓ Warehouse connected duckdb", "Generated plan.", diff --git a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/_test_types.py b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/_test_types.py index b79aa64a37..c690e2e4e3 100644 --- a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/_test_types.py +++ b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/_test_types.py @@ -143,3 +143,14 @@ class MigrationFingerprintWorkTestCase: expected_computations: int expected_migrations: int expected_stored_models: tuple[str, ...] = () + + +@dataclass(frozen=True) +class FingerprintStateReadTestCase: + description: str + prior_models: tuple[Callable[[], dict[str, str]], ...] + models: Callable[[], dict[str, str]] + plan_args: tuple[str, ...] + expected_reads: int + expected_rows_read: Callable[[int], int] + expected_migrations: tuple[tuple[str | None, str, str, str], ...] diff --git a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/helpers.py b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/helpers.py index c0e7c0a8c9..72d48a5e89 100644 --- a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/helpers.py +++ b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/helpers.py @@ -18,6 +18,7 @@ from sqlbuild.adapter.contract.types import MigrationTransfer from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter from sqlbuild.cli.commands.main.entrypoint.entry import main +from sqlbuild.compiler.fingerprints.constants import FINGERPRINT_TABLE_NAME from sqlbuild.executor.build._helpers import scheduler as scheduler_module DATABASE_FILE: str = "orders.duckdb" @@ -806,3 +807,53 @@ def build_each( for models in model_sets: write_project(project_dir=project_dir, models=models()) _ = build_ok(project_dir=project_dir, capsys=capsys) + + +def _is_latest_fingerprint_read(sql: str) -> bool: + return FINGERPRINT_TABLE_NAME in sql and "__sqlbuild_latest_rank" in sql + + +def record_fingerprint_state_reads(monkeypatch: pytest.MonkeyPatch) -> list[int]: + """Record the row count of every latest-fingerprint read the DuckDB adapter executes.""" + + rows_read: list[int] = [] + original: Callable[..., Any] = DuckDbAdapter._execute + + def recording(self: DuckDbAdapter, *, connection: Any, sql: str) -> Any: + rows_read.extend( + int(connection.execute(f"SELECT count(*) FROM ({read_sql})").fetchone()[0]) + for read_sql in filter(_is_latest_fingerprint_read, (sql,)) + ) + return original(self, connection=connection, sql=sql) + + monkeypatch.setattr(DuckDbAdapter, "_execute", recording) + return rows_read + + +def fingerprint_history(*, project_dir: Path) -> tuple[int, int]: + """Return (stored rows, distinct node identities) of the fingerprint state table.""" + + row: tuple[Any, ...] = query( + project_dir=project_dir, + sql=( + "SELECT count(*), count(DISTINCT (node_type, node_name)) " + f"FROM main.{FINGERPRINT_TABLE_NAME}" + ), + )[0] + return int(row[0]), int(row[1]) + + +def edited_original_order_models() -> dict[str, str]: + """Return the original connected models with a comment edit that rebuilds each one.""" + + return {name: f"{sql}-- revised\n" for name, sql in original_order_models().items()} + + +def declared_rename_models() -> dict[str, str]: + """Return the original models with the staging model renamed through migrate_from.""" + + return { + DESTINATION_MODEL: incremental_orders_sql(migrate_from=ORIGIN_MODEL), + "orders_enriched": enriched_view_sql(upstream=DESTINATION_MODEL, alias="o"), + "daily_order_totals": daily_totals_sql(upstream="orders_enriched", cte="orders_enriched"), + } diff --git a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/test_fingerprint_state_reads.py b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/test_fingerprint_state_reads.py new file mode 100644 index 0000000000..ce275f1177 --- /dev/null +++ b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/test_fingerprint_state_reads.py @@ -0,0 +1,88 @@ +"""Integration coverage for how often planning reads the fingerprint state table.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from tests.integration.src.sqlbuild.cli.commands.main.model_migrations._test_types import ( + FingerprintStateReadTestCase, +) +from tests.integration.src.sqlbuild.cli.commands.main.model_migrations.helpers import ( + build_each, + declared_rename_models, + edited_original_order_models, + fingerprint_history, + load_raw_orders, + original_order_models, + plan_json, + planned_migrations, + record_fingerprint_state_reads, + renamed_order_models, + write_project, +) + + +@pytest.mark.parametrize( + "test_case", + [ + FingerprintStateReadTestCase( + description="full plan discovering renames reads latest state once", + prior_models=(original_order_models, edited_original_order_models), + models=renamed_order_models, + plan_args=(), + expected_reads=1, + expected_rows_read=lambda identities: identities, + expected_migrations=( + ("daily_order_totals", "customer_daily_order_totals", "automatic", "migrate"), + ("orders_enriched", "customer_orders_enriched", "automatic", "renamed"), + ("stg_orders", "stg_customer_orders", "automatic", "migrate"), + ), + ), + FingerprintStateReadTestCase( + description="selected plan of a project declaring migrations reads latest state once", + prior_models=(original_order_models, edited_original_order_models), + models=declared_rename_models, + plan_args=("-s", "stg_customer_orders"), + expected_reads=1, + expected_rows_read=lambda identities: identities, + expected_migrations=(("stg_orders", "stg_customer_orders", "manual", "migrate"),), + ), + FingerprintStateReadTestCase( + description="selected plan without migration work reads only the selected state", + prior_models=(original_order_models, edited_original_order_models), + models=original_order_models, + plan_args=("-s", "stg_orders"), + expected_reads=1, + expected_rows_read=lambda identities: 1, + expected_migrations=(), + ), + ], + ids=lambda case: case.description, +) +def test_given_fingerprint_history_when_planning_then_reads_latest_state_once( + test_case: FingerprintStateReadTestCase, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + load_raw_orders(project_dir=tmp_path, first_day=1, last_day=5) + build_each(project_dir=tmp_path, model_sets=test_case.prior_models, capsys=capsys) + stored_rows: int + identities: int + stored_rows, identities = fingerprint_history(project_dir=tmp_path) + write_project(project_dir=tmp_path, models=test_case.models()) + rows_read: list[int] = record_fingerprint_state_reads(monkeypatch) + + migrations: tuple[tuple[str | None, str, str, str], ...] = planned_migrations( + plan_json(project_dir=tmp_path, capsys=capsys, args=test_case.plan_args) + ) + + assert stored_rows > identities + assert migrations == test_case.expected_migrations + assert rows_read == [test_case.expected_rows_read(identities)] * test_case.expected_reads + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/integration/src/sqlbuild/compiler/fingerprints/main/_test_types.py b/tests/integration/src/sqlbuild/compiler/fingerprints/main/_test_types.py index 4015e53c80..495329ab8c 100644 --- a/tests/integration/src/sqlbuild/compiler/fingerprints/main/_test_types.py +++ b/tests/integration/src/sqlbuild/compiler/fingerprints/main/_test_types.py @@ -87,3 +87,11 @@ class PruneFingerprintHistoryTestCase: fingerprints: tuple[Fingerprint, ...] expected_run_ids_by_identity: dict[tuple[str, str], tuple[str, ...]] expected_latest_run_id: str + + +@dataclass(frozen=True) +class SelectLatestTestCase: + description: str + node_names: tuple[str, ...] + filtered_node_types: tuple[str, ...] + expected_identities: frozenset[tuple[str, str]] diff --git a/tests/integration/src/sqlbuild/compiler/fingerprints/main/helpers.py b/tests/integration/src/sqlbuild/compiler/fingerprints/main/helpers.py index 100c220b4c..1ac1db016c 100644 --- a/tests/integration/src/sqlbuild/compiler/fingerprints/main/helpers.py +++ b/tests/integration/src/sqlbuild/compiler/fingerprints/main/helpers.py @@ -1,5 +1,6 @@ from __future__ import annotations +import itertools import threading from datetime import UTC, datetime from pathlib import Path @@ -83,3 +84,40 @@ def run_concurrent_fingerprint_write_round( ).fetchone() check_connection.close() return int(getattr(row, "__getitem__", lambda _index: 0)(0)) + + +_MIXED_IDENTITIES: tuple[tuple[str, str], ...] = ( + ("model", "orders"), + ("model", "customers"), + ("seed", "orders"), + ("udf", "is_large_order"), + ("python", "export_orders"), +) +_HISTORY_RUNS: tuple[tuple[str, int], ...] = (("run_001", 10), ("run_002", 12), ("run_003", 12)) + + +def _history_fingerprint(*, identity: tuple[str, str], run: tuple[str, int]) -> Fingerprint: + node_type, node_name = identity + run_id, hour = run + return Fingerprint( + node_type=node_type, + node_name=node_name, + target_database=None, + target_schema=None, + target_name=node_name, + run_id=run_id, + definition_hash=f"{node_name}_{run_id}", + version_hash=f"{node_name}_{run_id}", + schema_fingerprint="", + definition=f"SELECT '{run_id}'", + ts=datetime(2026, 1, 15, hour, 0, 0), + ) + + +def mixed_fingerprint_history() -> tuple[Fingerprint, ...]: + """Return three runs of history for models, a seed, a function, and a Python node.""" + + return tuple( + _history_fingerprint(identity=identity, run=run) + for identity, run in itertools.product(_MIXED_IDENTITIES, _HISTORY_RUNS) + ) diff --git a/tests/integration/src/sqlbuild/compiler/fingerprints/main/test_read_write.py b/tests/integration/src/sqlbuild/compiler/fingerprints/main/test_read_write.py index 2259c049d9..27d56ed8d0 100644 --- a/tests/integration/src/sqlbuild/compiler/fingerprints/main/test_read_write.py +++ b/tests/integration/src/sqlbuild/compiler/fingerprints/main/test_read_write.py @@ -10,6 +10,7 @@ from sqlbuild.adapter.contract.types import FrameworkType from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter from sqlbuild.compiler.fingerprints.constants import FINGERPRINT_TABLE_NAME +from sqlbuild.compiler.fingerprints.main._select_latest import select_latest_fingerprints from sqlbuild.compiler.fingerprints.main.read import read_latest_fingerprints from sqlbuild.compiler.fingerprints.main.write import write_fingerprint from sqlbuild.compiler.fingerprints.main.write_many import write_fingerprints @@ -21,9 +22,13 @@ OldFingerprintSchemaTestCase, PruneFingerprintHistoryTestCase, ReadNonExistentTableTestCase, + SelectLatestTestCase, WriteAndReadTestCase, WriteCreatesTableTestCase, ) +from tests.integration.src.sqlbuild.compiler.fingerprints.main.helpers import ( + mixed_fingerprint_history, +) RENDER_QUALIFIED_NAME: Callable[..., str | None] = DuckDbAdapter().render_qualified_name RENDER_FRAMEWORK_TYPE: Callable[[FrameworkType], str] = DuckDbAdapter().render_framework_type @@ -897,3 +902,76 @@ def test_given_invalid_definition_storage_when_reading_then_raises_contextual_er fragment: str for fragment in test_case.expected_error_fragments: assert fragment in message + + +@pytest.mark.parametrize( + "test_case", + [ + SelectLatestTestCase( + description="keeps named nodes of every type and unfiltered node types", + node_names=("orders",), + filtered_node_types=("model", "udf", "seed"), + expected_identities=frozenset( + {("model", "orders"), ("seed", "orders"), ("python", "export_orders")} + ), + ), + SelectLatestTestCase( + description="keeps only unfiltered node types without names", + node_names=(), + filtered_node_types=("model", "udf", "seed"), + expected_identities=frozenset({("python", "export_orders")}), + ), + SelectLatestTestCase( + description="keeps only named nodes without filtered types", + node_names=("customers", "is_large_order"), + filtered_node_types=(), + expected_identities=frozenset({("model", "customers"), ("udf", "is_large_order")}), + ), + SelectLatestTestCase( + description="keeps nothing without names or filtered types", + node_names=(), + filtered_node_types=(), + expected_identities=frozenset(), + ), + ], + ids=lambda case: case.description, +) +def test_given_unfiltered_latest_read_when_selecting_then_matches_filtered_read( + test_case: SelectLatestTestCase, + connection: Any, + execute: Any, +) -> None: + write_fingerprints( + connection=connection, + execute=execute, + database=None, + schema="test_schema", + fingerprints=mixed_fingerprint_history(), + render_qualified_name=RENDER_QUALIFIED_NAME, + render_framework_type=RENDER_FRAMEWORK_TYPE, + ) + + def read(node_names: tuple[str, ...] | None) -> FingerprintSet: + return read_latest_fingerprints( + connection=connection, + execute=execute, + table_exists=True, + database=None, + schema="test_schema", + render_qualified_name=RENDER_QUALIFIED_NAME, + render_read_latest_sql=RENDER_READ_LATEST_SQL, + node_names=node_names, + filtered_node_types=test_case.filtered_node_types, + ) + + filtered: FingerprintSet = read(test_case.node_names) + selected: FingerprintSet = select_latest_fingerprints( + fingerprint_set=read(None), + node_names=test_case.node_names, + filtered_node_types=test_case.filtered_node_types, + ) + + assert selected.fingerprints_by_identity == filtered.fingerprints_by_identity + assert selected.fingerprints == filtered.fingerprints + assert frozenset(selected.fingerprints_by_identity or {}) == test_case.expected_identities + assert {fingerprint.run_id for fingerprint in selected.fingerprints.values()} <= {"run_003"} diff --git a/tests/unit/src/sqlbuild/compiler/pipeline/main/_test_types.py b/tests/unit/src/sqlbuild/compiler/pipeline/main/_test_types.py index 898d3db9ab..1fb2794b64 100644 --- a/tests/unit/src/sqlbuild/compiler/pipeline/main/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/pipeline/main/_test_types.py @@ -40,3 +40,5 @@ class PipelinePhaseTimingTestCase: expected_compile_seconds: float expected_planning_seconds: float expected_total_seconds: float + report_progress: bool = False + expected_progress_fragments: tuple[str, ...] = () diff --git a/tests/unit/src/sqlbuild/compiler/pipeline/main/test_compile_phase_timings.py b/tests/unit/src/sqlbuild/compiler/pipeline/main/test_compile_phase_timings.py index cfb4afec0a..98da979be6 100644 --- a/tests/unit/src/sqlbuild/compiler/pipeline/main/test_compile_phase_timings.py +++ b/tests/unit/src/sqlbuild/compiler/pipeline/main/test_compile_phase_timings.py @@ -14,6 +14,7 @@ from sqlbuild.diagnostics.classes.build_phase_timing_tracker import BuildPhaseTimingTracker from sqlbuild.diagnostics.models import PartialBuildPhaseTimings from sqlbuild.rule_engine.models import RulesConfig, RulesRunResult +from sqlbuild.runtime.contracts.models import ConnectionHooks from tests.unit.src.sqlbuild.compiler.pipeline.main._test_types import PipelinePhaseTimingTestCase @@ -22,11 +23,23 @@ [ PipelinePhaseTimingTestCase( description="slow planning connection is excluded from direct compilation", - clock_values=(0.0, 0.0, 2.0, 2.0, 12.0, 12.0), + clock_values=(0.0, 0.0, 2.0, 2.0, 3.0, 13.0, 13.0), expected_compile_seconds=2.0, expected_planning_seconds=10.0, - expected_total_seconds=12.0, - ) + expected_total_seconds=13.0, + ), + PipelinePhaseTimingTestCase( + description="rules wall time is reported and excluded from compile and planning", + clock_values=(0.0, 0.0, 2.0, 2.0, 3.0, 3.0, 13.0, 13.0), + expected_compile_seconds=2.0, + expected_planning_seconds=10.0, + expected_total_seconds=13.0, + report_progress=True, + expected_progress_fragments=( + "Compiled project. (2.00s)", + "Evaluated rules. (1.00s; built-in 0.00s, custom 0.00s)", + ), + ), ], ids=lambda case: case.description, ) @@ -65,18 +78,25 @@ def test_given_slow_planning_connection_when_compiling_then_direct_phases_are_di ), ) tracker: BuildPhaseTimingTracker = BuildPhaseTimingTracker(monotonic=monotonic) + progress: list[str] = [] + progress_hooks: dict[bool, ConnectionHooks] = { + True: ConnectionHooks(on_progress=progress.append), + False: ConnectionHooks(), + } with tracker.scope(): result: CompilePipelineResult = compile_module.run_compile_pipeline( discovered_inputs=Mock(), adapter=adapter, options=CompilePipelineOptions(connection_config={}), + hooks=progress_hooks[test_case.report_progress], ) timings: PartialBuildPhaseTimings = tracker.snapshot() assert result.compile_seconds == test_case.expected_compile_seconds assert result.planning_seconds == test_case.expected_planning_seconds assert timings.total_seconds == test_case.expected_total_seconds - assert (result.compile_seconds or 0) + (result.planning_seconds or 0) <= ( + assert (result.compile_seconds or 0) + (result.planning_seconds or 0) < ( timings.total_seconds or 0 ) + assert set(test_case.expected_progress_fragments) <= set(progress) diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/_test_types.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/_test_types.py index 4428fb9bb9..61f81e3fba 100644 --- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/_test_types.py @@ -239,6 +239,7 @@ class GraphIdentityWritePerfTestCase: description: str layer_count: int expected_max_seconds: float + layer_width: int = 2 @dataclass(frozen=True) diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/helpers.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/helpers.py index 9bbd8e628f..78881f8001 100644 --- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/helpers.py +++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/helpers.py @@ -193,6 +193,30 @@ def node_key(name: str) -> GraphNodeKey: return nodes, tuple(order) +def build_wide_layered_identity_nodes( + *, layer_count: int, layer_width: int +) -> tuple[dict[GraphNodeKey, GraphIdentityNode], tuple[GraphNodeKey, ...]]: + """Build a wide project graph where each node reads two nodes of the prior layer.""" + + nodes: dict[GraphNodeKey, GraphIdentityNode] = {} + previous: tuple[GraphNodeKey, ...] = () + layer: int + for layer in range(layer_count): + current: list[GraphNodeKey] = [] + column: int + for column in range(layer_width): + key: GraphNodeKey = GraphNodeKey(node_type="model", node_name=f"L{layer}_{column}") + nodes[key] = GraphIdentityNode( + key=key, + resource_kind=GraphResourceKind.MODEL, + upstream_keys=previous[column : column + 2], + local_hash=f"local_{key.node_name}", + ) + current.append(key) + previous = tuple(current) + return nodes, tuple(nodes) + + def _append_relation_info(listed: list[RelationInfo], relation: RelationInfo) -> None: listed.append(relation) diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_graph_identity.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_graph_identity.py index 0dd6420da2..e1d04af0ef 100644 --- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_graph_identity.py +++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_graph_identity.py @@ -19,6 +19,7 @@ ) from tests.unit.src.sqlbuild.compiler.planner._helpers.helpers import ( build_diamond_ladder_identity_nodes, + build_wide_layered_identity_nodes, compose_hashed_identity, compose_readable_identity, ) @@ -190,3 +191,69 @@ def test_given_fully_selected_dense_graph_when_building_write_hashes_then_stays_ assert len(result) == len(nodes) assert elapsed < test_case.expected_max_seconds + + +@pytest.mark.parametrize( + "test_case", + [ + GraphIdentityWritePerfTestCase( + description="twenty thousand node project graph resolves in linear time", + layer_count=40, + layer_width=500, + expected_max_seconds=1.5, + ) + ], + ids=lambda case: case.description, +) +def test_given_wide_project_graph_when_building_expected_hashes_then_stays_linear( + test_case: GraphIdentityWritePerfTestCase, +) -> None: + nodes: dict[GraphNodeKey, GraphIdentityNode] + execution_order: tuple[GraphNodeKey, ...] + nodes, execution_order = build_wide_layered_identity_nodes( + layer_count=test_case.layer_count, layer_width=test_case.layer_width + ) + + start: float = time.monotonic() + result: dict[GraphNodeKey, str | None] = build_expected_graph_identity_hashes( + nodes=nodes, execution_order=execution_order, compose_identity=compose_hashed_identity + ) + elapsed: float = time.monotonic() - start + + assert len(result) == len(nodes) + assert elapsed < test_case.expected_max_seconds + + +@pytest.mark.parametrize( + "test_case", + [ + GraphIdentityWritePerfTestCase( + description="twenty thousand node project graph resolves in linear time", + layer_count=40, + layer_width=500, + expected_max_seconds=1.5, + ) + ], + ids=lambda case: case.description, +) +def test_given_wide_selected_project_graph_when_building_write_hashes_then_stays_linear( + test_case: GraphIdentityWritePerfTestCase, +) -> None: + nodes: dict[GraphNodeKey, GraphIdentityNode] + execution_order: tuple[GraphNodeKey, ...] + nodes, execution_order = build_wide_layered_identity_nodes( + layer_count=test_case.layer_count, layer_width=test_case.layer_width + ) + + start: float = time.monotonic() + result: dict[GraphNodeKey, str] = build_graph_write_identity_hashes( + nodes=nodes, + execution_order=execution_order, + selected_keys=frozenset(nodes.keys()), + base_identity_hashes={}, + compose_identity=compose_hashed_identity, + ) + elapsed: float = time.monotonic() - start + + assert len(result) == len(nodes) + assert elapsed < test_case.expected_max_seconds diff --git a/tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/_test_types.py b/tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/_test_types.py new file mode 100644 index 0000000000..928e4a1882 --- /dev/null +++ b/tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/_test_types.py @@ -0,0 +1,14 @@ +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class RulesProgressTestCase: + description: str + elapsed_seconds: float + built_in_ms: int + custom_ms: int + expected_message: str + skipped_type_proof_rules: tuple[str, ...] = () + note_skipped_rules: bool = False diff --git a/tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/test_render_rules_progress.py b/tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/test_render_rules_progress.py new file mode 100644 index 0000000000..0e28eb9de1 --- /dev/null +++ b/tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/test_render_rules_progress.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import pytest + +from sqlbuild.rule_engine.main.render_rules_progress import format_rules_progress +from sqlbuild.rule_engine.models import RulesRunResult +from tests.unit.src.sqlbuild.rule_engine.main.render_rules_progress._test_types import ( + RulesProgressTestCase, +) + + +@pytest.mark.parametrize( + "test_case", + ( + RulesProgressTestCase( + description="wall time leads concurrent built-in and custom rule time", + elapsed_seconds=14.5, + built_in_ms=5_600, + custom_ms=13_400, + expected_message="Evaluated rules. (14.50s; built-in 5.60s, custom 13.40s)", + ), + RulesProgressTestCase( + description="skipped type-proof rules follow the timings when requested", + elapsed_seconds=0.25, + built_in_ms=100, + custom_ms=0, + skipped_type_proof_rules=("SQBRCONTRACT105",), + note_skipped_rules=True, + expected_message=( + "Evaluated rules. (0.25s; built-in 0.10s, custom 0.00s); skipped type-proof " + "rules (SQBRCONTRACT105) because SQL analysis is disabled" + ), + ), + ), + ids=lambda case: case.description, +) +def test_given_rules_run_when_formatting_progress_then_reports_phase_wall_time_first( + test_case: RulesProgressTestCase, +) -> None: + result: RulesRunResult = RulesRunResult( + findings=(), + evaluated_models=0, + built_in_ms=test_case.built_in_ms, + custom_ms=test_case.custom_ms, + skipped_type_proof_rules=test_case.skipped_type_proof_rules, + ) + + message: str = format_rules_progress( + elapsed_seconds=test_case.elapsed_seconds, + result=result, + note_skipped_rules=test_case.note_skipped_rules, + ) + + assert message == test_case.expected_message diff --git a/tests/unit/src/sqlbuild/runtime/observability/_test_types.py b/tests/unit/src/sqlbuild/runtime/observability/_test_types.py index d272b95ae8..5105567e38 100644 --- a/tests/unit/src/sqlbuild/runtime/observability/_test_types.py +++ b/tests/unit/src/sqlbuild/runtime/observability/_test_types.py @@ -321,3 +321,56 @@ class ErrorCodePrivacyCase: code: object expected_error_code: str | None private_fragments: tuple[str, ...] + + +@dataclass(frozen=True) +class FastStatementMonitorCase: + description: str + statement_count: int + expected_monitor_threads: int + + +@dataclass(frozen=True) +class DeferredMonitorCase: + description: str + threshold_seconds: float + query_id: str + expected_submissions: tuple[str, ...] + + +@dataclass(frozen=True) +class FreezeJsonErrorCase: + description: str + payload: object + expected_error: str + + +@dataclass(frozen=True) +class InvocationSequenceCase: + description: str + sequence: object + expected_error: str = "" + expected_sequence: int | None = None + + +@dataclass(frozen=True) +class FinishedStatementRetentionCase: + description: str + statement_count: int + sql_bytes: int + expected_live_lifecycles: int + + +@dataclass(frozen=True) +class DeferredStartCancellationCase: + description: str + delay_seconds: float + expected_pending: int + expected_runs: int + + +@dataclass(frozen=True) +class FailedMonitorStartCase: + description: str + threshold_seconds: float + expected_event_types: tuple[str, ...] diff --git a/tests/unit/src/sqlbuild/runtime/observability/helpers.py b/tests/unit/src/sqlbuild/runtime/observability/helpers.py index 0a9495eed4..7a483738dc 100644 --- a/tests/unit/src/sqlbuild/runtime/observability/helpers.py +++ b/tests/unit/src/sqlbuild/runtime/observability/helpers.py @@ -1,19 +1,27 @@ """Test builders for runtime observability contracts.""" import asyncio -from collections.abc import Mapping +import gc +import threading +import time +import weakref +from collections.abc import Callable, Mapping from concurrent.futures import ThreadPoolExecutor from contextvars import Context, copy_context from datetime import UTC, datetime from threading import Lock from types import MappingProxyType +import pytest + from sqlbuild.runtime.observability._helpers.dispatcher import dispatcher_scope from sqlbuild.runtime.observability._helpers.factory import create_lifecycle_event from sqlbuild.runtime.observability._helpers.identity import invocation_scope +from sqlbuild.runtime.observability.classes.deferred_thread_starts import DeferredThreadStarts from sqlbuild.runtime.observability.classes.event_dispatcher import EventDispatcher from sqlbuild.runtime.observability.classes.microbatch_lifecycle import MicrobatchLifecycle from sqlbuild.runtime.observability.classes.statement_lifecycle import StatementLifecycle +from sqlbuild.runtime.observability.classes.statement_monitor import StatementMonitor from sqlbuild.runtime.observability.models import ( DiagnosticLog, LifecycleEvent, @@ -208,3 +216,144 @@ def diagnostic_log() -> DiagnosticLog: message="diagnostic", invocation_id="inv-1", ) + + +def count_monitor_thread_starts(*, monkeypatch: pytest.MonkeyPatch, statement_count: int) -> int: + """Run fast statements without query-ID providers and count monitor threads started.""" + + started: list[str] = [] + original_start: Callable[[threading.Thread], None] = threading.Thread.start + + def recording_start(thread: threading.Thread) -> None: + started.append(thread.name) + original_start(thread) + + monkeypatch.setattr(threading.Thread, "start", recording_start) + dispatcher: EventDispatcher + dispatcher, _ = capture_lifecycle_events() + with invocation_scope("inv-fast-statements"), dispatcher_scope(dispatcher): + for index in range(statement_count): + with StatementLifecycle(adapter="duckdb", sql=f"SELECT {index}", intent="execute"): + pass + return started.count("sqlbuild-statement-monitor") + + +def submissions_after_late_provider(*, threshold_seconds: float, query_id: str) -> tuple[str, ...]: + """Install a query-ID provider after a deferred start and return published submissions.""" + + submitted: threading.Event = threading.Event() + submissions: list[str] = [] + + def record_submission(value: str) -> None: + submissions.append(value) + submitted.set() + + monitor: StatementMonitor = StatementMonitor( + on_submitted=record_submission, + on_heartbeat=lambda elapsed, current_query_id: None, + threshold_seconds=threshold_seconds, + ) + monitor.start() + monitor.set_query_id_provider(lambda: query_id) + _ = submitted.wait(timeout=5.0) + _ = monitor.stop() + return tuple(submissions) + + +def heartbeats_after_early_stop(*, threshold_seconds: float) -> int: + """Stop a deferred monitor before its threshold and count heartbeats after the deadline.""" + + heartbeats: list[float] = [] + monitor: StatementMonitor = StatementMonitor( + on_submitted=lambda value: None, + on_heartbeat=lambda elapsed, current_query_id: heartbeats.append(elapsed), + threshold_seconds=threshold_seconds, + ) + monitor.start() + _ = monitor.stop() + time.sleep(threshold_seconds * 4) + return len(heartbeats) + + +def live_lifecycles_after_large_statements(*, statement_count: int, sql_bytes: int) -> int: + """Run statements with large SQL and count lifecycles still reachable after collection.""" + + references: list[weakref.ref[StatementLifecycle]] = [] + dispatcher: EventDispatcher + dispatcher, _ = capture_lifecycle_events() + with invocation_scope("inv-large-statements"), dispatcher_scope(dispatcher): + for index in range(statement_count): + lifecycle: StatementLifecycle = StatementLifecycle( + adapter="duckdb", sql=f"SELECT {index}, '{'x' * sql_bytes}'", intent="execute" + ) + with lifecycle: + pass + references.append(weakref.ref(lifecycle)) + del lifecycle + _ = gc.collect() + return len(references) - [reference() for reference in references].count(None) + + +def run_cancelled_deferred_start(*, delay_seconds: float) -> tuple[int, int]: + """Schedule then cancel one start and return pending starts and runs after its deadline.""" + + runs: list[int] = [] + starts: DeferredThreadStarts = DeferredThreadStarts(name="sqlbuild-test-deferred-starts") + token: int = starts.schedule(delay_seconds=delay_seconds, start=lambda: runs.append(1)) + starts.cancel(token) + time.sleep(delay_seconds * 4) + return starts.pending_count, len(runs) + + +def statement_events_when_monitor_start_fails( + *, monkeypatch: pytest.MonkeyPatch, threshold_seconds: float +) -> tuple[str, ...]: + """Fail every monitor thread start and return the statement's published event types.""" + + original_start: Callable[[threading.Thread], None] = threading.Thread.start + refusals: dict[str, Callable[[threading.Thread], None]] = { + "sqlbuild-statement-monitor": _refuse_thread_start + } + + def refusing_start(thread: threading.Thread) -> None: + refusals.get(thread.name, original_start)(thread) + + monkeypatch.setattr(threading.Thread, "start", refusing_start) + dispatcher: EventDispatcher + events: list[LifecycleEvent] + dispatcher, events = capture_lifecycle_events() + with invocation_scope("inv-refused-monitor"), dispatcher_scope(dispatcher): + with StatementLifecycle( + adapter="duckdb", + sql="CREATE TABLE orders AS SELECT 1", + intent="execute", + heartbeat_threshold_seconds=threshold_seconds, + ): + time.sleep(threshold_seconds * 5) + return tuple(event.event_type for event in events) + + +def _refuse_thread_start(thread: threading.Thread) -> None: + raise RuntimeError(f"cannot start {thread.name}") + + +def heartbeats_past_threshold(*, threshold_seconds: float) -> int: + """Run a deferred monitor past its threshold and count heartbeats before stopping it.""" + + heartbeat: threading.Event = threading.Event() + heartbeats: list[float] = [] + + def record_heartbeat(elapsed: float, query_id: str | None) -> None: + del query_id + heartbeats.append(elapsed) + heartbeat.set() + + monitor: StatementMonitor = StatementMonitor( + on_submitted=lambda value: None, + on_heartbeat=record_heartbeat, + threshold_seconds=threshold_seconds, + ) + monitor.start() + _ = heartbeat.wait(timeout=5.0) + _ = monitor.stop() + return len(heartbeats) diff --git a/tests/unit/src/sqlbuild/runtime/observability/test_models.py b/tests/unit/src/sqlbuild/runtime/observability/test_models.py index b197e7cb8c..25e67c29bf 100644 --- a/tests/unit/src/sqlbuild/runtime/observability/test_models.py +++ b/tests/unit/src/sqlbuild/runtime/observability/test_models.py @@ -1,13 +1,16 @@ """Tests for immutable observability envelope models.""" -from dataclasses import FrozenInstanceError +from dataclasses import FrozenInstanceError, replace import pytest +from sqlbuild.runtime.observability._helpers.validation import freeze_json from sqlbuild.runtime.observability.exceptions import ObservabilityValidationError from sqlbuild.runtime.observability.models import DiagnosticLog, LifecycleEvent from tests.unit.src.sqlbuild.runtime.observability._test_types import ( + FreezeJsonErrorCase, ImmutabilityCase, + InvocationSequenceCase, SchemaVersionCase, ) from tests.unit.src.sqlbuild.runtime.observability.helpers import OCCURRED_AT, lifecycle_event @@ -143,3 +146,82 @@ def test_given_malformed_schema_version_when_constructing_diagnostic_then_it_is_ source="runtime", message="working", ) + + +@pytest.mark.parametrize( + "test_case", + [ + FreezeJsonErrorCase( + description="nested non-finite number names its full path", + payload={"metadata": {"rows": [1, float("nan")]}}, + expected_error="payload.metadata.rows[1] must not contain NaN or infinity", + ), + FreezeJsonErrorCase( + description="nested non-string key names its parent path", + payload={"metadata": [{"ok": True}, {1: "orders"}]}, + expected_error="payload.metadata[1] keys must be strings", + ), + FreezeJsonErrorCase( + description="nested non-JSON value names its full path", + payload={"metadata": {"owner": object()}}, + expected_error="payload.metadata.owner contains non-JSON value of type object", + ), + ], + ids=lambda case: case.description, +) +def test_given_invalid_nested_payload_when_freezing_then_error_names_full_path( + test_case: FreezeJsonErrorCase, +) -> None: + with pytest.raises(ObservabilityValidationError) as error_info: + freeze_json(value=test_case.payload, path="payload") + + assert str(error_info.value) == test_case.expected_error + + +@pytest.mark.parametrize( + "test_case", + [ + InvocationSequenceCase( + description="negative sequence is rejected", + sequence=-1, + expected_error="invocation_sequence must be a non-negative integer excluding bool", + ), + InvocationSequenceCase( + description="boolean sequence is rejected", + sequence=True, + expected_error="invocation_sequence must be a non-negative integer excluding bool", + ), + ], + ids=lambda case: case.description, +) +def test_given_invalid_sequence_when_sequencing_event_then_it_is_rejected( + test_case: InvocationSequenceCase, +) -> None: + event: LifecycleEvent = lifecycle_event(payload={"command": "plan"}) + + with pytest.raises(ObservabilityValidationError) as error_info: + event.with_invocation_sequence(test_case.sequence) # ty: ignore[invalid-argument-type] + + assert str(error_info.value) == test_case.expected_error + + +@pytest.mark.parametrize( + "test_case", + [ + InvocationSequenceCase( + description="valid sequence replaces only the sequence", + sequence=7, + expected_sequence=7, + ), + ], + ids=lambda case: case.description, +) +def test_given_event_when_sequencing_then_only_sequence_changes( + test_case: InvocationSequenceCase, +) -> None: + event: LifecycleEvent = lifecycle_event(payload={"command": "plan"}) + + sequenced: LifecycleEvent = event.with_invocation_sequence(test_case.sequence) # ty: ignore[invalid-argument-type] + + assert sequenced.invocation_sequence == test_case.expected_sequence + assert replace(sequenced, invocation_sequence=event.invocation_sequence) == event diff --git a/tests/unit/src/sqlbuild/runtime/observability/test_statement_monitor.py b/tests/unit/src/sqlbuild/runtime/observability/test_statement_monitor.py new file mode 100644 index 0000000000..d2b07a4723 --- /dev/null +++ b/tests/unit/src/sqlbuild/runtime/observability/test_statement_monitor.py @@ -0,0 +1,171 @@ +"""Tests for statement monitoring cost and deferred monitor threads.""" + +import pytest + +from tests.unit.src.sqlbuild.runtime.observability._test_types import ( + DeferredMonitorCase, + DeferredStartCancellationCase, + FailedMonitorStartCase, + FastStatementMonitorCase, + FinishedStatementRetentionCase, +) +from tests.unit.src.sqlbuild.runtime.observability.helpers import ( + count_monitor_thread_starts, + heartbeats_after_early_stop, + heartbeats_past_threshold, + live_lifecycles_after_large_statements, + run_cancelled_deferred_start, + statement_events_when_monitor_start_fails, + submissions_after_late_provider, +) + + +@pytest.mark.parametrize( + "test_case", + [ + FastStatementMonitorCase( + description="fast statements without query IDs start no monitor thread", + statement_count=200, + expected_monitor_threads=0, + ), + ], + ids=lambda case: case.description, +) +def test_given_fast_statements_without_query_ids_when_executing_then_no_monitor_thread_starts( + test_case: FastStatementMonitorCase, monkeypatch: pytest.MonkeyPatch +) -> None: + started: int = count_monitor_thread_starts( + monkeypatch=monkeypatch, statement_count=test_case.statement_count + ) + + assert started == test_case.expected_monitor_threads + + +@pytest.mark.parametrize( + "test_case", + [ + DeferredMonitorCase( + description="provider installed after a deferred start is polled at once", + threshold_seconds=30.0, + query_id="01-query-orders", + expected_submissions=("01-query-orders",), + ), + ], + ids=lambda case: case.description, +) +def test_given_deferred_monitor_when_provider_arrives_then_query_id_is_submitted_once( + test_case: DeferredMonitorCase, +) -> None: + submissions: tuple[str, ...] = submissions_after_late_provider( + threshold_seconds=test_case.threshold_seconds, query_id=test_case.query_id + ) + + assert submissions == test_case.expected_submissions + + +@pytest.mark.parametrize( + "test_case", + [ + DeferredMonitorCase( + description="monitor stopped before its threshold never heartbeats", + threshold_seconds=0.02, + query_id="", + expected_submissions=(), + ), + ], + ids=lambda case: case.description, +) +def test_given_monitor_stopped_before_threshold_when_deadline_passes_then_no_heartbeat( + test_case: DeferredMonitorCase, +) -> None: + heartbeats: int = heartbeats_after_early_stop(threshold_seconds=test_case.threshold_seconds) + + assert heartbeats == len(test_case.expected_submissions) + + +@pytest.mark.parametrize( + "test_case", + [ + FinishedStatementRetentionCase( + description="finished statements with large SQL are not retained", + statement_count=200, + sql_bytes=100_000, + expected_live_lifecycles=0, + ), + ], + ids=lambda case: case.description, +) +def test_given_finished_statements_when_collecting_then_no_lifecycle_is_retained( + test_case: FinishedStatementRetentionCase, +) -> None: + live: int = live_lifecycles_after_large_statements( + statement_count=test_case.statement_count, sql_bytes=test_case.sql_bytes + ) + + assert live == test_case.expected_live_lifecycles + + +@pytest.mark.parametrize( + "test_case", + [ + DeferredStartCancellationCase( + description="cancelled start is forgotten and never runs", + delay_seconds=0.02, + expected_pending=0, + expected_runs=0, + ), + ], + ids=lambda case: case.description, +) +def test_given_cancelled_deferred_start_when_deadline_passes_then_it_never_runs( + test_case: DeferredStartCancellationCase, +) -> None: + pending, runs = run_cancelled_deferred_start(delay_seconds=test_case.delay_seconds) + + assert pending == test_case.expected_pending + assert runs == test_case.expected_runs + + +@pytest.mark.parametrize( + "test_case", + [ + DeferredMonitorCase( + description="statement outliving a lowered threshold still heartbeats", + threshold_seconds=0.02, + query_id="", + expected_submissions=("heartbeat",), + ), + ], + ids=lambda case: case.description, +) +def test_given_statement_past_threshold_when_deferred_start_runs_then_it_heartbeats( + test_case: DeferredMonitorCase, +) -> None: + heartbeats: int = heartbeats_past_threshold(threshold_seconds=test_case.threshold_seconds) + + assert heartbeats >= len(test_case.expected_submissions) + + +@pytest.mark.parametrize( + "test_case", + [ + FailedMonitorStartCase( + description="refused monitor thread still completes the statement once", + threshold_seconds=0.01, + expected_event_types=("statement_started", "statement_completed"), + ), + ], + ids=lambda case: case.description, +) +def test_given_monitor_thread_cannot_start_when_statement_finishes_then_it_completes( + test_case: FailedMonitorStartCase, monkeypatch: pytest.MonkeyPatch +) -> None: + event_types: tuple[str, ...] = statement_events_when_monitor_start_fails( + monkeypatch=monkeypatch, threshold_seconds=test_case.threshold_seconds + ) + + assert event_types == test_case.expected_event_types + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"])