Bind each metric and dimension to its column on the REST path too

datajunction-clients/python/tests/test_deploy.py

@@ -54,12 +54,11 @@ 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"
+        "metrics:\n  ns.revenue: revenue_sum\n"
+        "dimensions:\n  ns.date.day: day\n"
         "catalog: default\n"
         "schema: agg\n"
-        "table: revenue_agg\n"
-        "measure_columns:\n  ns.revenue: revenue_sum\n",
+        "table: revenue_agg\n",
     )
 
     svc = DeploymentService(MagicMock())
@@ -70,7 +69,8 @@ def test_reconstruct_deployment_spec_separates_preaggregations(tmp_path):
     assert len(spec["preaggregations"]) == 1
     preagg = spec["preaggregations"][0]
     assert preagg["name"] == "revenue_by_day"
-    assert preagg["measure_columns"] == {"ns.revenue": "revenue_sum"}
+    assert preagg["metrics"] == {"ns.revenue": "revenue_sum"}
+    assert preagg["dimensions"] == {"ns.date.day": "day"}
     # The discriminator is stripped before reaching the payload.
     assert "kind" not in preagg
 

datajunction-server/datajunction_server/api/preaggregations.py

@@ -735,11 +735,13 @@ async def register_preaggregations(
     Register an externally-built pre-aggregation table.
 
     Unlike ``/preaggs/plan`` (where DJ generates and owns the materialization),
-    this adopts a table built by an external pipeline. DJ decomposes the
-    requested metrics into component measures, binds each measure to a physical
-    column via ``measure_columns``, validates them against the table, records
-    the pre-aggregation, and — when ``valid_through_ts`` is supplied — marks it
-    available so grain resolution can route queries to it.
+    this adopts a table built by an external pipeline. Every metric and every
+    dimension is declared together with the physical column of the external table
+    that holds it, the same way a ``kind: preagg`` YAML spec declares it. DJ
+    decomposes the requested metrics into component measures, validates every
+    declared column against the table, records the pre-aggregation, and — when
+    ``valid_through_ts`` is supplied — marks it available so grain resolution can
+    route queries to it.
     """
     request_headers = dict(request.headers)
     if not query_service_client:
@@ -756,10 +758,14 @@ async def register_preaggregations(
     # inference and row creation). The same result is handed to the registration
     # below: resolving it twice can pick different parents, which would authorize
     # one set of nodes and write another.
+    # The request binds each reference to its physical column inline, so the
+    # references are the map keys and the bindings are the maps themselves.
+    metrics = list(data.metrics)
+    dimensions = list(data.dimensions)
     measures_result = await build_measures_sql(
         session=session,
-        metrics=data.metrics,
-        dimensions=data.dimensions,
+        metrics=metrics,
+        dimensions=dimensions,
         dialect=Dialect.SPARK,
         use_materialized=False,
     )
@@ -775,8 +781,6 @@ async def register_preaggregations(
         metrics=data.metrics,
         dimensions=data.dimensions,
         table=data.table,
-        measure_columns=data.measure_columns,
-        dimension_columns=data.dimension_columns,
         measures_result=measures_result,
     )
 

datajunction-server/datajunction_server/internal/deployment/orchestrator.py

@@ -1247,8 +1247,6 @@ async def _reconcile_preaggregations(self) -> None:
                             table=spec.table,
                             valid_through_ts=spec.valid_through_ts,
                         ),
-                        measure_columns=spec.rendered_measure_columns,
-                        dimension_columns=spec.rendered_dimension_columns,
                     )
                     upserted_ids.update(preagg.id for preagg in created)
                     for preagg in created:

datajunction-server/datajunction_server/internal/preaggregations.py

@@ -169,18 +169,22 @@ async def register_external_preaggregations(
     request_headers: dict[str, str],
     *,
     name: str | None,
-    metrics: list[str],
-    dimensions: list[str],
+    metrics: dict[str, str],
+    dimensions: dict[str, str],
     table: ExternalPreAggTable,
-    measure_columns: dict[str, str],
-    dimension_columns: dict[str, str] | None = None,
     measures_result: GeneratedMeasuresSQL | None = None,
 ) -> list[PreAggregation]:
     """
     Core logic for adopting an externally-built pre-aggregation table.
 
-    Decomposes ``metrics`` into component measures, binds each to a physical
-    column via ``measure_columns``, validates them against ``table``, and upserts
+    ``metrics`` and ``dimensions`` each map a reference to the physical column of
+    the external table that holds it — the shape both surfaces declare (a
+    ``kind: preagg`` YAML spec and the body of ``POST /preaggs/register``). The
+    references are the keys, so a reference can never be declared without its
+    binding or bound without being declared.
+
+    Decomposes the metrics into component measures, binds each to its declared
+    column, validates them against ``table``, and upserts
     the pre-aggregation(s) marked ``EXTERNAL``. Flushes but does NOT commit — the
     caller owns the transaction (the endpoint commits; the deploy orchestrator
     commits its whole plan). Callers must ensure ``query_service_client`` is
@@ -196,7 +200,7 @@ async def register_external_preaggregations(
     #    column. Keying on the hash alone would collapse SUM(x) and MAX(x),
     #    silently discarding one metric's declared column.
     measure_identity_to_column: dict[tuple[str, str], str] = {}
-    for metric_name, physical_column in measure_columns.items():
+    for metric_name, physical_column in metrics.items():
         node = await Node.get_by_name(
             session,
             metric_name,
@@ -209,7 +213,9 @@ async def register_external_preaggregations(
         )
         if not node or node.type != NodeType.METRIC:
             raise DJInvalidInputException(
-                message=f"'{metric_name}' in measure_columns is not a metric node.",
+                message=(
+                    f"'{metric_name}' declared under `metrics` is not a metric node."
+                ),
             )
         if not node.current.is_measure:
             raise DJInvalidInputException(
@@ -234,12 +240,12 @@ async def register_external_preaggregations(
     if measures_result is None:
         measures_result = await build_measures_sql(
             session=session,
-            metrics=metrics,
-            dimensions=dimensions,
+            metrics=list(metrics),
+            dimensions=list(dimensions),
             dialect=Dialect.SPARK,
             use_materialized=False,
         )
-    assert_dimension_refs_are_role_qualified(measures_result, dimensions)
+    assert_dimension_refs_are_role_qualified(measures_result, list(dimensions))
 
     # 3. Introspect the external table and confirm the declared columns exist.
     catalog = await get_catalog_by_name(session=session, name=table.catalog)
@@ -250,26 +256,17 @@ async def register_external_preaggregations(
         request_headers,
         catalog.engines[0] if catalog.engines else None,
     )
-    dimension_columns = dimension_columns or {}
-    unknown_dims = sorted(set(dimension_columns) - set(dimensions))
-    if unknown_dims:
-        raise DJInvalidInputException(
-            message=(
-                f"dimension_columns references {unknown_dims}, which are not in "
-                f"the pre-aggregation's dimensions {sorted(dimensions)}."
-            ),
-        )
     table_columns_by_name = {col.name: col.type for col in table_columns}
     missing_columns = sorted(
         column
-        for column in (*measure_columns.values(), *dimension_columns.values())
+        for column in (*metrics.values(), *dimensions.values())
         if column not in table_columns_by_name
     )
     if missing_columns:
         raise DJInvalidInputException(
             message=(
-                f"Columns {missing_columns} declared in measure_columns/"
-                f"dimension_columns were not found in table "
+                f"Columns {missing_columns} declared under `metrics`/`dimensions` "
+                f"were not found in table "
                 f"{table.catalog}.{table.schema_}.{table.table}."
             ),
         )
@@ -312,14 +309,20 @@ async def register_external_preaggregations(
         for component in grain_group.components:
             expr_hash = compute_expression_hash(component.expression)
             identity = (expr_hash, component.normalized_aggregation)
-            if identity not in measure_identity_to_column:
+            # Defensive since the map form landed: the grain groups are built
+            # from the very metrics `metrics` binds, and each is_measure metric
+            # contributes exactly one component, so every component is bound. It
+            # stays as the clear failure a bare KeyError below would not be, in
+            # case a metric ever decomposes differently in a grain group than it
+            # does on its own.
+            if identity not in measure_identity_to_column:  # pragma: no cover
                 raise DJInvalidInputException(
                     message=(
                         f"Measure '{component.name}' "
                         f"({component.normalized_aggregation} over "
                         f"'{component.expression}') required by the requested "
-                        f"metrics is not covered by measure_columns. Add the "
-                        f"is_measure metric it corresponds to."
+                        f"metrics is not covered by any column declared under "
+                        f"`metrics`. Add the is_measure metric it corresponds to."
                     ),
                 )
             physical_column = measure_identity_to_column[identity]
@@ -346,13 +349,13 @@ async def register_external_preaggregations(
                 ),
             )
 
-        # Bind dimension columns to physical columns via dimension_columns, keyed
-        # by dimension reference. Measures carry their binding on PreAggMeasure.
+        # Bind each grain dimension to its declared physical column, keyed by
+        # dimension reference. Measures carry their binding on PreAggMeasure.
         columns = []
         for col in grain_group.columns:
             source_column = None
             if col.semantic_type == "dimension":
-                source_column = dimension_columns.get(col.semantic_name)
+                source_column = dimensions.get(col.semantic_name)
                 if source_column is not None:
                     assert_column_type_compatible(
                         subject=f"dimension '{col.semantic_name}'",

datajunction-server/datajunction_server/models/deployment.py

@@ -37,6 +37,10 @@
     NodeType,
 )
 from datajunction_server.models.partition import Granularity, PartitionType
+from datajunction_server.models.preagg_binding import (
+    SPEC_SURFACE,
+    validate_column_bindings,
+)
 from datajunction_server.models.unit import (
     Unit,
     legacy_unit_to_structured,
@@ -208,6 +212,10 @@ class PreAggSpec(NamespacedSpec):
     separate ``measure_columns``/``dimension_columns`` maps -- is no longer
     accepted, and a spec still using it is rejected with a message describing
     what to write instead.
+
+    ``POST /preaggs/register`` takes the same declaration in JSON, and both
+    surfaces enforce these rules from
+    :mod:`datajunction_server.models.preagg_binding` so they cannot drift apart.
     """
 
     # Metric/dimension reference -> the physical column of the external table
@@ -222,89 +230,30 @@ class PreAggSpec(NamespacedSpec):
 
     model_config = ConfigDict(populate_by_name=True)
 
-    # Each removed field and the map that replaced it.
-    _REPLACED_FIELDS: ClassVar[dict[str, str]] = {
-        "measure_columns": "metrics",
-        "dimension_columns": "dimensions",
-    }
-
-    # Why a missing column can't be defaulted, per axis.
-    _UNBOUND_REASONS: ClassVar[dict[str, str]] = {
-        "metrics": (
-            "A measure's DJ-side name is auto-generated with an expression-hash "
-            "suffix, so there is no name to fall back on."
-        ),
-        "dimensions": (
-            "Write the column out even when it matches the DJ column name, so "
-            "the file says what the table actually holds."
-        ),
-    }
-
     @model_validator(mode="before")
     @classmethod
     def require_column_bindings(cls, data: Any) -> Any:
         """
         Hold the author to the map form: every metric and dimension names the
         physical column that holds it, and the fields that used to carry those
-        bindings separately are gone.
+        bindings separately are gone. Shared with ``POST /preaggs/register``,
+        which enforces the identical rules against a JSON body.
         """
-        if not isinstance(data, dict):
-            return data
-
-        name = data.get("name")
-        for removed, replacement in cls._REPLACED_FIELDS.items():
-            if removed in data:
-                raise DJInvalidDeploymentConfig(
-                    message=(
-                        f"Pre-aggregation '{name}' declares `{removed}`, which is no "
-                        f"longer a pre-aggregation field. Declare the physical column "
-                        f"alongside what it holds instead, as `{replacement}: "
-                        f"{{<reference>: <column>}}`, and drop the `{removed}` block."
-                    ),
-                )
-
-        for field, reason in cls._UNBOUND_REASONS.items():
-            value = data.get(field)
-            if value is None:
-                continue
-            if not isinstance(value, dict):
-                raise DJInvalidDeploymentConfig(
-                    message=(
-                        f"Pre-aggregation '{name}' declares `{field}` as a "
-                        f"{type(value).__name__}. `{field}` is a map from each "
-                        f"reference to the physical column of the external table "
-                        f"that holds it, e.g. `{field}: {{<reference>: <column>}}`."
-                    ),
-                )
-            unbound = [
-                reference for reference, column in value.items() if column is None
-            ]
-            if unbound:
-                raise DJInvalidDeploymentConfig(
-                    message=(
-                        f"Pre-aggregation '{name}' leaves the physical column empty "
-                        f"under `{field}` for {unbound}. {reason}"
-                    ),
-                )
-        return data
-
-    @property
-    def rendered_metrics(self) -> list[str]:
-        return [render_prefixes(metric, self.namespace) for metric in self.metrics]
-
-    @property
-    def rendered_dimensions(self) -> list[str]:
-        return [render_prefixes(dim, self.namespace) for dim in self.dimensions]
+        return validate_column_bindings(data, SPEC_SURFACE)
 
     @property
-    def rendered_measure_columns(self) -> dict[str, str]:
+    def rendered_metrics(self) -> dict[str, str]:
+        """Each metric reference rendered against the namespace, still bound to
+        its physical column."""
         return {
             render_prefixes(metric, self.namespace): column
             for metric, column in self.metrics.items()
         }
 
     @property
-    def rendered_dimension_columns(self) -> dict[str, str]:
+    def rendered_dimensions(self) -> dict[str, str]:
+        """Each dimension reference rendered against the namespace, still bound to
+        its physical column."""
         return {
             render_prefixes(dimension, self.namespace): column
             for dimension, column in self.dimensions.items()