update dynamic table config to include transient

dbt/adapters/snowflake/relation.py

@@ -56,6 +56,7 @@ class SnowflakeRelation(BaseRelation):
             }
         )
     )
+    transient: Optional[bool] = None
 
     @property
     def is_dynamic_table(self) -> bool:
@@ -147,7 +148,9 @@ def can_be_renamed(self) -> bool:
         """
         return self.type in self.renameable_relations and not self.is_iceberg_format
 
-    def get_ddl_prefix_for_create(self, config: RelationConfig, temporary: bool) -> str:
+    def get_ddl_prefix_for_create(
+        self, config: RelationConfig, temporary: bool, transient_default: bool = True
+    ) -> str:
         """
         This macro renders the appropriate DDL prefix during the create_table_as
         macro. It decides based on mutually exclusive table configuration options:
@@ -187,7 +190,7 @@ def get_ddl_prefix_for_create(self, config: RelationConfig, temporary: bool) ->
 
         # Always supply transient on table create DDL unless user specifically sets
         # transient to false or unset. Might as well update the object attribute too!
-        elif transient_explicitly_set_true or config.get("transient", True):
+        elif transient_explicitly_set_true or config.get("transient", transient_default):
             return "transient"
         else:
             return ""

dbt/adapters/snowflake/relation_configs/dynamic_table.py

@@ -63,6 +63,7 @@ class SnowflakeDynamicTableConfig(SnowflakeRelationConfigBase):
     catalog: SnowflakeCatalogConfig
     refresh_mode: Optional[RefreshMode] = RefreshMode.default()
     initialize: Optional[Initialize] = Initialize.default()
+    transient: Optional[bool] = False
 
     @classmethod
     def from_dict(cls, config_dict: Dict[str, Any]) -> Self:
@@ -78,6 +79,7 @@ def from_dict(cls, config_dict: Dict[str, Any]) -> Self:
             "catalog": SnowflakeCatalogConfig.from_dict(config_dict["catalog"]),
             "refresh_mode": config_dict.get("refresh_mode"),
             "initialize": config_dict.get("initialize"),
+            "transient": config_dict.get("transient"),
         }
 
         return super().from_dict(kwargs_dict)
@@ -92,6 +94,7 @@ def parse_relation_config(cls, relation_config: RelationConfig) -> Dict[str, Any
             "target_lag": relation_config.config.extra.get("target_lag"),
             "snowflake_warehouse": relation_config.config.extra.get("snowflake_warehouse"),
             "catalog": SnowflakeCatalogConfig.parse_relation_config(relation_config),
+            "transient": relation_config.config.extra.get("transient", False),
         }
 
         if refresh_mode := relation_config.config.extra.get("refresh_mode"):
@@ -105,7 +108,6 @@ def parse_relation_config(cls, relation_config: RelationConfig) -> Dict[str, Any
     @classmethod
     def parse_relation_results(cls, relation_results: RelationResults) -> Dict[str, Any]:
         dynamic_table: "agate.Row" = relation_results["dynamic_table"].rows[0]
-
         config_dict = {
             "name": dynamic_table.get("name"),
             "schema_name": dynamic_table.get("schema_name"),

dbt/include/snowflake/macros/relations/dynamic_table/create.sql

@@ -41,7 +41,7 @@
 --      A valid DDL statement which will result in a new dynamic standard table.
 -#}
 
-    {%- set materialization_prefix = relation.get_ddl_prefix_for_create(config.model.config, False) -%}
+    {%- set materialization_prefix = relation.get_ddl_prefix_for_create(config.model.config, False, False) -%}
 
     create {{ materialization_prefix }} dynamic table {{ relation }}
         target_lag = '{{ dynamic_table.target_lag }}'

dbt/include/snowflake/macros/relations/dynamic_table/replace.sql

@@ -40,7 +40,7 @@
 --      A valid DDL statement which will result in a new dynamic standard table.
 -#}
 
-    {%- set materialization_prefix = relation.get_ddl_prefix_for_create(config.model.config, False) -%}
+    {%- set materialization_prefix = relation.get_ddl_prefix_for_create(config.model.config, False, False) -%}
 
     create or replace {{ materialization_prefix }} dynamic table {{ relation }}
         target_lag = '{{ dynamic_table.target_lag }}'

tests/functional/relation_tests/dynamic_table_tests/models.py

@@ -17,6 +17,18 @@
 """
 
 
+DYNAMIC_TRANSIENT_TABLE = """
+{{ config(
+    materialized='dynamic_table',
+    snowflake_warehouse='DBT_TESTING',
+    target_lag='2 minutes',
+    refresh_mode='INCREMENTAL',
+    transient=True,
+) }}
+select * from {{ ref('my_seed') }}
+"""
+
+
 DYNAMIC_TABLE_DOWNSTREAM = """
 {{ config(
     materialized='dynamic_table',

tests/functional/relation_tests/dynamic_table_tests/test_basic.py

@@ -18,6 +18,7 @@ def models(self):
         my_models = {
             "my_dynamic_table.sql": models.DYNAMIC_TABLE,
             "my_dynamic_table_downstream.sql": models.DYNAMIC_TABLE_DOWNSTREAM,
+            "my_dynamic_transient_table.sql": models.DYNAMIC_TRANSIENT_TABLE,
         }
         if self.iceberg:
             my_models.update(
@@ -36,6 +37,9 @@ def test_dynamic_table_full_refresh(self, project):
         run_dbt(["run", "--full-refresh"])
         assert query_relation_type(project, "my_dynamic_table") == "dynamic_table"
         assert query_relation_type(project, "my_dynamic_table_downstream") == "dynamic_table"
+        assert (
+            query_relation_type(project, "my_dynamic_transient_table") == "dynamic_table_transient"
+        )
         if self.iceberg:
             assert query_relation_type(project, "my_dynamic_iceberg_table") == "dynamic_table"
 

tests/functional/utils.py

@@ -16,6 +16,7 @@ def query_relation_type(project, name: str) -> Optional[str]:
         select
             case table_type
                 when 'BASE TABLE' then iff(is_dynamic = 'YES', 'dynamic_table', 'table')
+                                    || iff(is_transient = 'YES', '_transient', '')
                 when 'VIEW' then 'view'
                 when 'EXTERNAL TABLE' then 'external_table'
             end as relation_type
@@ -25,7 +26,6 @@ def query_relation_type(project, name: str) -> Optional[str]:
         and table_catalog like '{relation.database.upper()}'
     """
     results = project.run_sql(sql, fetch="all")
-
     assert len(results) > 0, f"Relation {relation} not found"
     assert len(results) == 1, f"Multiple relations found"