[Backport 1.6.latest] [ADAP-774] fix snowflake_warehouse issue with dynamic tables

dbt/adapters/snowflake/relation.py

@@ -50,10 +50,12 @@ def dynamic_table_config_changeset(
                 context=new_dynamic_table.target_lag,
             )
 
-        if new_dynamic_table.warehouse != existing_dynamic_table.warehouse:
-            config_change_collection.warehouse = SnowflakeDynamicTableWarehouseConfigChange(
-                action=RelationConfigChangeAction.alter,
-                context=new_dynamic_table.warehouse,
+        if new_dynamic_table.snowflake_warehouse != existing_dynamic_table.snowflake_warehouse:
+            config_change_collection.snowflake_warehouse = (
+                SnowflakeDynamicTableWarehouseConfigChange(
+                    action=RelationConfigChangeAction.alter,
+                    context=new_dynamic_table.snowflake_warehouse,
+                )
             )
 
         if config_change_collection.has_changes:

dbt/adapters/snowflake/relation_configs/dynamic_table.py

@@ -23,7 +23,7 @@ class SnowflakeDynamicTableConfig(SnowflakeRelationConfigBase):
     - name: name of the dynamic table
     - query: the query behind the table
     - target_lag: the maximum amount of time that the dynamic table’s content should lag behind updates to the base tables
-    - warehouse: the name of the warehouse that provides the compute resources for refreshing the dynamic table
+    - snowflake_warehouse: the name of the warehouse that provides the compute resources for refreshing the dynamic table
 
     There are currently no non-configurable parameters.
     """
@@ -33,7 +33,7 @@ class SnowflakeDynamicTableConfig(SnowflakeRelationConfigBase):
     database_name: str
     query: str
     target_lag: SnowflakeDynamicTableTargetLagConfig
-    warehouse: str
+    snowflake_warehouse: str
 
     @classmethod
     def from_dict(cls, config_dict) -> "SnowflakeDynamicTableConfig":
@@ -44,7 +44,7 @@ def from_dict(cls, config_dict) -> "SnowflakeDynamicTableConfig":
                 ComponentName.Database, config_dict.get("database_name")
             ),
             "query": config_dict.get("query"),
-            "warehouse": config_dict.get("warehouse"),
+            "snowflake_warehouse": config_dict.get("snowflake_warehouse"),
         }
 
         if target_lag := config_dict.get("target_lag"):
@@ -62,7 +62,7 @@ def parse_model_node(cls, model_node: ModelNode) -> dict:
             "schema_name": model_node.schema,
             "database_name": model_node.database,
             "query": model_node.compiled_code,
-            "warehouse": model_node.config.extra.get("snowflake_warehouse"),
+            "snowflake_warehouse": model_node.config.extra.get("snowflake_warehouse"),
         }
 
         if model_node.config.extra.get("target_lag"):
@@ -81,7 +81,7 @@ def parse_relation_results(cls, relation_results: RelationResults) -> dict:
             "schema_name": dynamic_table.get("schema_name"),
             "database_name": dynamic_table.get("database_name"),
             "query": dynamic_table.get("text"),
-            "warehouse": dynamic_table.get("warehouse"),
+            "snowflake_warehouse": dynamic_table.get("warehouse"),
         }
 
         if dynamic_table.get("target_lag"):
@@ -108,17 +108,19 @@ def requires_full_refresh(self) -> bool:
 @dataclass
 class SnowflakeDynamicTableConfigChangeset:
     target_lag: Optional[SnowflakeDynamicTableTargetLagConfigChange] = None
-    warehouse: Optional[SnowflakeDynamicTableWarehouseConfigChange] = None
+    snowflake_warehouse: Optional[SnowflakeDynamicTableWarehouseConfigChange] = None
 
     @property
     def requires_full_refresh(self) -> bool:
         return any(
             [
                 self.target_lag.requires_full_refresh if self.target_lag else False,
-                self.warehouse.requires_full_refresh if self.warehouse else False,
+                self.snowflake_warehouse.requires_full_refresh
+                if self.snowflake_warehouse
+                else False,
             ]
         )
 
     @property
     def has_changes(self) -> bool:
-        return any([self.target_lag, self.warehouse])
+        return any([self.target_lag, self.snowflake_warehouse])

dbt/include/snowflake/macros/materializations/dynamic_table/ddl.sql

@@ -15,12 +15,12 @@
 
         {%- set target_lag = configuration_changes.target_lag -%}
         {%- if target_lag -%}{{- log('Applying UPDATE TARGET_LAG to: ' ~ existing_relation) -}}{%- endif -%}
-        {%- set warehouse = configuration_changes.warehouse -%}
-        {%- if warehouse -%}{{- log('Applying UPDATE WAREHOUSE to: ' ~ existing_relation) -}}{%- endif -%}
+        {%- set snowflake_warehouse = configuration_changes.snowflake_warehouse -%}
+        {%- if snowflake_warehouse -%}{{- log('Applying UPDATE WAREHOUSE to: ' ~ existing_relation) -}}{%- endif -%}
 
         alter dynamic table {{ existing_relation }} set
             {% if target_lag %}target_lag = '{{ target_lag.context }}'{% endif %}
-            {% if warehouse %}warehouse = {{ warehouse.context }}{% endif %}
+            {% if snowflake_warehouse %}warehouse = {{ snowflake_warehouse.context }}{% endif %}
 
     {%- endif -%}