Fix metric declared column check

datajunction-server/datajunction_server/api/graphql/schema.graphql

@@ -199,6 +199,7 @@ enum ErrorCode {
   TAG_NOT_FOUND
   CATALOG_NOT_FOUND
   INVALID_NAMESPACE
+  INVALID_SPEC_FIELD
 }
 
 type GeneratedSQL {

datajunction-server/datajunction_server/errors.py

@@ -65,6 +65,7 @@ class ErrorCode(IntEnum):
     TAG_NOT_FOUND = 700
     CATALOG_NOT_FOUND = 701
     INVALID_NAMESPACE = 702
+    INVALID_SPEC_FIELD = 703
 
 
 class DebugType(TypedDict, total=False):

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

@@ -488,14 +488,23 @@ def validate_query_node(
                     validation.output_columns,
                     spec,
                 )
+                internal_field_error = self._check_internal_fields(spec)
+                # Declaring a field the server owns makes every downstream
+                # complaint about that field a symptom. Report the cause only.
+                declared_columns_error = (
+                    None
+                    if internal_field_error
+                    else self._check_declared_columns_exist(
+                        spec,
+                        validation.output_columns,
+                    )
+                )
                 errors = [
                     err
                     for err in [
                         self._check_inferred_columns(inferred_columns),
-                        self._check_declared_columns_exist(
-                            spec,
-                            validation.output_columns,
-                        ),
+                        internal_field_error,
+                        declared_columns_error,
                         self._check_primary_key(inferred_columns, spec),
                         self._check_metric_query(spec, spec.query_ast),
                     ]
@@ -601,6 +610,27 @@ def _check_inferred_columns(self, columns: list[ColumnSpec]) -> DJError | None:
             )
         return None
 
+    @staticmethod
+    def _check_internal_fields(spec: NodeSpec) -> DJError | None:
+        """
+        Reject a spec that sets a field the server owns.
+
+        Specs built from existing nodes carry these, but they reach validation
+        only through the `_skip_validation` fast path above, so a value here
+        was written by an author.
+        """
+        authored = spec.authored_internal_fields()
+        if authored:
+            label = spec.node_type.value.capitalize()
+            return DJError(
+                code=ErrorCode.INVALID_SPEC_FIELD,
+                message=" ".join(
+                    f"{label} {spec.rendered_name} must not declare {field}. {remedy}"
+                    for field, remedy in authored
+                ),
+            )
+        return None
+
     @staticmethod
     def _check_declared_columns_exist(
         spec: NodeSpec,

datajunction-server/datajunction_server/models/deployment.py

@@ -645,6 +645,12 @@ class NodeSpec(NamespacedSpec):
         "owners": ChangeTier.NONE,
         "tags": ChangeTier.NONE,
     }
+
+    # Fields the server owns, mapped to the remedy shown when an author sets
+    # one. Each class declares only the fields it introduces; lookup walks the
+    # MRO. Validation rejects a deployment that provides any of them.
+    INTERNAL_FIELDS: ClassVar[dict[str, str]] = {}
+
     _query_ast: Any | None = PrivateAttr(default=None)
     # Internal: marks specs from already-validated sources (e.g., branch copies)
     # that can skip expensive SQL parsing and validation
@@ -794,6 +800,22 @@ def has_explicit_order_change_tier(cls, field: str) -> bool:
         """Whether some class in the MRO classifies reordering `field`."""
         return cls._declared_tier("FIELD_ORDER_CHANGE_TIERS", field) is not None
 
+    def authored_internal_fields(self) -> list[tuple[str, str]]:
+        """
+        Internal-only fields carrying a value, with each one's remedy.
+
+        Tested by value rather than by `model_fields_set`, so that server code
+        populating a field with nothing in it reads as unset.
+        """
+        declared: dict[str, str] = {}
+        for klass in reversed(type(self).__mro__):
+            declared.update(klass.__dict__.get("INTERNAL_FIELDS") or {})
+        return [
+            (field, remedy)
+            for field, remedy in sorted(declared.items())
+            if getattr(self, field, None)
+        ]
+
     @classmethod
     def unclassified_fields(cls) -> list[str]:
         """Fields on this spec class that nobody classified. Should always be empty."""
@@ -1034,6 +1056,12 @@ class MetricSpec(NodeSpec):
     min_decimal_exponent: int | None = None
     max_decimal_exponent: int | None = None
 
+    # A metric's one output column is always named after the node, so a
+    # declared name can never match it.
+    INTERNAL_FIELDS: ClassVar[dict[str, str]] = {
+        "columns": "Remove the columns block; set `unit` on the metric.",
+    }
+
     FIELD_CHANGE_TIERS: ClassVar[dict[str, ChangeTier]] = {
         "query": ChangeTier.MAJOR,
         "columns": ChangeTier.NONE,

datajunction-server/tests/internal/deployment/validation_test.py

@@ -13,6 +13,7 @@
 from datajunction_server.database.node import Node, NodeRevision
 from datajunction_server.database.user import OAuthProvider, User
 from datajunction_server.errors import ErrorCode
+from datajunction_server.internal.deployment.utils import extract_node_graph
 from datajunction_server.internal.deployment.validation import (
     NodeSpecBulkValidator,
     NodeValidationResult,
@@ -276,13 +277,128 @@ async def test_validate_query_node_flags_unmatched_declared_column(
         validator = NodeSpecBulkValidator(validation_context)
         result = validator.validate_query_node(spec)
 
+        # `test.parent` does not resolve here: the spec is named `transform`
+        # while the fixture's graph is keyed on `test.transform`, so the parent
+        # columns map comes back empty and nothing is inferred. The declared
+        # columns are unmatched for that reason as well as on their own merits.
+        assert [(e.code, e.message) for e in result.errors] == [
+            (
+                ErrorCode.INVALID_SQL_QUERY,
+                "No columns could be inferred from the SQL query.",
+            ),
+            (
+                ErrorCode.INVALID_COLUMN,
+                "Declared column(s) ['full_name', 'id'] on node transform do not "
+                "match any column produced by the query. Check for a missing or "
+                "mismatched column alias.",
+            ),
+            (
+                ErrorCode.TYPE_INFERENCE,
+                "Table `test.parent` not found in parent columns map. Available: []",
+            ),
+        ]
         assert result.status == NodeStatus.INVALID
-        error_codes = [e.code for e in result.errors]
-        assert ErrorCode.INVALID_COLUMN in error_codes
-        message = next(
-            e.message for e in result.errors if e.code == ErrorCode.INVALID_COLUMN
+
+    @pytest.mark.asyncio
+    async def test_validate_query_node_rejects_declared_columns_on_metric(
+        self,
+        session: AsyncSession,
+        parent_node: Node,
+    ):
+        """A metric must not declare columns, whatever its query aliases."""
+        context = ValidationContext(
+            session=session,
+            node_graph={"test.weekly_active_players": [parent_node.name]},
+            dependency_nodes={parent_node.name: parent_node},
+        )
+        spec = MetricSpec(
+            name="test.weekly_active_players",
+            query="SELECT SUM(value) FROM test.parent",
+            description="A test metric",
+            mode="published",
+            columns=[
+                ColumnSpec(name="weekly_active_players", display_name="WAP"),
+            ],
+        )
+        validator = NodeSpecBulkValidator(context)
+        result = validator.validate_query_node(spec)
+
+        assert [(e.code, e.message) for e in result.errors] == [
+            (
+                ErrorCode.INVALID_SPEC_FIELD,
+                "Metric test.weekly_active_players must not declare columns. "
+                "Remove the columns block; set `unit` on the metric.",
+            ),
+        ]
+        assert result.status == NodeStatus.INVALID
+
+    @pytest.mark.asyncio
+    async def test_validate_query_node_rejects_declared_columns_on_aliased_metric(
+        self,
+        session: AsyncSession,
+        parent_node: Node,
+    ):
+        """A metric aliasing its own short name is still rejected.
+
+        Going through ``extract_node_graph`` caches the metric-aliased AST on
+        the spec, which is the form a deploy validates. This is the shape that
+        failed in production.
+        """
+        spec = MetricSpec(
+            name="test.weekly_active_players",
+            query="SELECT SUM(value) AS weekly_active_players FROM test.parent",
+            description="A test metric",
+            mode="published",
+            columns=[
+                ColumnSpec(name="weekly_active_players", display_name="WAP"),
+            ],
+        )
+        node_graph = extract_node_graph([spec])
+        assert (
+            spec.query_ast.select.projection[0].alias_or_name.identifier()
+            == "test_DOT_weekly_active_players"
         )
-        assert "full_name" in message
+        context = ValidationContext(
+            session=session,
+            node_graph=node_graph,
+            dependency_nodes={parent_node.name: parent_node},
+        )
+        validator = NodeSpecBulkValidator(context)
+        result = validator.validate_query_node(spec)
+
+        assert [(e.code, e.message) for e in result.errors] == [
+            (
+                ErrorCode.INVALID_SPEC_FIELD,
+                "Metric test.weekly_active_players must not declare columns. "
+                "Remove the columns block; set `unit` on the metric.",
+            ),
+        ]
+        assert result.status == NodeStatus.INVALID
+
+    @pytest.mark.asyncio
+    async def test_validate_query_node_allows_metric_without_columns(
+        self,
+        session: AsyncSession,
+        parent_node: Node,
+    ):
+        """A metric that declares no columns validates clean."""
+        spec = MetricSpec(
+            name="test.weekly_active_players",
+            query="SELECT SUM(value) AS weekly_active_players FROM test.parent",
+            description="A test metric",
+            mode="published",
+        )
+        node_graph = extract_node_graph([spec])
+        context = ValidationContext(
+            session=session,
+            node_graph=node_graph,
+            dependency_nodes={parent_node.name: parent_node},
+        )
+        validator = NodeSpecBulkValidator(context)
+        result = validator.validate_query_node(spec)
+
+        assert result.errors == []
+        assert result.status == NodeStatus.VALID
 
     @pytest.mark.asyncio
     async def test_validate_query_node_flags_hardcoded_namespace(