skip clone if target and defer relation are the same (#1198)

Author: colin-rogers-dbt

dbt-adapters/.changes/unreleased/Fixes-20250714-103018.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Skip cloning objects if the target and defferal are the same
+time: 2025-07-14T10:30:18.802612-07:00
+custom:
+  Author: colin-rogers-dbt
+  Issue: "1198"

dbt-adapters/src/dbt/include/global_project/macros/materializations/models/clone/clone.sql

@@ -32,15 +32,15 @@
       {% endif %}
 
       -- as a general rule, data platforms that can clone tables can also do atomic 'create or replace'
-      {% call statement('main') %}
-          {% if target_relation and defer_relation and target_relation == defer_relation %}
-              {{ log("Target relation and defer relation are the same, skipping clone for relation: " ~ target_relation.render()) }}
-          {% else %}
-              {{ create_or_replace_clone(target_relation, defer_relation) }}
-          {% endif %}
-
-      {% endcall %}
-
+      {% if target_relation.database == defer_relation.database and
+            target_relation.schema == defer_relation.schema and
+            target_relation.identifier == defer_relation.identifier %}
+        {{ log("Target relation and defer relation are the same, skipping clone for relation: " ~ target_relation.render()) }}
+      {% else %}
+        {% call statement('main') %}
+            {{ create_or_replace_clone(target_relation, defer_relation) }}
+        {% endcall %}
+      {% endif %}
       {% set should_revoke = should_revoke(existing_relation, full_refresh_mode=True) %}
       {% do apply_grants(target_relation, grant_config, should_revoke=should_revoke) %}
       {% do persist_docs(target_relation, model) %}

dbt-bigquery/tests/functional/adapter/dbt_clone/test_dbt_clone.py

@@ -1,5 +1,9 @@
 import pytest
-from dbt.tests.adapter.dbt_clone.test_dbt_clone import BaseClonePossible
+from dbt.tests.adapter.dbt_clone.test_dbt_clone import (
+    BaseClonePossible,
+    BaseCloneSameSourceAndTarget,
+    BaseCloneNotPossible,
+)
 
 
 class TestBigQueryClonePossible(BaseClonePossible):
@@ -18,3 +22,11 @@ def clean_up(self, project):
             project.adapter.drop_schema(relation)
 
     pass
+
+
+class TestBigQueryCloneSameSourceAndTarget(BaseCloneSameSourceAndTarget):
+    pass
+
+
+class TestBigQueryCloneNotPossible(BaseCloneNotPossible):
+    pass

dbt-snowflake/tests/functional/adapter/dbt_clone/test_dbt_clone.py

@@ -3,7 +3,10 @@
 import os
 from copy import deepcopy
 from dbt.tests.util import run_dbt
-from dbt.tests.adapter.dbt_clone.test_dbt_clone import BaseClonePossible
+from dbt.tests.adapter.dbt_clone.test_dbt_clone import (
+    BaseClonePossible,
+    BaseCloneSameSourceAndTarget,
+)
 
 
 class TestSnowflakeClonePossible(BaseClonePossible):
@@ -80,3 +83,7 @@ def test_can_clone_transient_table(self, project, other_schema):
 
         results = run_dbt(clone_args)
         assert len(results) == 1
+
+
+class TestSnowflakeCloneSameSourceAndTarget(BaseCloneSameSourceAndTarget):
+    pass

dbt-tests-adapter/.changes/unreleased/Features-20250714-104451.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add test for scenario where soure/target relations are the same
+time: 2025-07-14T10:44:51.045547-07:00
+custom:
+  Author: colin-rogers-dbt
+  Issue: "1198"

dbt-tests-adapter/src/dbt/tests/adapter/dbt_clone/fixtures.py

@@ -104,3 +104,50 @@
     {{ return(False) }}
 {% endmacro %}
 """
+source_table_sql = """
+{{ config(materialized='table') }}
+select 1 as id, 'test_data' as name
+union all
+select 2 as id, 'more_data' as name
+"""
+
+source_based_model_sql = """
+{{ config(materialized='table') }}
+select * from {{ source('test_source', 'source_table') }}
+"""
+
+source_schema_yml = """
+version: 2
+sources:
+  - name: test_source
+    schema: "{{ target.schema }}"
+    tables:
+      - name: source_table
+models:
+  - name: source_based_model
+    columns:
+      - name: id
+        data_tests:
+          - not_null
+          - unique
+      - name: name
+        data_tests:
+          - not_null
+"""
+
+source_based_model_snapshot_sql = """
+{% snapshot source_based_model_snapshot %}
+
+    {{
+        config(
+            target_database=database,
+            target_schema=schema,
+            unique_key='id',
+            strategy='check',
+            check_cols=['id'],
+        )
+    }}
+    select * from {{ ref('source_based_model') }}
+
+{% endsnapshot %}
+"""

dbt-tests-adapter/src/dbt/tests/adapter/dbt_clone/test_dbt_clone.py

@@ -186,6 +186,48 @@ def test_can_clone_false(self, project, unique_schema, other_schema):
         assert all("no-op" in r.message.lower() for r in results)
 
 
+class BaseCloneSameSourceAndTarget(BaseClone):
+    def models(self):
+        return {
+            "source_based_model.sql": fixtures.source_based_model_sql,
+            "source_schema.yml": fixtures.source_schema_yml,
+        }
+
+    def snapshots(self):
+        return {
+            "snapshot_model.sql": fixtures.source_based_model_snapshot_sql,
+        }
+
+    def test_clone_same_source_and_target(self, project, unique_schema):
+        """Test that cloning a table to itself is handled gracefully"""
+        # Create a source table first
+        project.run_sql(f"DROP TABLE IF EXISTS {project.database}.{unique_schema}.source_table")
+        project.run_sql(
+            f"""
+            CREATE TABLE {project.database}.{unique_schema}.source_table AS
+            SELECT 1 as id, 'test_data' as name
+            UNION ALL
+            SELECT 2 as id, 'more_data' as name
+        """
+        )
+
+        # Run dbt to create the source-based model
+        run_dbt(["seed"])
+        run_dbt(["run"])
+
+        # Save state
+        self.copy_state(project.project_root)
+
+        # Attempt to clone source_based_model to itself with --full-refresh
+        # This should not fail but should log a skip message
+        clone_args = ["clone", "--state", "state", "--full-refresh", "--log-level", "debug"]
+
+        _, output = run_dbt_and_capture(clone_args)
+
+        # Verify the skip message is logged
+        assert "skipping clone for relation" in output
+
+
 class TestPostgresCloneNotPossible(BaseCloneNotPossible):
     @pytest.fixture(autouse=True)
     def clean_up(self, project):