Unify single-node and bulk deployment validation paths

datajunction-clients/python/tests/examples.py

@@ -372,7 +372,7 @@
                         FROM default.hard_hats hh
                         LEFT JOIN default.hard_hat_state hhs
                         ON hh.hard_hat_id = hhs.hard_hat_id
-                        WHERE hh.state_id = 'NY'
+                        WHERE hhs.state_id = 'NY'
                     """,
             "mode": "published",
             "name": "default.local_hard_hats",

datajunction-server/datajunction_server/database/node.py

@@ -1320,7 +1320,12 @@ def default_load_options(cls):
                 ),
                 joinedload(DimensionLink.node_revision),
             ),
-            selectinload(NodeRevision.required_dimensions),
+            selectinload(NodeRevision.required_dimensions).options(
+                # Column.node_revision back-ref is read by Column.full_name()
+                # during required-dimensions resolution; preload it so
+                # accessing it in async context doesn't trip MissingGreenlet.
+                joinedload(Column.node_revision).load_only(NodeRevision.name),
+            ),
             selectinload(NodeRevision.availability),
             # Load created_by for API responses (but noload in /sql/ endpoint's custom options)
             selectinload(NodeRevision.created_by),

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

@@ -24,7 +24,7 @@
 from datajunction_server.database.history import History
 from datajunction_server.database.metricmetadata import MetricMetadata
 from datajunction_server.database.namespace import NodeNamespace
-from datajunction_server.database.node import NodeRelationship
+from datajunction_server.database.node import MissingParent, NodeRelationship
 from datajunction_server.database.partition import Partition
 from datajunction_server.database.tag import Tag
 from datajunction_server.database.user import User, OAuthProvider
@@ -37,6 +37,7 @@
     ErrorCode,
 )
 from datajunction_server.internal.deployment.utils import (
+    classify_parents,
     extract_node_graph,
     topological_levels,
     DeploymentContext,
@@ -3122,6 +3123,24 @@ def _create_or_update_node(
             new_node.tags = tags  # type: ignore
         return new_node
 
+    @staticmethod
+    def _classify_parents(
+        spec: NodeSpec,
+        dep_names: list[str],
+        dependency_nodes: dict[str, Node],
+    ) -> tuple[list[Node], list[str]]:
+        """Split a spec's dependency names into resolved parents and missing names.
+
+        Thin wrapper around the shared utils.classify_parents helper — derives
+        the is_derived_metric flag from the spec so callers don't have to.
+        """
+        is_derived_metric = (
+            spec.node_type == NodeType.METRIC
+            and spec.query_ast is not None
+            and spec.query_ast.select.from_ is None
+        )
+        return classify_parents(is_derived_metric, dep_names, dependency_nodes)
+
     async def _create_node_revision(
         self,
         new_node: Node,
@@ -3132,11 +3151,11 @@ async def _create_node_revision(
         """Create node revision with inferred columns and dependencies"""
         existing = self.registry.nodes.get(result.spec.rendered_name)
         old_node_revision = existing.current if existing else None
-        parents = [
-            dependency_nodes.get(parent)
-            for parent in node_graph.get(result.spec.rendered_name, [])
-            if parent in dependency_nodes
-        ]
+        parents, missing_parent_names = self._classify_parents(
+            result.spec,
+            node_graph.get(result.spec.rendered_name, []),
+            dependency_nodes,
+        )
         if result.spec.node_type != NodeType.SOURCE:
             # Pick the first parent with a non-virtual catalog to assign as the
             # catalog inherited from source parents.
@@ -3175,11 +3194,8 @@ async def _create_node_revision(
             node=new_node,
             catalog=catalog,
             status=result.status,
-            parents=[
-                dependency_nodes.get(parent)
-                for parent in node_graph.get(result.spec.rendered_name, [])
-                if parent in dependency_nodes
-            ],
+            parents=parents,
+            missing_parents=[MissingParent(name=n) for n in missing_parent_names],
             created_by_id=self.context.current_user.id,
             custom_metadata=result.spec.custom_metadata,
             # Initialize to empty so _deploy_links can append without triggering a