Fix refresh_strategy = auto semantics for dynamic tables (#1268)

* Add failing test.

* Parametrize cases

* Change comparison logic for when to rebuild if DT has auto refresh strategy (implicit or explicit).

* Add changelog

---------

Co-authored-by: VersusFacit <[email protected]>

Author: VersusFacit

.changes/unreleased/Fixes-20241209-131530.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: AUTO should no longer lead to rebuilds of dynamic tables.
+time: 2024-12-09T13:15:30.554566-08:00
+custom:
+    Author: versusfacit
+    Issue: "1267"

dbt/adapters/snowflake/relation.py

@@ -17,6 +17,7 @@
 from dbt_common.events.functions import fire_event, warn_or_error
 
 from dbt.adapters.snowflake.relation_configs import (
+    RefreshMode,
     SnowflakeCatalogConfigChange,
     SnowflakeDynamicTableConfig,
     SnowflakeDynamicTableConfigChangeset,
@@ -109,7 +110,10 @@ def dynamic_table_config_changeset(
                 )
             )
 
-        if new_dynamic_table.refresh_mode != existing_dynamic_table.refresh_mode:
+        if (
+            new_dynamic_table.refresh_mode != RefreshMode.AUTO
+            and new_dynamic_table.refresh_mode != existing_dynamic_table.refresh_mode
+        ):
             config_change_collection.refresh_mode = SnowflakeDynamicTableRefreshModeConfigChange(
                 action=RelationConfigChangeAction.create,
                 context=new_dynamic_table.refresh_mode,

dbt/adapters/snowflake/relation_configs/__init__.py

@@ -3,6 +3,7 @@
     SnowflakeCatalogConfigChange,
 )
 from dbt.adapters.snowflake.relation_configs.dynamic_table import (
+    RefreshMode,
     SnowflakeDynamicTableConfig,
     SnowflakeDynamicTableConfigChangeset,
     SnowflakeDynamicTableRefreshModeConfigChange,

tests/functional/relation_tests/dynamic_table_tests/models.py

@@ -17,6 +17,26 @@
 """
 
 
+EXPLICIT_AUTO_DYNAMIC_TABLE = """
+{{ config(
+    materialized='dynamic_table',
+    snowflake_warehouse='DBT_TESTING',
+    target_lag='2 minutes',
+    refresh_mode='AUTO',
+) }}
+select * from {{ ref('my_seed') }}
+"""
+
+IMPLICIT_AUTO_DYNAMIC_TABLE = """
+{{ config(
+    materialized='dynamic_table',
+    snowflake_warehouse='DBT_TESTING',
+    target_lag='2 minutes',
+) }}
+select * from {{ ref('my_seed') }}
+"""
+
+
 DYNAMIC_TABLE_DOWNSTREAM = """
 {{ config(
     materialized='dynamic_table',

tests/functional/relation_tests/dynamic_table_tests/test_basic.py

@@ -1,6 +1,6 @@
 import pytest
 
-from dbt.tests.util import run_dbt
+from dbt.tests.util import assert_message_in_logs, run_dbt, run_dbt_and_capture
 
 from tests.functional.relation_tests.dynamic_table_tests import models
 from tests.functional.utils import query_relation_type
@@ -46,3 +46,44 @@ class TestBasicIcebergOn(TestBasic):
     @pytest.fixture(scope="class")
     def project_config_update(self):
         return {"flags": {"enable_iceberg_materializations": True}}
+
+
+class TestAutoConfigDoesntFullRefresh:
+    """
+    AUTO refresh_strategy will be compared accurately with both INCREMENTAL and FULL.
+    https://github.com/dbt-labs/dbt-snowflake/issues/1267
+    """
+
+    DT_NAME = "my_dynamic_table"
+
+    @pytest.fixture(scope="class", autouse=True)
+    def seeds(self):
+        return {"my_seed.csv": models.SEED}
+
+    @pytest.fixture(scope="class", autouse=True)
+    def models(self):
+        yield {
+            f"explicit_{self.DT_NAME}.sql": models.EXPLICIT_AUTO_DYNAMIC_TABLE,
+            f"implicit_{self.DT_NAME}.sql": models.IMPLICIT_AUTO_DYNAMIC_TABLE,
+        }
+
+    @pytest.mark.parametrize("test_dt", [f"explicit_{DT_NAME}", f"implicit_{DT_NAME}"])
+    def test_auto_config_doesnt_full_refresh(self, project, test_dt):
+        model_qualified_name = f"{project.database}.{project.test_schema}.{test_dt}"
+
+        run_dbt(["seed"])
+        _, logs = run_dbt_and_capture(["--debug", "run", "--select", f"{test_dt}.sql"])
+        assert_message_in_logs(f"create dynamic table {model_qualified_name}", logs)
+        assert_message_in_logs("refresh_mode = AUTO", logs)
+
+        _, logs = run_dbt_and_capture(["--debug", "run", "--select", f"{test_dt}.sql"])
+
+        assert_message_in_logs(f"create dynamic table {model_qualified_name}", logs, False)
+        assert_message_in_logs(
+            f"create or replace dynamic table {model_qualified_name}", logs, False
+        )
+        assert_message_in_logs("refresh_mode = AUTO", logs, False)
+        assert_message_in_logs(
+            f"No configuration changes were identified on: `{model_qualified_name}`. Continuing.",
+            logs,
+        )