From 8cb70d13cd3b28e4531c2226c7e36fe7c24d3171 Mon Sep 17 00:00:00 2001 From: Yian Shang Date: Wed, 12 Aug 2026 03:12:07 -0700 Subject: [PATCH] Stop asking for a name on an externally-registered pre-aggregation The name was ceremony. Reconciliation never read it: the apply finds the rows a spec maps to by decomposing its metrics and matching on revision, grain and measure identity, and deletes by row identity. The column comment claimed an availability-by-name callback, but the only callback is POST /preaggs/{preagg_id}/availability/, addressed by integer id. Supplying a name is now an error rather than silently ignored, on both the deploy spec and POST /preaggs/register (which passed it unrendered where the deploy path passed the rendered name, so the two were never comparable anyway). The database column stays, nullable, so rows written before this still load. Deployment results now label a pre-aggregation by the external table it adopts plus the grain it covers -- the table alone is ambiguous, since one registration can produce a pre-agg per grain group. Dropping the name also exposes a real divergence: the dry run planned creates, updates and deletes by name membership while the apply worked by row identity, so editing a spec's grain but keeping its name previewed a single update where the apply did a create plus a delete. The preview now resolves each spec to its grain groups and matching rows the same way the apply does -- decomposition needs no query service, only the physical column binding does -- and reports the operations the apply will perform. A spec it cannot resolve yet (a first deploy also creates the metrics it refers to) is reported once as UNKNOWN, and the leftover rows are then reported as UNKNOWN too rather than promised for deletion. --- .../python/tests/test_deploy.py | 3 +- .../api/preaggregations.py | 1 - .../database/preaggregation.py | 8 +- .../internal/deployment/orchestrator.py | 169 +++++++--- .../internal/preaggregations.py | 68 +++- .../datajunction_server/models/deployment.py | 24 +- .../models/preaggregation.py | 40 ++- .../tests/api/deployments_test.py | 310 ++++++++++++++++-- .../tests/api/preaggregations_test.py | 46 ++- .../tests/models/deployment_test.py | 35 +- .../docs/dj-concepts/aggregate-awareness.md | 8 +- 11 files changed, 622 insertions(+), 90 deletions(-) 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