[Feature] dbt_clone sf implementation

.changes/unreleased/Features-20230622-114621.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: add dbt-snowflake portion of dbt_clone functionality
+time: 2023-06-22T11:46:21.399594-05:00
+custom:
+  Author: McKnight-42 aranke
+  Issue: "7256"

dbt/include/snowflake/macros/materializations/clone.sql

@@ -0,0 +1,11 @@
+{% macro snowflake__can_clone_table() %}
+    {{ return(True) }}
+{% endmacro %}
+
+{% macro snowflake__create_or_replace_clone(this_relation, defer_relation) %}
+    create or replace
+      {{ "transient" if config.get("transient", true) }}
+      table {{ this_relation }}
+      clone {{ defer_relation }}
+      {{ "copy grants" if config.get("copy_grants", false) }}
+{% endmacro %}

tests/functional/adapter/dbt_clone/test_dbt_clone.py

@@ -0,0 +1,83 @@
+import pytest
+import shutil
+import os
+from copy import deepcopy
+from dbt.tests.util import run_dbt
+from dbt.tests.adapter.dbt_clone.test_dbt_clone import BaseClonePossible
+
+
+class TestSnowflakeClonePossible(BaseClonePossible):
+    @pytest.fixture(autouse=True)
+    def clean_up(self, project):
+        yield
+        with project.adapter.connection_named("__test"):
+            relation = project.adapter.Relation.create(
+                database=project.database, schema=f"{project.test_schema}_SEEDS"
+            )
+            project.adapter.drop_schema(relation)
+
+            relation = project.adapter.Relation.create(
+                database=project.database, schema=project.test_schema
+            )
+            project.adapter.drop_schema(relation)
+
+    pass
+
+
+table_model_1_sql = """
+    {{ config(
+        materialized='table',
+        transient=true,
+    ) }}
+
+    select 1 as fun
+    """
+
+
+class TestSnowflakeCloneTrainsentTable:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "table_model.sql": table_model_1_sql,
+        }
+
+    @pytest.fixture(scope="class")
+    def other_schema(self, unique_schema):
+        return unique_schema + "_other"
+
+    @pytest.fixture(scope="class")
+    def profiles_config_update(self, dbt_profile_target, unique_schema, other_schema):
+        outputs = {"default": dbt_profile_target, "otherschema": deepcopy(dbt_profile_target)}
+        outputs["default"]["schema"] = unique_schema
+        outputs["otherschema"]["schema"] = other_schema
+        return {"test": {"outputs": outputs, "target": "default"}}
+
+    def copy_state(self, project_root):
+        state_path = os.path.join(project_root, "state")
+        if not os.path.exists(state_path):
+            os.makedirs(state_path)
+        shutil.copyfile(
+            f"{project_root}/target/manifest.json", f"{project_root}/state/manifest.json"
+        )
+
+    def run_and_save_state(self, project_root, with_snapshot=False):
+        results = run_dbt(["run"])
+        assert len(results) == 1
+        assert not any(r.node.deferred for r in results)
+
+        self.copy_state(project_root)
+
+    def test_can_clone_transient_table(self, project, other_schema):
+        project.create_test_schema(other_schema)
+        self.run_and_save_state(project.project_root)
+
+        clone_args = [
+            "clone",
+            "--state",
+            "state",
+            "--target",
+            "otherschema",
+        ]
+
+        results = run_dbt(clone_args)
+        assert len(results) == 1