adding transient dynamic table capabilities

.changes/unreleased/Features-20241017-103843.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Adding capability to create transient dynamic tables.
+time: 2024-10-17T10:38:43.820283+01:00
+custom:
+    Author: jhsb25
+    Issue: "1089"

dbt/adapters/snowflake/impl.py

@@ -93,7 +93,18 @@ def _behavior_flags(self) -> List[BehaviorFlag]:
                     "prevent unnecessary latency for other users."
                 ),
                 "docs_url": "https://docs.getdbt.com/reference/resource-configs/snowflake-configs#iceberg-table-format",
-            }
+            },
+            {
+                "name": "default_dynamic_tables_to_transient",
+                "default": False,
+                "description": (
+                    "Standard Tables are created as transient by default. This can be overridden "
+                    "by setting transient: false in the model config. This saves on storages costs "
+                    "and for most users is the right decision. We are changing the default behavior for "
+                    "dynamic tables to align with this."
+                    "Read more: https://docs.getdbt.com/reference/resource-configs/snowflake-configs#transient-tables"
+                ),
+            },
         ]
 
     @classmethod

dbt/adapters/snowflake/relation.py

@@ -147,7 +147,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 +189,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

@@ -105,7 +105,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

@@ -40,8 +40,15 @@
 --  Returns:
 --      A valid DDL statement which will result in a new dynamic standard table.
 -#}
+    {%- if adapter.behavior.default_dynamic_tables_to_transient -%}
+        {%- set transient = True -%}
+    {%- else -%}
+        {%- set transient = False -%}
+    {%- endif -%}
+
+    {%- set materialization_prefix = relation.get_ddl_prefix_for_create(config.model.config, False, transient) -%}
 
-    create dynamic table {{ relation }}
+    create {{ materialization_prefix }} dynamic table {{ relation }}
         target_lag = '{{ dynamic_table.target_lag }}'
         warehouse = {{ dynamic_table.snowflake_warehouse }}
         {{ optional('refresh_mode', dynamic_table.refresh_mode) }}

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

@@ -40,7 +40,15 @@
 --      A valid DDL statement which will result in a new dynamic standard table.
 -#}
 
-    create or replace dynamic table {{ relation }}
+    {%- if adapter.behavior.default_dynamic_tables_to_transient -%}
+        {%- set transient = True -%}
+    {%- else -%}
+        {%- set transient = False -%}
+    {%- endif -%}
+
+    {%- set materialization_prefix = relation.get_ddl_prefix_for_create(config.model.config, False, transient) -%}
+
+    create or replace {{ materialization_prefix }} dynamic table {{ relation }}
         target_lag = '{{ dynamic_table.target_lag }}'
         warehouse = {{ dynamic_table.snowflake_warehouse }}
         {{ optional('refresh_mode', dynamic_table.refresh_mode) }}

tests/functional/relation_tests/dynamic_table_tests/models.py

@@ -17,12 +17,25 @@
 """
 
 
+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',
     snowflake_warehouse='DBT_TESTING',
     target_lag='DOWNSTREAM',
     refresh_mode='INCREMENTAL',
+    transient=False,
 ) }}
 select * from {{ ref('my_seed') }}
 """

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"
 
@@ -46,3 +50,18 @@ class TestBasicIcebergOn(TestBasic):
     @pytest.fixture(scope="class")
     def project_config_update(self):
         return {"flags": {"enable_iceberg_materializations": True}}
+
+
+class TestDefaultTransient(TestBasic):
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"flags": {"default_dynamic_tables_to_transient": True}}
+
+    def test_dynamic_table_full_refresh(self, project):
+        run_dbt(["run", "--full-refresh"])
+        assert (
+            query_relation_type(project, "my_dynamic_transient_table") == "dynamic_table_transient"
+        )
+        assert query_relation_type(project, "my_dynamic_table_downstream") == "dynamic_table"
+        assert query_relation_type(project, "my_dynamic_table") == "dynamic_table_transient"

tests/functional/relation_tests/test_relation_type_change.py

@@ -53,8 +53,8 @@ def uses_iceberg(self) -> bool:
 
 relations = [
     Model(models.VIEW, "view"),
-    Model(models.TABLE, "table", "default"),
-    Model(models.INCREMENTAL_TABLE, "table", "default", is_incremental=True),
+    Model(models.TABLE, "table_transient", "default"),
+    Model(models.INCREMENTAL_TABLE, "table_transient", "default", is_incremental=True),
     Model(models.DYNAMIC_TABLE, "dynamic_table", "default"),
     Model(models.ICEBERG_TABLE, "table", "iceberg"),
     Model(models.INCREMENTAL_ICEBERG_TABLE, "table", "iceberg", is_incremental=True),

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"