From 0b25b9c95343474d988ee953c0ec0854a9b6d565 Mon Sep 17 00:00:00 2001
From: Gerda Shank <gerda@dbtlabs.com>
Date: Tue, 16 May 2023 16:17:29 -0400
Subject: [PATCH] use StateRelation instead of RelationalNode

---
 core/dbt/contracts/graph/manifest.py |  4 ++--
 core/dbt/contracts/graph/nodes.py    | 17 ++++++++++++-----
 2 files changed, 14 insertions(+), 7 deletions(-)

diff --git a/core/dbt/contracts/graph/manifest.py b/core/dbt/contracts/graph/manifest.py
index 731202bbf7a..d72b1032049 100644
--- a/core/dbt/contracts/graph/manifest.py
+++ b/core/dbt/contracts/graph/manifest.py
@@ -37,7 +37,7 @@
     GraphMemberNode,
     ResultNode,
     BaseNode,
-    RelationalNode,
+    StateRelation,
     ManifestOrPublicNode,
 )
 from dbt.contracts.graph.unparsed import SourcePatch, NodeVersion, UnparsedVersion
@@ -1119,7 +1119,7 @@ def add_from_artifact(
             current = self.nodes.get(unique_id)
             if current and (node.resource_type in refables and not node.is_ephemeral):
                 other_node = other.nodes[unique_id]
-                state_relation = RelationalNode(
+                state_relation = StateRelation(
                     other_node.database, other_node.schema, other_node.alias
                 )
                 self.nodes[unique_id] = current.replace(state_relation=state_relation)
diff --git a/core/dbt/contracts/graph/nodes.py b/core/dbt/contracts/graph/nodes.py
index c3240a764cc..4e9ccfa59e2 100644
--- a/core/dbt/contracts/graph/nodes.py
+++ b/core/dbt/contracts/graph/nodes.py
@@ -271,7 +271,9 @@ def add_public_node(self, value: str):
 
 
 @dataclass
-class RelationalNode(HasRelationMetadata):
+class StateRelation(dbtClassMixin):
+    database: Optional[str]
+    schema: str
     alias: str
 
     @property
@@ -280,10 +282,15 @@ def identifier(self):
 
 
 @dataclass
-class ParsedNodeMandatory(GraphNode, RelationalNode, Replaceable):
+class ParsedNodeMandatory(GraphNode, HasRelationMetadata, Replaceable):
+    alias: str
     checksum: FileHash
     config: NodeConfig = field(default_factory=NodeConfig)
 
+    @property
+    def identifier(self):
+        return self.alias
+
 
 # This needs to be in all ManifestNodes and also in SourceDefinition,
 # because of "source freshness"
@@ -619,7 +626,7 @@ class ModelNode(CompiledNode):
     constraints: List[ModelLevelConstraint] = field(default_factory=list)
     version: Optional[NodeVersion] = None
     latest_version: Optional[NodeVersion] = None
-    state_relation: Optional[RelationalNode] = None
+    state_relation: Optional[StateRelation] = None
 
     @property
     def is_latest_version(self) -> bool:
@@ -802,7 +809,7 @@ class SeedNode(ParsedNode):  # No SQLDefaults!
     # and we need the root_path to load the seed later
     root_path: Optional[str] = None
     depends_on: MacroDependsOn = field(default_factory=MacroDependsOn)
-    state_relation: Optional[RelationalNode] = None
+    state_relation: Optional[StateRelation] = None
 
     def same_seeds(self, other: "SeedNode") -> bool:
         # for seeds, we check the hashes. If the hashes are different types,
@@ -1001,7 +1008,7 @@ class IntermediateSnapshotNode(CompiledNode):
 class SnapshotNode(CompiledNode):
     resource_type: NodeType = field(metadata={"restrict": [NodeType.Snapshot]})
     config: SnapshotConfig
-    state_relation: Optional[RelationalNode] = None
+    state_relation: Optional[StateRelation] = None
 
 
 # ====================================