diff --git a/datajunction-server/datajunction_server/internal/deployment/fingerprints.py b/datajunction-server/datajunction_server/internal/deployment/fingerprints.py index b78ad720c..ba4e3e055 100644 --- a/datajunction-server/datajunction_server/internal/deployment/fingerprints.py +++ b/datajunction-server/datajunction_server/internal/deployment/fingerprints.py @@ -1,4 +1,5 @@ import logging +import time from collections.abc import Callable, Iterable from heapq import heappop, heappush @@ -289,11 +290,17 @@ async def _load_external_specs( parent_cache: ParentCandidateCache, ) -> dict[str, NodeSpec]: """Transitively load names unresolved within the deployment's own specs.""" + started = time.perf_counter() + db_elapsed = conversion_elapsed = parent_elapsed = 0.0 + rounds = 0 known_names = set(known_names) external_specs: dict[str, NodeSpec] = {} frontier = sorted(set(initial_frontier) - known_names) + while frontier: + rounds += 1 known_names.update(frontier) + step_started = time.perf_counter() nodes = await Node.get_by_names( session, frontier, @@ -305,8 +312,12 @@ async def _load_external_specs( selectinload(Node.owners), ], ) + db_elapsed += time.perf_counter() - step_started + step_started = time.perf_counter() pending = [await node.to_spec(session) for node in nodes] + conversion_elapsed += time.perf_counter() - step_started external_specs.update({spec.rendered_name: spec for spec in pending}) + step_started = time.perf_counter() candidates: set[str] = set() for spec in pending: try: @@ -319,6 +330,20 @@ async def _load_external_specs( exc, ) frontier = sorted(candidates - known_names) + parent_elapsed += time.perf_counter() - step_started + + if rounds: + logger.info( + "External fingerprint specs: %d nodes in %d rounds; " + "db=%0.fms, conversion=%0.fms, parents=%0.fms, total=%0.fms", + len(external_specs), + rounds, + db_elapsed * 1000, + conversion_elapsed * 1000, + parent_elapsed * 1000, + (time.perf_counter() - started) * 1000, + ) + return external_specs @@ -389,6 +414,7 @@ def _compute_merkle_fingerprints( shared_fingerprints: dict[int, SemanticFingerprintValue] | None = None, ) -> FingerprintMap: """`shared_fingerprints` is a cache keyed by id(spec), shared across snapshots.""" + started = time.perf_counter() parent_cache = parent_cache if parent_cache is not None else {} shared_fingerprints = shared_fingerprints if shared_fingerprints is not None else {} names_to_process = sorted(specs) if only_names is None else sorted(only_names) @@ -433,6 +459,7 @@ def _compute_merkle_fingerprints( if name not in ignored_parse_errors: logger.warning("Fingerprint unavailable for %s: %s", name, exc) + graph_elapsed = time.perf_counter() - started components = strongly_connected_components(graph) component_by_name = { name: component_index @@ -465,6 +492,7 @@ def _compute_merkle_fingerprints( if remaining == 0: heappush(ready, component_index) + scc_elapsed = time.perf_counter() - started - graph_elapsed processed = 0 while ready: component_index = heappop(ready) @@ -555,6 +583,16 @@ def _compute_merkle_fingerprints( if processed != len(components): # pragma: no cover raise RuntimeError("SCC condensation graph contains a cycle") + logger.info( + "Semantic fingerprint graph: %d nodes, %d cached, %d components; " + "parents=%0.fms, components=%0.fms, hashing=%0.fms", + len(names_to_process), + len(cached_results), + len(components), + graph_elapsed * 1000, + scc_elapsed * 1000, + (time.perf_counter() - started - graph_elapsed - scc_elapsed) * 1000, + ) return { name: component_results[component_by_name[name]][name] for name in names_to_process @@ -637,6 +675,8 @@ async def build_deployment_fingerprints( pre_parsed_queries: dict[str, ReusableQuery] | None = None, ) -> tuple[FingerprintMap, FingerprintMap]: """`only_proposed_names` limits which nodes get a fresh proposed hash.""" + started = time.perf_counter() + timings: dict[str, float] = {} proposed_specs = list(proposed_specs) additional_target_names = set(additional_target_names) deleted_names = {spec.rendered_name for spec in deleted_specs} @@ -652,11 +692,13 @@ async def build_deployment_fingerprints( deleted_names, unchanged_names, ) + timings["setup"] = time.perf_counter() - started target_specs: dict[str, NodeSpec] = {} target_names_to_load = ( additional_target_names - existing_specs.keys() - submitted_names ) if target_names_to_load: + step_started = time.perf_counter() target_nodes = await Node.get_by_names( session, sorted(target_names_to_load), @@ -668,11 +710,15 @@ async def build_deployment_fingerprints( selectinload(Node.owners), ], ) + timings["target_db"] = time.perf_counter() - step_started + step_started = time.perf_counter() target_specs = { spec.rendered_name: spec for spec in [await node.to_spec(session) for node in target_nodes] } + timings["target_to_spec"] = time.perf_counter() - step_started + step_started = time.perf_counter() parent_cache: ParentCandidateCache = {} # Reuse ASTs propagate_impact already parsed. current_seed_specs = {**existing_specs, **target_specs} @@ -693,6 +739,8 @@ async def build_deployment_fingerprints( proposed, parent_cache, ) + timings["ancestor_closure"] = time.perf_counter() - step_started + step_started = time.perf_counter() external = await _load_external_specs( session, current_unresolved | proposed_unresolved, @@ -700,13 +748,10 @@ async def build_deployment_fingerprints( ignored_parse_errors=deleted_names, parent_cache=parent_cache, ) - # target_specs need a separate copy per side so the id()-keyed - # shared_fingerprints cache can't cross-reuse a stale ancestor chain. + timings["external_load"] = time.perf_counter() - step_started + step_started = time.perf_counter() current_external = {**external, **target_specs} - proposed_external = { - **external, - **{name: spec.model_copy(deep=True) for name, spec in target_specs.items()}, - } + proposed_external = {**external, **target_specs} shared_fingerprints: dict[int, SemanticFingerprintValue] = {} current_graph = SemanticFingerprintGraph( {**current_external, **existing_specs}, @@ -721,9 +766,30 @@ async def build_deployment_fingerprints( version=version, shared_fingerprints=shared_fingerprints, ) + timings["snapshot_setup"] = time.perf_counter() - step_started + step_started = time.perf_counter() # unchanged_names gives no-op nodes a current fingerprint to fall back on. current = current_graph.fingerprints( deleted_names | additional_target_names | unchanged_names, ) + timings["current_hashes"] = time.perf_counter() - step_started + # The shared cache is keyed only by spec identity. A target can keep the + # same spec object but have a new hash when its proposed ancestors change, + # so evict its current hash before evaluating the proposed graph. + for spec in target_specs.values(): + shared_fingerprints.pop(id(spec), None) + step_started = time.perf_counter() proposed_hashes = proposed_graph.fingerprints(proposed_target_names) + timings["proposed_hashes"] = time.perf_counter() - step_started + logger.info( + "Deployment fingerprints: %d targets (%d loaded), %d external ancestors; " + "total=%.0fms, %s", + len(additional_target_names), + len(target_specs), + len(external), + (time.perf_counter() - started) * 1000, + ", ".join( + f"{name}={elapsed * 1000:.0f}ms" for name, elapsed in timings.items() + ), + ) return current, proposed_hashes diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/v1.py b/datajunction-server/datajunction_server/semantic_fingerprints/v1.py index 58ddcdc9b..1704e533b 100644 --- a/datajunction-server/datajunction_server/semantic_fingerprints/v1.py +++ b/datajunction-server/datajunction_server/semantic_fingerprints/v1.py @@ -69,6 +69,12 @@ def build_fingerprint( """Build a version 1 semantic fingerprint.""" fingerprint_fields = semantic_fields(type(spec)) rendered = spec.rendered_spec() + # rendered_spec() reconstructs the model and drops private attributes. + # Both specs have the same rendered query, so reuse its parsed AST across + # current and proposed fingerprints instead of parsing each copy again. + if rendered.rendered_query is not None: + rendered._query_ast = spec.query_ast + fields = { field: normalize_field( rendered, diff --git a/datajunction-server/tests/internal/deployment/fingerprints_test.py b/datajunction-server/tests/internal/deployment/fingerprints_test.py index 65a254d72..74dca8cb3 100644 --- a/datajunction-server/tests/internal/deployment/fingerprints_test.py +++ b/datajunction-server/tests/internal/deployment/fingerprints_test.py @@ -30,7 +30,11 @@ UNKNOWN_SEMANTIC_FINGERPRINT, SemanticFingerprint, ) -from datajunction_server.semantic_fingerprints.engine import local_node_fingerprint +from datajunction_server.semantic_fingerprints.engine import ( + compose_node_fingerprint, + local_node_fingerprint, +) +from datajunction_server.sql.parsing.backends.antlr4 import parse def source_spec( @@ -594,6 +598,42 @@ async def test_build_deployment_fingerprints_only_proposed_names_reuses_unchange assert scoped == {downstream.rendered_name: full[downstream.rendered_name]} +@pytest.mark.asyncio +async def test_external_target_is_rehashed_when_its_ancestor_changes(): + """An external target needs separate current and proposed graph hashes.""" + current_source = source_spec("source", table="before") + proposed_source = source_spec("source", table="after") + downstream = transform_spec( + "downstream", + "SELECT * FROM ns.source", + namespace="other", + ) + node = MagicMock() + node.to_spec = AsyncMock(return_value=downstream) + + with patch( + "datajunction_server.internal.deployment.fingerprints.Node.get_by_names", + AsyncMock(return_value=[node]), + ): + current, proposed = await build_deployment_fingerprints( + MagicMock(), + {current_source.rendered_name: current_source}, + [proposed_source], + [], + additional_target_names=[downstream.rendered_name], + ) + + expected_current = SemanticFingerprintGraph( + spec_map(current_source, downstream), + ).fingerprint(downstream.rendered_name) + expected_proposed = SemanticFingerprintGraph( + spec_map(proposed_source, downstream), + ).fingerprint(downstream.rendered_name) + assert current[downstream.rendered_name] == expected_current + assert proposed[downstream.rendered_name] == expected_proposed + assert expected_current != expected_proposed + + @pytest.mark.asyncio async def test_build_deployment_fingerprints_without_external_parents(): source = source_spec("source", table="table") @@ -808,3 +848,21 @@ def test_semantic_fingerprint_graph_caches_whole_graph_result(): # A subsequent scoped call also reuses the already-cached result. scoped = graph.fingerprints({transform.rendered_name}) assert scoped[transform.rendered_name] == first[transform.rendered_name] + + +def test_fingerprinting_reuses_parsed_query_across_rendered_copies(): + metric = MetricSpec( + namespace="ns", + name="metric", + query="SELECT COUNT(*) FROM ${prefix}source", + ) + + with patch( + "datajunction_server.sql.parsing.backends.antlr4.parse", + wraps=parse, + ) as parse_query: + first = compose_node_fingerprint(metric, parent_fingerprints=[]) + second = compose_node_fingerprint(metric, parent_fingerprints=[]) + + assert first == second + assert parse_query.call_count == 1