feat(snowflake) Add COPY TAGS support mirroring COPY GRANTS behavior

dbt-snowflake/.changes/unreleased/Features-20260413-120000.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add support for the Snowflake COPY TAGS parameter in CREATE OR REPLACE TABLE, clone operations, and built-in Iceberg table creation
+time: 2026-04-13T12:00:00.000000+00:00
+custom:
+  Author: yostrinsky
+  Issue: "1857"

dbt-snowflake/src/dbt/adapters/snowflake/impl.py

@@ -62,6 +62,7 @@ class SnowflakeConfig(AdapterConfig):
     automatic_clustering: Optional[bool] = None
     secure: Optional[bool] = None
     copy_grants: Optional[bool] = None
+    copy_tags: Optional[bool] = None
     snowflake_warehouse: Optional[str] = None
     snowflake_initialization_warehouse: Optional[str] = None
     refresh_warehouse: Optional[str] = None

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

@@ -8,4 +8,5 @@
       table {{ this_relation }}
       clone {{ defer_relation }}
       {{ "copy grants" if config.get("copy_grants", false) }}
+      {{ "copy tags" if config.get("copy_tags", false) }}
 {% endmacro %}

dbt-snowflake/src/dbt/include/snowflake/macros/relations/table/create.sql

@@ -109,6 +109,7 @@ as (
 {%- endif -%}
 
 {%- set copy_grants = config.get('copy_grants', default=false) -%}
+{%- set copy_tags = config.get('copy_tags', default=false) -%}
 
 {%- set row_access_policy = config.get('row_access_policy', default=none) -%}
 {%- set table_tag = config.get('table_tag', default=none) -%}
@@ -128,6 +129,7 @@ create or replace {{ transient }}table {{ relation }}
     {{ get_table_columns_and_constraints() }}
     {%- endif %}
     {% if copy_grants -%} copy grants {%- endif %}
+    {% if copy_tags -%} copy tags {%- endif %}
     {% if row_access_policy -%} with row access policy {{ row_access_policy }} {%- endif %}
     {% if table_tag -%} with tag ({{ table_tag }}) {%- endif %}
     {{ optional('change_tracking', catalog_relation.change_tracking)}}
@@ -175,6 +177,7 @@ alter table {{ relation }} resume recluster;
 {%- endif -%}
 
 {%- set copy_grants = config.get('copy_grants', default=false) -%}
+{%- set copy_tags = config.get('copy_tags', default=false) -%}
 
 {%- set row_access_policy = config.get('row_access_policy', default=none) -%}
 {%- set table_tag = config.get('table_tag', default=none) -%}
@@ -204,6 +207,7 @@ create or replace iceberg table {{ relation }}
     {% if row_access_policy -%} with row access policy {{ row_access_policy }} {%- endif %}
     {% if table_tag -%} with tag ({{ table_tag }}) {%- endif %}
     {% if copy_grants -%} copy grants {%- endif %}
+    {% if copy_tags -%} copy tags {%- endif %}
 as (
     {%- if catalog_relation.cluster_by is not none -%}
     select * from (

dbt-snowflake/tests/functional/adapter/test_copy_tags.py

@@ -0,0 +1,86 @@
+import re
+
+import pytest
+
+from dbt.tests.util import run_dbt
+
+
+def get_cleaned_model_ddl_from_file(file_name: str) -> str:
+    with open(f"target/run/test/models/{file_name}", "r") as ddl_file:
+        return re.sub(r"\s+", " ", ddl_file.read())
+
+
+_MODEL_TABLE_COPY_TAGS = """
+{{ config(
+    materialized = 'table',
+    copy_tags = true,
+) }}
+
+select 1 as id
+"""
+
+_MODEL_TABLE_COPY_TAGS_AND_GRANTS = """
+{{ config(
+    materialized = 'table',
+    copy_grants = true,
+    copy_tags = true,
+) }}
+
+select 1 as id
+"""
+
+_MODEL_TABLE_NO_COPY_TAGS = """
+{{ config(
+    materialized = 'table',
+) }}
+
+select 1 as id
+"""
+
+_MODEL_INCREMENTAL_COPY_TAGS = """
+{{ config(
+    materialized = 'incremental',
+    copy_tags = true,
+) }}
+
+select 1 as id
+"""
+
+_DDL_COPY_TAGS = "copy tags"
+_DDL_COPY_GRANTS = "copy grants"
+
+
+class TestCopyTags:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "table_copy_tags.sql": _MODEL_TABLE_COPY_TAGS,
+            "table_copy_tags_and_grants.sql": _MODEL_TABLE_COPY_TAGS_AND_GRANTS,
+            "table_no_copy_tags.sql": _MODEL_TABLE_NO_COPY_TAGS,
+            "incremental_copy_tags.sql": _MODEL_INCREMENTAL_COPY_TAGS,
+        }
+
+    def test_copy_tags(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 4
+
+        table_copy_tags_ddl = get_cleaned_model_ddl_from_file("table_copy_tags.sql")
+        assert _DDL_COPY_TAGS in table_copy_tags_ddl
+        assert _DDL_COPY_GRANTS not in table_copy_tags_ddl
+
+        table_both_ddl = get_cleaned_model_ddl_from_file("table_copy_tags_and_grants.sql")
+        assert _DDL_COPY_TAGS in table_both_ddl
+        assert _DDL_COPY_GRANTS in table_both_ddl
+
+        table_no_copy_tags_ddl = get_cleaned_model_ddl_from_file("table_no_copy_tags.sql")
+        assert _DDL_COPY_TAGS not in table_no_copy_tags_ddl
+
+        incremental_copy_tags_ddl = get_cleaned_model_ddl_from_file("incremental_copy_tags.sql")
+        assert _DDL_COPY_TAGS in incremental_copy_tags_ddl
+
+    def test_copy_tags_full_refresh(self, project):
+        results = run_dbt(["run", "--select", "table_copy_tags", "--full-refresh"])
+        assert len(results) == 1
+
+        table_copy_tags_ddl = get_cleaned_model_ddl_from_file("table_copy_tags.sql")
+        assert _DDL_COPY_TAGS in table_copy_tags_ddl

dbt-snowflake/tests/unit/test_copy_tags.py

@@ -0,0 +1,29 @@
+from dbt.adapters.snowflake.impl import SnowflakeConfig
+
+
+class TestCopyTagsConfig:
+    def test_copy_tags_defaults_to_none(self):
+        config = SnowflakeConfig()
+        assert config.copy_tags is None
+
+    def test_copy_tags_accepts_true(self):
+        config = SnowflakeConfig(copy_tags=True)
+        assert config.copy_tags is True
+
+    def test_copy_tags_accepts_false(self):
+        config = SnowflakeConfig(copy_tags=False)
+        assert config.copy_tags is False
+
+    def test_copy_tags_independent_of_copy_grants(self):
+        config = SnowflakeConfig(copy_grants=True, copy_tags=False)
+        assert config.copy_grants is True
+        assert config.copy_tags is False
+
+        config = SnowflakeConfig(copy_grants=False, copy_tags=True)
+        assert config.copy_grants is False
+        assert config.copy_tags is True
+
+    def test_copy_tags_and_copy_grants_both_enabled(self):
+        config = SnowflakeConfig(copy_grants=True, copy_tags=True)
+        assert config.copy_grants is True
+        assert config.copy_tags is True