diff --git a/datajunction-clients/python/tests/test_deploy.py b/datajunction-clients/python/tests/test_deploy.py index f45c68890..0f1fa3b74 100644 --- a/datajunction-clients/python/tests/test_deploy.py +++ b/datajunction-clients/python/tests/test_deploy.py @@ -53,7 +53,6 @@ def test_reconstruct_deployment_spec_separates_preaggregations(tmp_path): ) (tmp_path / "revenue_by_day.yaml").write_text( "kind: preagg\n" - "name: revenue_by_day\n" "metrics:\n - ns.revenue\n" "dimensions:\n - ns.date.day\n" "catalog: default\n" @@ -69,7 +68,7 @@ def test_reconstruct_deployment_spec_separates_preaggregations(tmp_path): assert [node["name"] for node in spec["nodes"]] == ["ns.revenue"] assert len(spec["preaggregations"]) == 1 preagg = spec["preaggregations"][0] - assert preagg["name"] == "revenue_by_day" + assert preagg["table"] == "revenue_agg" assert preagg["measure_columns"] == {"ns.revenue": "revenue_sum"} # The discriminator is stripped before reaching the payload. assert "kind" not in preagg diff --git a/datajunction-server/datajunction_server/api/preaggregations.py b/datajunction-server/datajunction_server/api/preaggregations.py index a70f17d8d..60a000183 100644 --- a/datajunction-server/datajunction_server/api/preaggregations.py +++ b/datajunction-server/datajunction_server/api/preaggregations.py @@ -771,7 +771,6 @@ async def register_preaggregations( session, query_service_client, request_headers, - name=data.name, metrics=data.metrics, dimensions=data.dimensions, table=data.table, diff --git a/datajunction-server/datajunction_server/database/preaggregation.py b/datajunction-server/datajunction_server/database/preaggregation.py index 8cd15df6f..ecfdda73b 100644 --- a/datajunction-server/datajunction_server/database/preaggregation.py +++ b/datajunction-server/datajunction_server/database/preaggregation.py @@ -264,8 +264,12 @@ class PreAggregation(Base): unique=True, ) - # Optional stable, human-supplied handle for externally-registered pre-aggs. - # Used by YAML deploy reconciliation and availability-by-name callbacks. + # Legacy human-supplied handle for externally-registered pre-aggs. Retired: + # registration no longer accepts a name and nothing reads this column. It + # was never load-bearing -- reconciliation matches on revision + grain + + # measure identity, and the availability callback is addressed by id + # (POST /preaggs/{preagg_id}/availability/). Kept nullable so rows written + # before it was retired still load. name: Mapped[str | None] = mapped_column(String, nullable=True) # === Materialization Config === diff --git a/datajunction-server/datajunction_server/internal/deployment/orchestrator.py b/datajunction-server/datajunction_server/internal/deployment/orchestrator.py index 933e659a7..52a013bf6 100644 --- a/datajunction-server/datajunction_server/internal/deployment/orchestrator.py +++ b/datajunction-server/datajunction_server/internal/deployment/orchestrator.py @@ -31,6 +31,7 @@ from datajunction_server.database.namespace import NodeNamespace from datajunction_server.database.node import MissingParent, NodeRelationship from datajunction_server.database.partition import Partition +from datajunction_server.database.preaggregation import PreAggregation from datajunction_server.database.tag import Tag from datajunction_server.database.user import OAuthProvider, User from datajunction_server.errors import ( @@ -89,6 +90,7 @@ MaterializationSpec, MetricSpec, NodeSpec, + PreAggSpec, SourceSpec, TagSpec, bump_version, @@ -127,6 +129,32 @@ logger = logging.getLogger(__name__) +def preagg_label(table_ref: str, grain_columns: list[str]) -> str: + """ + How a pre-aggregation is named in deployment results: the external table it + adopts, plus the grain it covers. + + The table alone would be ambiguous, since one registration can produce + several pre-aggregations on the same table -- one per grain group -- and the + two together are what reconciliation actually matches on. Grain columns are + sorted so the label is stable no matter how the spec ordered them. + """ + return f"{table_ref} [{', '.join(sorted(grain_columns))}]" + + +def existing_preagg_label(preagg: PreAggregation) -> str: + """ + The same label for a pre-aggregation already in the database. Its table + comes from availability, which a pre-aggregation registered without a + ``valid_through_ts`` does not have yet; those fall back to naming the node + they roll up. + """ + return preagg_label( + preagg.materialized_table_ref or preagg.node_revision.name, + preagg.grain_columns, + ) + + def _extract_dimension_refs_from_filters( filters: list[str], ) -> list[tuple[str, str]]: @@ -476,8 +504,6 @@ async def _plan_and_execute(self) -> DeploymentExecuteResult: # Pre-aggregations still need reconciling on an otherwise-empty # deploy: to register specs, or to delete external pre-aggs that were # dropped from the spec. - from datajunction_server.database.preaggregation import PreAggregation - needs_preagg_reconcile = bool( self.deployment_spec.preaggregations, ) or bool( @@ -1118,10 +1144,11 @@ async def _reconcile_preaggregations(self) -> None: remove EXTERNAL pre-aggs in this namespace that were dropped from it. Runs after nodes/links/cubes so the referenced metrics and dimensions - exist. Skipped during dry runs (registration introspects the external - table and mutates state). Reuses the same core as POST /preaggs/register. + exist. A dry run registers nothing (registration introspects the + external table and mutates state) and previews instead, see + ``_preview_preaggregations``. Reuses the same core as + POST /preaggs/register. """ - from datajunction_server.database.preaggregation import PreAggregation from datajunction_server.internal.preaggregations import ( register_external_preaggregations, ) @@ -1175,39 +1202,11 @@ async def _reconcile_preaggregations(self) -> None: ) return - declared_names = {spec.rendered_name for spec in specs} existing_ids = {preagg.id for preagg in existing} - existing_names = {preagg.name for preagg in existing} - # Dry run: report the planned register/delete without mutating anything - # (no registration, so no external-table introspection either). Delete - # planning is name-level here; the wet run below deletes by row identity. + # Dry run: report the planned register/delete without mutating anything. if self.dry_run: - for spec in specs: - self.deployed_results.append( - DeploymentResult( - name=spec.rendered_name, - deploy_type=DeploymentResult.Type.PREAGG, - status=DeploymentResult.Status.SUCCESS, - operation=( - DeploymentResult.Operation.UPDATE - if spec.rendered_name in existing_names - else DeploymentResult.Operation.CREATE - ), - message="externally-built pre-aggregation", - ), - ) - for preagg in existing: - if preagg.name not in declared_names: - self.deployed_results.append( - DeploymentResult( - name=preagg.name or f"preaggregation:{preagg.id}", - deploy_type=DeploymentResult.Type.PREAGG, - status=DeploymentResult.Status.SUCCESS, - operation=DeploymentResult.Operation.DELETE, - message="dropped from spec", - ), - ) + await self._preview_preaggregations(specs, existing) return query_service_client = self.context.query_service_client @@ -1238,7 +1237,6 @@ async def _reconcile_preaggregations(self) -> None: self.session, query_service_client, request_headers, - name=spec.rendered_name, metrics=spec.rendered_metrics, dimensions=spec.rendered_dimensions, table=ExternalPreAggTable( @@ -1254,7 +1252,10 @@ async def _reconcile_preaggregations(self) -> None: for preagg in created: self.deployed_results.append( DeploymentResult( - name=spec.rendered_name, + name=preagg_label( + spec.table_ref, + preagg.grain_columns, + ), deploy_type=DeploymentResult.Type.PREAGG, status=DeploymentResult.Status.SUCCESS, operation=( @@ -1270,7 +1271,7 @@ async def _reconcile_preaggregations(self) -> None: DJError( code=ErrorCode.INVALID_ARGUMENTS_TO_FUNCTION, message=( - f"Pre-aggregation '{spec.rendered_name}': {exc.message}" + f"Pre-aggregation on '{spec.table_ref}': {exc.message}" ), ), ) @@ -1291,7 +1292,7 @@ async def _reconcile_preaggregations(self) -> None: if preagg.id not in upserted_ids: self.deployed_results.append( DeploymentResult( - name=preagg.name or f"preaggregation:{preagg.id}", + name=existing_preagg_label(preagg), deploy_type=DeploymentResult.Type.PREAGG, status=DeploymentResult.Status.SUCCESS, operation=DeploymentResult.Operation.DELETE, @@ -1303,6 +1304,96 @@ async def _reconcile_preaggregations(self) -> None: await self.session.flush() logger.info("Reconciled %d pre-aggregation spec(s)", len(specs)) + async def _preview_preaggregations( + self, + specs: list[PreAggSpec], + existing: list[PreAggregation], + ) -> None: + """ + Report what pre-aggregation reconciliation would do, keyed exactly as the + apply keys it: the rows each spec resolves to. + + The apply only learns those rows by registering, which a dry run cannot + do -- registration introspects the external table through the query + service. Everything the row identity depends on is pure decomposition, + though, so the preview resolves each spec to its grain groups and their + matching rows itself and reports the same creates and updates the apply + will perform, rather than diffing user-supplied handles (which reported a + single update where the apply did a create plus a delete). + + A spec that cannot be resolved yet -- most often because this very deploy + is what creates its metrics -- is reported once as UNKNOWN instead of + being guessed at. Because such a spec might still have claimed an + existing row, the leftover rows are then reported as UNKNOWN too: a + preview must never promise a delete the apply will not perform. + """ + from datajunction_server.internal.preaggregations import ( + resolve_external_preagg_targets, + ) + + claimed_ids: set[int] = set() + unresolved = False + for spec in specs: + try: + targets = await resolve_external_preagg_targets( + self.session, + metrics=spec.rendered_metrics, + dimensions=spec.rendered_dimensions, + ) + except DJException as exc: + unresolved = True + self.deployed_results.append( + DeploymentResult( + name=spec.table_ref, + deploy_type=DeploymentResult.Type.PREAGG, + status=DeploymentResult.Status.SUCCESS, + operation=DeploymentResult.Operation.UNKNOWN, + message=( + f"cannot be previewed until its metrics and " + f"dimensions exist: {exc.message}" + ), + ), + ) + continue + for target in targets: + if target.existing: + claimed_ids.add(target.existing.id) + self.deployed_results.append( + DeploymentResult( + name=preagg_label(spec.table_ref, target.grain_columns), + deploy_type=DeploymentResult.Type.PREAGG, + status=DeploymentResult.Status.SUCCESS, + operation=( + DeploymentResult.Operation.UPDATE + if target.existing + else DeploymentResult.Operation.CREATE + ), + message="externally-built pre-aggregation", + ), + ) + + for preagg in existing: + if preagg.id in claimed_ids: + continue + self.deployed_results.append( + DeploymentResult( + name=existing_preagg_label(preagg), + deploy_type=DeploymentResult.Type.PREAGG, + status=DeploymentResult.Status.SUCCESS, + operation=( + DeploymentResult.Operation.UNKNOWN + if unresolved + else DeploymentResult.Operation.DELETE + ), + message=( + "may be dropped from spec, depending on the " + "pre-aggregation(s) that could not be previewed" + if unresolved + else "dropped from spec" + ), + ), + ) + async def _setup_owners(self): """ Validate that all owners defined in the deployment spec exist. diff --git a/datajunction-server/datajunction_server/internal/preaggregations.py b/datajunction-server/datajunction_server/internal/preaggregations.py index b3b6d351e..95041bef9 100644 --- a/datajunction-server/datajunction_server/internal/preaggregations.py +++ b/datajunction-server/datajunction_server/internal/preaggregations.py @@ -8,6 +8,7 @@ direction api -> internal. """ +from dataclasses import dataclass from typing import cast from sqlalchemy.ext.asyncio import AsyncSession @@ -28,6 +29,7 @@ compute_grain_group_hash, compute_preagg_hash, get_measure_identities, + measure_identity_token, ) from datajunction_server.errors import DJInvalidInputException from datajunction_server.models.decompose import PreAggMeasure @@ -120,7 +122,6 @@ async def register_external_preaggregations( query_service_client: QueryServiceClient, request_headers: dict[str, str], *, - name: str | None, metrics: list[str], dimensions: list[str], table: ExternalPreAggTable, @@ -318,7 +319,6 @@ async def register_external_preaggregations( existing.columns = columns existing.sql = grain_group.sql existing.strategy = MaterializationStrategy.EXTERNAL - existing.name = name preagg = existing else: preagg = PreAggregation( @@ -334,7 +334,6 @@ async def register_external_preaggregations( grain_measures, ), strategy=MaterializationStrategy.EXTERNAL, - name=name, ) session.add(preagg) created_preaggs.append(preagg) @@ -354,3 +353,66 @@ async def register_external_preaggregations( await session.flush() return created_preaggs + + +@dataclass +class PreAggTarget: + """ + One grain group a pre-aggregation spec resolves to, together with the + existing row (if any) that registering it would land on. + """ + + grain_columns: list[str] + existing: PreAggregation | None + + +async def resolve_external_preagg_targets( + session: AsyncSession, + *, + metrics: list[str], + dimensions: list[str], +) -> list[PreAggTarget]: + """ + Work out which pre-aggregation rows a spec maps to, without introspecting + the external table. + + ``register_external_preaggregations`` only discovers this by registering: it + decomposes the metrics into grain groups and, for each, looks for a row that + already covers the same measures at the same grain. Neither step needs the + query service -- only binding the physical columns does -- so a deploy dry + run can reproduce the lookup exactly and preview the very operations the + apply will perform, instead of guessing from a user-supplied handle. + """ + measures_result = await build_measures_sql( + session=session, + metrics=metrics, + dimensions=dimensions, + dialect=Dialect.SPARK, + use_materialized=False, + ) + assert_dimension_refs_are_role_qualified(measures_result, dimensions) + + targets: list[PreAggTarget] = [] + for grain_group in measures_result.grain_groups: + parent_node = measures_result.ctx.nodes.get(grain_group.parent_name) + if not parent_node or not parent_node.current: # pragma: no cover + continue + grain_columns = list(measures_result.requested_dimensions) + targets.append( + PreAggTarget( + grain_columns=grain_columns, + existing=await PreAggregation.find_matching( + session=session, + node_revision_id=parent_node.current.id, + grain_columns=grain_columns, + measure_identities={ + measure_identity_token( + compute_expression_hash(component.expression), + component.normalized_aggregation, + ) + for component in grain_group.components + }, + ), + ), + ) + return targets diff --git a/datajunction-server/datajunction_server/models/deployment.py b/datajunction-server/datajunction_server/models/deployment.py index 4ac01c096..63ed18514 100644 --- a/datajunction-server/datajunction_server/models/deployment.py +++ b/datajunction-server/datajunction_server/models/deployment.py @@ -37,6 +37,7 @@ NodeType, ) from datajunction_server.models.partition import Granularity, PartitionType +from datajunction_server.models.preaggregation import reject_preagg_name from datajunction_server.models.unit import ( Unit, legacy_unit_to_structured, @@ -174,15 +175,19 @@ class HierarchySpec(NamespacedSpec): levels: list[HierarchyLevelSpec] = Field(default_factory=list, min_length=2) -class PreAggSpec(NamespacedSpec): +class PreAggSpec(BaseModel): """ Specification for an externally-built pre-aggregation table adopted at deploy - time (equivalent to POST /preaggs/register). ``name`` is a stable handle used - for reconciliation and availability callbacks. Metric/dimension references may - use ``${prefix}`` or be fully qualified; they are rendered against the + time (equivalent to POST /preaggs/register). It carries no name: a + pre-aggregation is identified by the table it is bound to and the grain it + covers, which is what reconciliation matches on. Metric/dimension references + may use ``${prefix}`` or be fully qualified; they are rendered against the deployment namespace. """ + # Not user-supplied, gets injected by DeploymentSpec.set_namespaces + namespace: str | None = Field(default=None, exclude=True) + metrics: list[str] = Field(default_factory=list) dimensions: list[str] = Field(default_factory=list) catalog: str @@ -196,6 +201,17 @@ class PreAggSpec(NamespacedSpec): model_config = ConfigDict(populate_by_name=True) + @model_validator(mode="before") + @classmethod + def reject_name(cls, values: Any) -> Any: + """Refuse the retired ``name`` handle (see ``reject_preagg_name``).""" + return reject_preagg_name(values) + + @property + def table_ref(self) -> str: + """Fully qualified reference to the external table this spec adopts.""" + return f"{self.catalog}.{self.schema_}.{self.table}" + @property def rendered_metrics(self) -> list[str]: return [render_prefixes(metric, self.namespace) for metric in self.metrics] diff --git a/datajunction-server/datajunction_server/models/preaggregation.py b/datajunction-server/datajunction_server/models/preaggregation.py index b6454a232..e484d3673 100644 --- a/datajunction-server/datajunction_server/models/preaggregation.py +++ b/datajunction-server/datajunction_server/models/preaggregation.py @@ -5,7 +5,7 @@ from datetime import date, datetime from typing import Any -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel, Field, field_validator, model_validator from datajunction_server.enum import StrEnum from datajunction_server.models.decompose import PreAggMeasure @@ -16,6 +16,27 @@ from datajunction_server.models.query import ColumnMetadata, V3ColumnMetadata +#: Raised when a caller still supplies the retired ``name`` handle on an +#: externally-registered pre-aggregation. Shared by the request model and the +#: deployment spec so both paths say the same thing. +PREAGG_NAME_REMOVED_ERROR = ( + "Pre-aggregations no longer take a `name`. DJ identifies an " + "externally-registered pre-aggregation by the table it is bound to and the " + "grain it covers, and reconciles a deploy against those, so the handle was " + "never read. Remove `name` from the pre-aggregation." +) + + +def reject_preagg_name(values: Any) -> Any: + """ + Refuse a supplied ``name`` outright rather than dropping it silently, so a + spec that still carries the retired handle fails loudly at the edge. + """ + if isinstance(values, dict) and "name" in values: + raise ValueError(PREAGG_NAME_REMOVED_ERROR) + return values + + class WorkflowStatus(StrEnum): """Status of a pre-aggregation workflow.""" @@ -117,13 +138,6 @@ class RegisterPreAggregationsRequest(BaseModel): pre-aggregation so grain resolution can route queries to it. """ - name: str | None = Field( - default=None, - description=( - "Optional stable handle for the pre-aggregation, used by YAML deploy " - "reconciliation and availability-by-name callbacks." - ), - ) metrics: list[str] = Field( description="Metric node names the external table should serve", ) @@ -148,6 +162,12 @@ class RegisterPreAggregationsRequest(BaseModel): ), ) + @model_validator(mode="before") + @classmethod + def reject_name(cls, values: Any) -> Any: + """Refuse the retired ``name`` handle (see ``reject_preagg_name``).""" + return reject_preagg_name(values) + class UpdatePreAggregationAvailabilityRequest(BaseModel): """Request model for updating pre-aggregation availability.""" @@ -222,7 +242,9 @@ class PreAggregationInfo(BaseModel): sql: str # The generated SQL for materializing this pre-agg grain_group_hash: str preagg_hash: str # Unique hash including measures (used for table/workflow naming) - name: str | None = None # Stable handle for externally-registered pre-aggs + # Legacy handle carried by pre-aggs registered before the name was retired. + # Nothing reads it; new registrations leave it unset. + name: str | None = None # Materialization config strategy: MaterializationStrategy | None = None diff --git a/datajunction-server/tests/api/deployments_test.py b/datajunction-server/tests/api/deployments_test.py index 9dbc90d7b..fcb7ab084 100644 --- a/datajunction-server/tests/api/deployments_test.py +++ b/datajunction-server/tests/api/deployments_test.py @@ -17,6 +17,7 @@ ) from datajunction_server.database.materialization import Materialization from datajunction_server.database.node import Node, NodeRelationship +from datajunction_server.database.preaggregation import PreAggregation from datajunction_server.database.tag import Tag from datajunction_server.errors import DJInvalidInputException from datajunction_server.internal.git.github_service import GitHubServiceError @@ -1849,7 +1850,6 @@ async def _fake_columns(*args, **kwargs): fact_node = "preagg_deploy.default.hard_hat_facts" try: preagg_spec = PreAggSpec( - name="hard_hats_by_state", metrics=["${prefix}default.count_hard_hats"], dimensions=["${prefix}default.us_state.state_name"], catalog="default", @@ -1870,13 +1870,33 @@ async def _fake_columns(*args, **kwargs): ) assert data["status"] == "success", data["results"] + # The deploy result labels the pre-agg by what identifies it: the + # external table it adopts, plus the grain it covers. + assert [ + result + for result in data["results"] + if result["deploy_type"] == "preaggregation" + ] == [ + { + "name": "default.analytics.hard_hats_agg " + "[preagg_deploy.default.us_state.state_name]", + "deploy_type": "preaggregation", + "status": "success", + "operation": "create", + "message": "externally-built pre-aggregation", + "changed_fields": [], + }, + ] + # The external pre-agg was registered against the fact node. listing = await client.get("/preaggs/", params={"node_name": fact_node}) assert listing.status_code == 200, listing.text items = listing.json()["items"] assert len(items) == 1 preagg = items[0] - assert preagg["name"] == "preagg_deploy.hard_hats_by_state" + # Registration no longer records a handle -- the row is identified + # by the table it is bound to and the grain it covers. + assert preagg["name"] is None assert preagg["strategy"] == "external" assert any( measure["source_column"] == "hard_hat_count" @@ -1956,7 +1976,6 @@ async def _fake_columns(*args, **kwargs): client.app.dependency_overrides[get_query_service_client] = lambda: mock_qs try: preagg = PreAggSpec( - name="dimcol_agg", metrics=["${prefix}default.dimcol_count"], dimensions=[ "${prefix}default.us_state.state_short", @@ -7669,9 +7688,8 @@ def _clear_query_service(client): client.app.dependency_overrides.pop(get_query_service_client, None) -def _preagg_spec(name, table, measure_columns, dimensions=None, dimension_columns=None): +def _preagg_spec(table, measure_columns, dimensions=None, dimension_columns=None): return PreAggSpec( - name=name, metrics=["${prefix}default.count_hard_hats"], dimensions=dimensions or ["${prefix}default.us_state.state_name"], catalog="default", @@ -7699,7 +7717,6 @@ async def test_deploy_preagg_invalid_metric_fails( _override_query_service(client, ["hard_hat_count", "state_name"]) try: bad = _preagg_spec( - "bad_preagg", "hh_bad", {"${prefix}default.does_not_exist": "hard_hat_count"}, ) @@ -7738,7 +7755,6 @@ async def test_deploy_preagg_without_query_service( client.app.dependency_overrides[get_query_service_client] = lambda: None try: spec = _preagg_spec( - "needs_qs", "hh_qs", {"${prefix}default.count_hard_hats": "hard_hat_count"}, ) @@ -7788,7 +7804,6 @@ async def test_dry_run_reports_preagg_reconcile( assert data["status"] == "success", data["results"] spec = _preagg_spec( - "dry_preagg", "hh_dry", {"${prefix}default.count_hard_hats": "hard_hat_count"}, ) @@ -7811,9 +7826,17 @@ async def _impact_preagg_results(preaggs, allow_empty=False): # Not yet registered -> planned CREATE, nothing persisted. results = await _impact_preagg_results([spec]) - assert len(results) == 1 - assert results[0]["name"] == "preagg_dry.dry_preagg" - assert results[0]["operation"] == "create" + assert results == [ + { + "name": "default.analytics.hh_dry " + "[preagg_dry.default.us_state.state_name]", + "deploy_type": "preaggregation", + "status": "success", + "operation": "create", + "message": "externally-built pre-aggregation", + "changed_fields": [], + }, + ] listing = await client.get( "/preaggs/", params={"node_name": "preagg_dry.default.hard_hat_facts"}, @@ -7831,14 +7854,33 @@ async def _impact_preagg_results(preaggs, allow_empty=False): ) assert data["status"] == "success", data["results"] - # Re-declaring it -> planned UPDATE. + # Re-declaring it -> planned UPDATE, under the same label. results = await _impact_preagg_results([spec]) - assert [r["operation"] for r in results] == ["update"] + assert results == [ + { + "name": "default.analytics.hh_dry " + "[preagg_dry.default.us_state.state_name]", + "deploy_type": "preaggregation", + "status": "success", + "operation": "update", + "message": "externally-built pre-aggregation", + "changed_fields": [], + }, + ] # Dropping it (allow_empty) -> planned DELETE, still there afterward. results = await _impact_preagg_results([], allow_empty=True) - assert len(results) == 1 - assert results[0]["operation"] == "delete" + assert results == [ + { + "name": "default.analytics.hh_dry " + "[preagg_dry.default.us_state.state_name]", + "deploy_type": "preaggregation", + "status": "success", + "operation": "delete", + "message": "dropped from spec", + "changed_fields": [], + }, + ] listing = await client.get( "/preaggs/", params={"node_name": "preagg_dry.default.hard_hat_facts"}, @@ -7847,6 +7889,238 @@ async def _impact_preagg_results(preaggs, allow_empty=False): finally: _clear_query_service(client) + @pytest.mark.asyncio + async def test_dry_run_matches_apply_when_the_grain_changes( + self, + client, + default_hard_hats, + default_us_states, + default_us_state, + ): + """Widening a pre-aggregation's grain is a create plus a delete, because + the old row no longer covers the requested grain. The dry run resolves + each spec to the rows it lands on -- the same key the apply reconciles + by -- so preview and apply now report exactly the same operations.""" + _override_query_service(client, ["hard_hat_count", "state_name"]) + try: + nodes = _hard_hat_deploy_nodes( + default_hard_hats, + default_us_states, + default_us_state, + ) + measure_columns = {"${prefix}default.count_hard_hats": "hard_hat_count"} + by_state = _preagg_spec("hh_grain", measure_columns) + data = await deploy_and_wait( + client, + DeploymentSpec( + namespace="preagg_grain", + nodes=nodes, + preaggregations=[by_state], + ), + ) + assert data["status"] == "success", data["results"] + + # Same table, new grain: the pre-agg is rebuilt per hard hat too. + by_state_and_hat = _preagg_spec( + "hh_grain", + measure_columns, + dimensions=[ + "${prefix}default.us_state.state_name", + "${prefix}default.hard_hat_facts.hard_hat_id", + ], + ) + regrained = DeploymentSpec( + namespace="preagg_grain", + nodes=nodes, + preaggregations=[by_state_and_hat], + ) + impact = await client.post( + "/deployments/impact", + json=regrained.model_dump(), + ) + previewed = [ + (result["name"], result["operation"]) + for result in impact.json()["results"] + if result["deploy_type"] == "preaggregation" + ] + data = await deploy_and_wait(client, regrained) + assert data["status"] == "success", data["results"] + applied = [ + (result["name"], result["operation"]) + for result in data["results"] + if result["deploy_type"] == "preaggregation" + ] + + assert previewed == [ + ( + "default.analytics.hh_grain " + "[preagg_grain.default.hard_hat_facts.hard_hat_id, " + "preagg_grain.default.us_state.state_name]", + "create", + ), + ( + "default.analytics.hh_grain " + "[preagg_grain.default.us_state.state_name]", + "delete", + ), + ] + assert applied == previewed + finally: + _clear_query_service(client) + + @pytest.mark.asyncio + async def test_dry_run_is_coarse_when_a_spec_cannot_be_resolved( + self, + client, + default_hard_hats, + default_us_states, + default_us_state, + ): + """A first deploy is also what creates the metrics a pre-agg refers to, + so the dry run cannot resolve the spec to rows yet. It says so, once, + instead of guessing -- and, since the unresolved spec might still have + claimed the pre-agg already registered here, that one is reported as + uncertain rather than promised for deletion.""" + _override_query_service(client, ["hard_hat_count", "state_name"]) + try: + nodes = _hard_hat_deploy_nodes( + default_hard_hats, + default_us_states, + default_us_state, + ) + measure_columns = {"${prefix}default.count_hard_hats": "hard_hat_count"} + data = await deploy_and_wait( + client, + DeploymentSpec( + namespace="preagg_unres", + nodes=nodes, + preaggregations=[_preagg_spec("hh_unres", measure_columns)], + ), + ) + assert data["status"] == "success", data["results"] + + # A pre-agg on metrics this deployment has not created (and this dry + # run will not create) cannot be resolved to rows. + unresolvable = PreAggSpec( + metrics=["${prefix}default.not_yet_deployed"], + dimensions=["${prefix}default.us_state.state_name"], + catalog="default", + schema="analytics", + table="hh_future", + measure_columns={ + "${prefix}default.not_yet_deployed": "hard_hat_count", + }, + ) + impact = await client.post( + "/deployments/impact", + json=DeploymentSpec( + namespace="preagg_unres", + nodes=nodes, + preaggregations=[unresolvable], + ).model_dump(), + ) + results = [ + result + for result in impact.json()["results"] + if result["deploy_type"] == "preaggregation" + ] + assert [(result["name"], result["operation"]) for result in results] == [ + ("default.analytics.hh_future", "unknown"), + ( + "default.analytics.hh_unres " + "[preagg_unres.default.us_state.state_name]", + "unknown", + ), + ] + assert results[1]["message"] == ( + "may be dropped from spec, depending on the pre-aggregation(s) " + "that could not be previewed" + ) + assert results[0]["message"].startswith( + "cannot be previewed until its metrics and dimensions exist: ", + ) + finally: + _clear_query_service(client) + + @pytest.mark.asyncio + async def test_legacy_named_preagg_still_reconciles( + self, + session, + client, + default_hard_hats, + default_us_states, + default_us_state, + ): + """Rows written before the handle was retired keep their ``name`` in the + database. Reconciliation never read it -- it matches on revision, grain + and measures -- so such a row still loads, updates in place rather than + being duplicated, and is labelled by its table and grain like any + other.""" + _override_query_service(client, ["hard_hat_count", "state_name"]) + try: + nodes = _hard_hat_deploy_nodes( + default_hard_hats, + default_us_states, + default_us_state, + ) + spec = _preagg_spec( + "hh_legacy", + {"${prefix}default.count_hard_hats": "hard_hat_count"}, + ) + deployment = DeploymentSpec( + namespace="preagg_legacy", + nodes=nodes, + preaggregations=[spec], + ) + data = await deploy_and_wait(client, deployment) + assert data["status"] == "success", data["results"] + + # Back-date the row to what an older DJ would have written. + fact_node = "preagg_legacy.default.hard_hat_facts" + listing = await client.get("/preaggs/", params={"node_name": fact_node}) + preagg_id = listing.json()["items"][0]["id"] + legacy = await session.get(PreAggregation, preagg_id) + legacy.name = "preagg_legacy.hard_hats_by_state" + await session.commit() + + expected = [ + { + "name": "default.analytics.hh_legacy " + "[preagg_legacy.default.us_state.state_name]", + "deploy_type": "preaggregation", + "status": "success", + "operation": "update", + "message": "externally-built pre-aggregation", + "changed_fields": [], + }, + ] + impact = await client.post( + "/deployments/impact", + json=deployment.model_dump(), + ) + assert [ + result + for result in impact.json()["results"] + if result["deploy_type"] == "preaggregation" + ] == expected + + data = await deploy_and_wait(client, deployment) + assert data["status"] == "success", data["results"] + assert [ + result + for result in data["results"] + if result["deploy_type"] == "preaggregation" + ] == expected + + # Updated in place: still one row, still carrying its legacy handle. + listing = await client.get("/preaggs/", params={"node_name": fact_node}) + items = listing.json()["items"] + assert [(item["id"], item["name"]) for item in items] == [ + (preagg_id, "preagg_legacy.hard_hats_by_state"), + ] + finally: + _clear_query_service(client) + @pytest.mark.asyncio async def test_delete_on_removal_scoped_to_namespace( self, @@ -7864,7 +8138,6 @@ async def test_delete_on_removal_scoped_to_namespace( default_us_state, ) spec = _preagg_spec( - "keep_me", "hh_keep", {"${prefix}default.count_hard_hats": "hard_hat_count"}, ) @@ -7917,7 +8190,6 @@ async def test_deploy_updates_existing_preagg( nodes=nodes, preaggregations=[ _preagg_spec( - "hh_agg", "hh_agg_v1", {"${prefix}default.count_hard_hats": "hh_count_v1"}, ), @@ -7934,7 +8206,6 @@ async def test_deploy_updates_existing_preagg( nodes=nodes, preaggregations=[ _preagg_spec( - "hh_agg", "hh_agg_v2", {"${prefix}default.count_hard_hats": "hh_count_v2"}, ), @@ -7979,7 +8250,6 @@ async def test_preagg_kept_when_none_declared( nodes=nodes, preaggregations=[ _preagg_spec( - "keep_me2", "hh_keep2", {"${prefix}default.count_hard_hats": "hard_hat_count"}, ), @@ -8038,7 +8308,6 @@ async def test_preagg_kept_warning_is_surfaced( nodes=nodes, preaggregations=[ _preagg_spec( - "warn_me", "hh_warn", {"${prefix}default.count_hard_hats": "hard_hat_count"}, ), @@ -8118,7 +8387,6 @@ async def test_deploy_preagg_with_dimension_columns( nodes=nodes, preaggregations=[ _preagg_spec( - "hh_by_state", "hh_by_state_agg", {"${prefix}default.count_hard_hats": "hard_hat_count"}, dimension_columns={ diff --git a/datajunction-server/tests/api/preaggregations_test.py b/datajunction-server/tests/api/preaggregations_test.py index 443d62392..a6f238b78 100644 --- a/datajunction-server/tests/api/preaggregations_test.py +++ b/datajunction-server/tests/api/preaggregations_test.py @@ -14,7 +14,10 @@ from datajunction_server.database.preaggregation import PreAggregation from datajunction_server.models.materialization import MaterializationStrategy from datajunction_server.models.partition import Granularity, PartitionType -from datajunction_server.models.preaggregation import WorkflowUrl +from datajunction_server.models.preaggregation import ( + PREAGG_NAME_REMOVED_ERROR, + WorkflowUrl, +) from datajunction_server.utils import get_query_service_client from datajunction_server.construction.build_v3.builder import build_measures_sql from datajunction_server.models.access import ResourceAction @@ -2683,17 +2686,17 @@ async def test_register_requires_query_service( del client_with_build_v3.app.dependency_overrides[get_query_service_client] @pytest.mark.asyncio - async def test_register_sets_external_strategy_and_name( + async def test_register_sets_external_strategy_and_no_name( self, client_with_build_v3: AsyncClient, ): - """A registered pre-agg is marked EXTERNAL and keeps its handle.""" + """A registered pre-agg is marked EXTERNAL and carries no handle: it is + identified by the table it is bound to and the grain it covers.""" _mock_query_service(client_with_build_v3, ["revenue_total", "order_cnt"]) try: response = await client_with_build_v3.post( "/preaggs/register", json={ - "name": "aov_by_category", "metrics": ["v3.avg_order_value"], "dimensions": ["v3.product.category"], "table": { @@ -2711,10 +2714,43 @@ async def test_register_sets_external_strategy_and_name( assert response.status_code == 201, response.text preagg = response.json()["preaggs"][0] assert preagg["strategy"] == "external" - assert preagg["name"] == "aov_by_category" + assert preagg["name"] is None finally: del client_with_build_v3.app.dependency_overrides[get_query_service_client] + @pytest.mark.asyncio + async def test_register_rejects_a_name( + self, + client_with_build_v3: AsyncClient, + ): + """The retired handle is refused with an explanation, on the same terms + as the deploy spec, rather than being accepted and ignored.""" + response = await client_with_build_v3.post( + "/preaggs/register", + json={ + "name": "aov_by_category", + "metrics": ["v3.avg_order_value"], + "dimensions": ["v3.product.category"], + "table": { + "catalog": "default", + "schema": "analytics", + "table": "aov_agg", + }, + "measure_columns": {"v3.total_revenue": "revenue_total"}, + }, + ) + assert response.status_code == 422, response.text + assert [ + (error["loc"], error["type"], error["msg"]) + for error in response.json()["detail"] + ] == [ + ( + ["body"], + "value_error", + f"Value error, {PREAGG_NAME_REMOVED_ERROR}", + ), + ] + async def _register_external(self, client, table_name): """Register a simple external pre-agg and return its id.""" response = await client.post( diff --git a/datajunction-server/tests/models/deployment_test.py b/datajunction-server/tests/models/deployment_test.py index 324cdff61..a3614e509 100644 --- a/datajunction-server/tests/models/deployment_test.py +++ b/datajunction-server/tests/models/deployment_test.py @@ -1,4 +1,5 @@ import pytest +from pydantic import ValidationError from datajunction_server.errors import DJInvalidDeploymentConfig from datajunction_server.models.deployment import ( @@ -28,6 +29,7 @@ ) from datajunction_server.models.materialization import MaterializationStrategy from datajunction_server.models.node import MetricUnit, NodeMode, NodeType +from datajunction_server.models.preaggregation import PREAGG_NAME_REMOVED_ERROR def test_source_spec(): @@ -618,7 +620,6 @@ def test_deployment_spec_preserves_explicit_preagg_namespace(): namespace="ns", preaggregations=[ PreAggSpec( - name="already_scoped", namespace="explicit", catalog="c", schema="s", @@ -626,7 +627,6 @@ def test_deployment_spec_preserves_explicit_preagg_namespace(): measure_columns={}, ), PreAggSpec( - name="unscoped", catalog="c", schema="s", table="t", @@ -638,11 +638,40 @@ def test_deployment_spec_preserves_explicit_preagg_namespace(): assert spec.preaggregations[1].namespace == "ns" +def test_preagg_spec_rejects_a_name(): + """ + The retired handle is refused outright rather than quietly dropped, so a + spec that still carries one is fixed at the source instead of deploying + with a field nobody reads. + """ + with pytest.raises(ValidationError) as exc_info: + PreAggSpec( + name="views_by_page", + catalog="warehouse", + schema="analytics", + table="views_by_page_daily", + ) + assert [ + (error["loc"], error["type"], error["msg"]) for error in exc_info.value.errors() + ] == [((), "value_error", f"Value error, {PREAGG_NAME_REMOVED_ERROR}")] + assert PREAGG_NAME_REMOVED_ERROR == ( + "Pre-aggregations no longer take a `name`. DJ identifies an " + "externally-registered pre-aggregation by the table it is bound to and " + "the grain it covers, and reconciles a deploy against those, so the " + "handle was never read. Remove `name` from the pre-aggregation." + ) + + +def test_preagg_spec_table_ref(): + """The table reference is what labels a pre-aggregation in deploy results.""" + spec = PreAggSpec(catalog="warehouse", schema="analytics", table="views_daily") + assert spec.table_ref == "warehouse.analytics.views_daily" + + def test_preagg_spec_renders_dimension_columns(): """dimension_columns keys are prefix-rendered against the namespace, like measure_columns; values (physical columns) are left as-is.""" spec = PreAggSpec( - name="p", namespace="ns", metrics=["${prefix}count"], dimensions=["${prefix}d.attr"], diff --git a/docs/content/0.1.0/docs/dj-concepts/aggregate-awareness.md b/docs/content/0.1.0/docs/dj-concepts/aggregate-awareness.md index 7dc11d9fb..b65dbfbed 100644 --- a/docs/content/0.1.0/docs/dj-concepts/aggregate-awareness.md +++ b/docs/content/0.1.0/docs/dj-concepts/aggregate-awareness.md @@ -315,7 +315,6 @@ like this: ```yaml # views_by_page.yaml kind: preagg -name: views_by_page metrics: - ${prefix}view_rate dimensions: @@ -345,6 +344,13 @@ dimension can even be a joined attribute the table has denormalized (say it stor rather than an account key you'd otherwise join through), which DJ then reads straight from the table with no join. +You'll notice the spec has no name of its own. It doesn't need one: a pre-aggregation is identified by +the table it is bound to together with the grain it covers, which is exactly what DJ matches on when it +reconciles a deploy, and that's how you'll see it referred to in deployment results (something like +`warehouse.analytics.views_by_page_daily [geo_country_d.country_iso_code, page_d.page_id]`). Earlier +versions asked for a `name` here; supplying one now is an error, so that a spec carrying a field nothing +reads gets fixed rather than deployed. + On deploy, DJ registers any pre-aggregation specs it finds. Because deployments are the source of truth, it also removes a previously-registered pre-aggregation once you drop its spec from a deploy that still declares others — the same way removing a node file deletes that node. As a safeguard against an