From d0ddb96aea48af342faacb8c468214679dcc9efa Mon Sep 17 00:00:00 2001
From: Kshitij Aranke <kshitij.aranke@dbtlabs.com>
Date: Mon, 5 Jun 2023 10:53:55 -0700
Subject: [PATCH] wip

---
 core/dbt/contracts/graph/manifest.py | 25 +++++++++++++++++++++++++
 core/dbt/contracts/graph/nodes.py    |  9 +++++++++
 2 files changed, 34 insertions(+)

diff --git a/core/dbt/contracts/graph/manifest.py b/core/dbt/contracts/graph/manifest.py
index 3f4f3adec2c..4c9f9d4f95f 100644
--- a/core/dbt/contracts/graph/manifest.py
+++ b/core/dbt/contracts/graph/manifest.py
@@ -39,6 +39,7 @@
     BaseNode,
     ManifestOrPublicNode,
     ModelNode,
+    RelationalNode,
 )
 from dbt.contracts.graph.unparsed import SourcePatch, NodeVersion, UnparsedVersion
 from dbt.contracts.graph.manifest_upgrade import upgrade_manifest_json
@@ -1133,6 +1134,30 @@ def merge_from_artifact(
         sample = list(islice(merged, 5))
         fire_event(MergedFromState(num_merged=len(merged), sample=sample))
 
+    # Called by CloneTask.defer_to_manifest
+    def add_from_artifact(
+        self,
+        other: "WritableManifest",
+    ) -> None:
+        """Update this manifest by *adding* information about each node's location
+        in the other manifest.
+
+        Only non-ephemeral refable nodes are examined.
+        """
+        refables = set(NodeType.refable())
+        for unique_id, node in other.nodes.items():
+            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(
+                    other_node.database, other_node.schema, other_node.alias
+                )
+                self.nodes[unique_id] = current.replace(state_relation=state_relation)
+
+        # Rebuild the flat_graph, which powers the 'graph' context variable,
+        # now that we've deferred some nodes
+        self.build_flat_graph()
+
     # Methods that were formerly in ParseResult
 
     def add_macro(self, source_file: SourceFile, macro: Macro):
diff --git a/core/dbt/contracts/graph/nodes.py b/core/dbt/contracts/graph/nodes.py
index 24cd8f7ddcb..a3b988f35ff 100644
--- a/core/dbt/contracts/graph/nodes.py
+++ b/core/dbt/contracts/graph/nodes.py
@@ -259,6 +259,15 @@ def add_macro(self, value: str):
             self.macros.append(value)
 
 
+@dataclass
+class RelationalNode(HasRelationMetadata):
+    alias: str
+
+    @property
+    def identifier(self):
+        return self.alias
+
+
 @dataclass
 class DependsOn(MacroDependsOn):
     nodes: List[str] = field(default_factory=list)