feat(snowflake): add transient support for dynamic tables (#1603)

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: igorbelianski-cyber

dbt-snowflake/.changes/unreleased/Breaking Changes-20260209-113822.yaml

@@ -0,0 +1,6 @@
+kind: Breaking Changes
+body: Add transient support for Snowflake dynamic tables via `transient` config option and `snowflake_default_transient_dynamic_tables` behavior flag.
+time: 2026-02-09T11:38:22.629597-08:00
+custom:
+    Author: igorbelianski-cyber
+    Issue: "716"

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

@@ -1,6 +1,6 @@
 from copy import deepcopy
 from dataclasses import dataclass
-from typing import Mapping, Any, Optional, List, Union, Dict, FrozenSet, Tuple, TYPE_CHECKING
+from typing import Mapping, Any, Optional, List, Union, Dict, FrozenSet, Tuple
 
 from dbt.adapters.base.impl import AdapterConfig, ConstraintSupport
 from dbt.adapters.base.meta import available
@@ -22,6 +22,7 @@
     CatalogTable,
     ColumnMetadata,
 )
+from dbt_common.behavior_flags import BehaviorFlag
 from dbt_common.events.functions import fire_event
 from dbt_common.exceptions import CompilationError, DbtDatabaseError, DbtRuntimeError
 from dbt_common.utils import filter_null_values
@@ -38,11 +39,19 @@
 from dbt.adapters.snowflake import SnowflakeConnectionManager
 from dbt.adapters.snowflake import SnowflakeRelation
 
-if TYPE_CHECKING:
-    import agate
+import agate
 
 SHOW_OBJECT_METADATA_MACRO_NAME = "snowflake__show_object_metadata"
 
+SNOWFLAKE_DEFAULT_TRANSIENT_DYNAMIC_TABLES = BehaviorFlag(
+    name="snowflake_default_transient_dynamic_tables",
+    default=False,
+    description=(
+        "When enabled, dynamic tables default to transient (matching regular table behavior). "
+        "This is a breaking change from previous behavior where dynamic tables were non-transient."
+    ),
+)
+
 
 @dataclass
 class SnowflakeConfig(AdapterConfig):
@@ -99,6 +108,10 @@ class SnowflakeAdapter(SQLAdapter):
         }
     )
 
+    @property
+    def _behavior_flags(self) -> list[BehaviorFlag]:
+        return [SNOWFLAKE_DEFAULT_TRANSIENT_DYNAMIC_TABLES]
+
     def __init__(self, config, mp_context) -> None:
         super().__init__(config, mp_context)
         self.add_catalog_integration(constants.DEFAULT_INFO_SCHEMA_CATALOG)
@@ -512,7 +525,9 @@ def build_catalog_relation(self, model: RelationConfig) -> Optional[CatalogRelat
         return None
 
     @available
-    def describe_dynamic_table(self, relation: SnowflakeRelation) -> Dict[str, Any]:
+    def describe_dynamic_table(
+        self, relation: SnowflakeRelation, include_transient: bool = False
+    ) -> Dict[str, Any]:
         """
         Get all relevant metadata about a dynamic table to return as a dict to Agate Table row
 
@@ -548,7 +563,35 @@ def describe_dynamic_table(self, relation: SnowflakeRelation) -> Dict[str, Any]:
         if "initialization_warehouse" in available_columns:
             base_columns.insert(base_columns.index("warehouse") + 1, "initialization_warehouse")
 
-        return {"dynamic_table": dt_table.select(base_columns)}
+        selected = dt_table.select(base_columns)
+
+        if include_transient:
+            is_transient = self._query_dynamic_table_transient_status(relation)
+            # choosing a future proof column name
+            selected = selected.compute(
+                [("transient", agate.Formula(agate.Boolean(), lambda row: is_transient))]
+            )
+
+        return {"dynamic_table": selected}
+
+    def _query_dynamic_table_transient_status(self, relation: SnowflakeRelation) -> bool:
+        """
+        Query SHOW TABLES to determine if a dynamic table is transient.
+
+        SHOW DYNAMIC TABLES does not expose transient status, so we fall back to
+        SHOW TABLES where the "kind" column contains "TRANSIENT" for transient tables.
+        """
+        quoting = relation.quote_policy
+        schema = f'"{relation.schema}"' if quoting.schema else relation.schema
+        database = f'"{relation.database}"' if quoting.database else relation.database
+        show_tables_sql = f"show tables like '{relation.identifier}' in schema {database}.{schema}"
+        _, tables_table = self.execute(show_tables_sql, fetch=True)
+        if len(tables_table.rows) > 0:
+            tables_table = tables_table.rename(
+                column_names=[name.lower() for name in tables_table.column_names]
+            )
+            return tables_table.rows[0].get("kind") == "TRANSIENT"
+        return False
 
     def expand_column_types(self, goal, current):
         reference_columns = {c.name: c for c in self.get_columns_in_relation(goal)}

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

@@ -26,6 +26,7 @@
     SnowflakeDynamicTableWarehouseConfigChange,
     SnowflakeDynamicTableImmutableWhereConfigChange,
     SnowflakeDynamicTableClusterByConfigChange,
+    SnowflakeDynamicTableTransientConfigChange,
     SnowflakeQuotePolicy,
     SnowflakeRelationType,
 )
@@ -93,7 +94,9 @@ def from_config(cls, config: RelationConfig) -> RelationConfigBase:
 
     @classmethod
     def dynamic_table_config_changeset(
-        cls, relation_results: RelationResults, relation_config: RelationConfig
+        cls,
+        relation_results: RelationResults,
+        relation_config: RelationConfig,
     ) -> Optional[SnowflakeDynamicTableConfigChangeset]:
         existing_dynamic_table = SnowflakeDynamicTableConfig.from_relation_results(
             relation_results
@@ -150,6 +153,19 @@ def dynamic_table_config_changeset(
                 context=new_dynamic_table.cluster_by,
             )
 
+        # Transient is only compared when both sides are explicitly known:
+        # - new is None when the user omitted transient from their config ("don't care")
+        # - existing is None when describe_dynamic_table was called without include_transient
+        if (
+            new_dynamic_table.transient is not None
+            and existing_dynamic_table.transient is not None
+            and new_dynamic_table.transient != existing_dynamic_table.transient
+        ):
+            config_change_collection.transient = SnowflakeDynamicTableTransientConfigChange(
+                action=RelationConfigChangeAction.create,  # type: ignore
+                context=new_dynamic_table.transient,
+            )
+
         if config_change_collection.has_changes:
             return config_change_collection
         return None

dbt-snowflake/src/dbt/adapters/snowflake/relation_configs/__init__.py

@@ -8,6 +8,7 @@
     SnowflakeDynamicTableTargetLagConfigChange,
     SnowflakeDynamicTableImmutableWhereConfigChange,
     SnowflakeDynamicTableClusterByConfigChange,
+    SnowflakeDynamicTableTransientConfigChange,
 )
 from dbt.adapters.snowflake.relation_configs.policies import (
     SnowflakeIncludePolicy,

dbt-snowflake/src/dbt/adapters/snowflake/relation_configs/dynamic_table.py

@@ -49,6 +49,7 @@ class SnowflakeDynamicTableConfig(SnowflakeRelationConfigBase):
     - initialize: specifies the behavior of the initial refresh of the dynamic table
     - cluster_by: specifies the columns to cluster on
     - immutable_where: specifies an immutability constraint expression
+    - transient: specifies whether the dynamic table is transient (no fail-safe). snowflake_default_transient_dynamic_tables determines the default value
 
     There are currently no non-configurable parameters.
     """
@@ -66,6 +67,7 @@ class SnowflakeDynamicTableConfig(SnowflakeRelationConfigBase):
     table_tag: Optional[str] = None
     cluster_by: Optional[Union[str, list[str]]] = None
     immutable_where: Optional[str] = None
+    transient: Optional[bool] = None
 
     @classmethod
     def from_dict(cls, config_dict: Dict[str, Any]) -> Self:
@@ -91,6 +93,7 @@ def from_dict(cls, config_dict: Dict[str, Any]) -> Self:
             "table_tag": config_dict.get("table_tag"),
             "cluster_by": config_dict.get("cluster_by"),
             "immutable_where": config_dict.get("immutable_where"),
+            "transient": config_dict.get("transient"),
         }
 
         return super().from_dict(kwargs_dict)  # type:ignore
@@ -117,6 +120,7 @@ def parse_relation_config(cls, relation_config: RelationConfig) -> Dict[str, Any
             "immutable_where": relation_config.config.extra.get(  # type:ignore
                 "immutable_where"
             ),
+            "transient": relation_config.config.extra.get("transient"),  # type:ignore
         }
 
         if refresh_mode := relation_config.config.extra.get("refresh_mode"):  # type:ignore
@@ -172,6 +176,9 @@ def parse_relation_results(cls, relation_results: RelationResults) -> Dict[str,
             "table_tag": dynamic_table.get("table_tag"),
             "cluster_by": cluster_by,
             "immutable_where": immutable_where,
+            # agate.Row.get() returns None when the column is absent, which is the
+            # correct default -- it means "not queried" and skips transient comparison.
+            "transient": dynamic_table.get("transient"),
             # we don't get initialize since that's a one-time scheduler attribute, not a DT attribute
         }
 
@@ -232,6 +239,16 @@ def requires_full_refresh(self) -> bool:
         return False
 
 
+@dataclass(frozen=True, eq=True, unsafe_hash=True)
+class SnowflakeDynamicTableTransientConfigChange(RelationConfigChange):
+    context: Optional[bool] = None
+
+    @property
+    def requires_full_refresh(self) -> bool:
+        # Transient cannot be changed via ALTER, requires full table recreation
+        return True
+
+
 @dataclass
 class SnowflakeDynamicTableConfigChangeset:
     target_lag: Optional[SnowflakeDynamicTableTargetLagConfigChange] = None
@@ -242,6 +259,7 @@ class SnowflakeDynamicTableConfigChangeset:
     refresh_mode: Optional[SnowflakeDynamicTableRefreshModeConfigChange] = None
     immutable_where: Optional[SnowflakeDynamicTableImmutableWhereConfigChange] = None
     cluster_by: Optional[SnowflakeDynamicTableClusterByConfigChange] = None
+    transient: Optional[SnowflakeDynamicTableTransientConfigChange] = None
 
     @property
     def requires_full_refresh(self) -> bool:
@@ -261,6 +279,7 @@ def requires_full_refresh(self) -> bool:
                 self.refresh_mode.requires_full_refresh if self.refresh_mode else False,
                 self.immutable_where.requires_full_refresh if self.immutable_where else False,
                 self.cluster_by.requires_full_refresh if self.cluster_by else False,
+                self.transient.requires_full_refresh if self.transient else False,
             ]
         )
 
@@ -274,5 +293,6 @@ def has_changes(self) -> bool:
                 self.refresh_mode,
                 self.immutable_where,
                 self.cluster_by,
+                self.transient,
             ]
         )

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

@@ -91,7 +91,8 @@
 
 
 {% macro snowflake__get_dynamic_table_configuration_changes(existing_relation, new_config) -%}
-    {% set _existing_dynamic_table = snowflake__describe_dynamic_table(existing_relation) %}
+    {% set _include_transient = new_config.get("transient") is not none %}
+    {% set _existing_dynamic_table = adapter.describe_dynamic_table(existing_relation, _include_transient) %}
     {% set _configuration_changes = existing_relation.dynamic_table_config_changeset(_existing_dynamic_table, new_config.model) %}
     {% do return(_configuration_changes) %}
 {%- endmacro %}

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

@@ -31,7 +31,16 @@
         A valid DDL statement which will result in a new dynamic info schema table.
 -#}
 
-    create dynamic table {{ relation }}
+    {#- Determine transient: explicit config takes precedence, otherwise use behavior flag default -#}
+    {%- if dynamic_table.transient is not none -%}
+        {%- set is_transient = dynamic_table.transient -%}
+    {%- elif adapter.behavior.snowflake_default_transient_dynamic_tables.no_warn -%}
+        {%- set is_transient = true -%}
+    {%- else -%}
+        {%- set is_transient = false -%}
+    {%- endif -%}
+    {%- set transient_keyword = 'transient ' if is_transient else '' -%}
+    create {{ transient_keyword }}dynamic table {{ relation }}
         target_lag = '{{ dynamic_table.target_lag }}'
         warehouse = {{ dynamic_table.snowflake_warehouse }}
         {{ optional('initialization_warehouse', dynamic_table.snowflake_initialization_warehouse) }}

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

@@ -44,7 +44,16 @@
         A valid DDL statement which will result in a new dynamic info schema table.
 -#}
 
-create or replace dynamic table {{ relation }}
+    {#- Determine transient: explicit config takes precedence, otherwise use behavior flag default -#}
+    {%- if dynamic_table.transient is not none -%}
+        {%- set is_transient = dynamic_table.transient -%}
+    {%- elif adapter.behavior.snowflake_default_transient_dynamic_tables.no_warn -%}
+        {%- set is_transient = true -%}
+    {%- else -%}
+        {%- set is_transient = false -%}
+    {%- endif -%}
+    {%- set transient_keyword = 'transient ' if is_transient else '' -%}
+create or replace {{ transient_keyword }}dynamic table {{ relation }}
     target_lag = '{{ dynamic_table.target_lag }}'
     warehouse = {{ dynamic_table.snowflake_warehouse }}
     {{ optional('initialization_warehouse', dynamic_table.snowflake_initialization_warehouse) }}

dbt-snowflake/tests/functional/relation_tests/dynamic_table_tests/models.py

@@ -308,3 +308,40 @@
 ) }}
 select id, "NONE" from {{ ref('my_seed_none') }}
 """
+
+
+# Transient dynamic table fixtures
+DYNAMIC_TABLE_TRANSIENT = """
+{{ config(
+    materialized='dynamic_table',
+    snowflake_warehouse='DBT_TESTING',
+    target_lag='2 minutes',
+    refresh_mode='INCREMENTAL',
+    transient=True,
+) }}
+select * from {{ ref('my_seed') }}
+"""
+
+
+DYNAMIC_TABLE_NON_TRANSIENT = """
+{{ config(
+    materialized='dynamic_table',
+    snowflake_warehouse='DBT_TESTING',
+    target_lag='2 minutes',
+    refresh_mode='INCREMENTAL',
+    transient=False,
+) }}
+select * from {{ ref('my_seed') }}
+"""
+
+
+# For testing default behavior (no explicit transient config)
+DYNAMIC_TABLE_DEFAULT_TRANSIENT = """
+{{ config(
+    materialized='dynamic_table',
+    snowflake_warehouse='DBT_TESTING',
+    target_lag='2 minutes',
+    refresh_mode='INCREMENTAL',
+) }}
+select * from {{ ref('my_seed') }}
+"""