feat(snowflake): add transient tmp relation type for incremental models

dbt-snowflake/.changes/unreleased/Features-20260424-transient-tmp-relation.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add `tmp_relation_type=transient` support for incremental models, enabling Snowflake lineage tracking; add `snowflake__resolve_incremental_tmp_relation` dispatch macro for controlling tmp relation destination
+time: 2026-04-24T00:00:00.000000-00:00
+custom:
+    Author: b-per
+    Issue: "1893"

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

@@ -12,48 +12,64 @@
 
        Low-level specifics:
        If an invalid option is specified, then we will raise an
-       excpetion with corresponding message.
+       exception with a corresponding message.
 
        Languages other than SQL (like Python) will use a temporary table.
-       With the default strategy of merge, the user may choose between a temporary
-       table and view (defaulting to view).
-
-       The append strategy can use a view because it will run a single INSERT statement.
-
-       When unique_key is none, the delete+insert and microbatch strategies can use a view beacuse a
-       single INSERT statement is run with no DELETES as part of the statement.
-       Otherwise, play it safe by using a temporary table.
-
-       Catalog-linked databases (Iceberg tables) does not support using temporary relations.
+       With the default strategy of merge, the user may choose between a
+       temporary table and view (defaulting to view).
+
+       The append strategy can use a view because it will run a single INSERT
+       statement.
+
+       When unique_key is none, the delete+insert and microbatch strategies
+       can use a view because a single INSERT statement is run with no DELETES
+       as part of the statement. Otherwise, play it safe by using a table.
+
+       Catalog-linked databases (Iceberg tables) do not support temporary
+       relations or transient tables — only Iceberg tables are allowed. A
+       permanent table is used as the tmp relation for CLD models.
+
+       'transient' is also available as a user-facing tmp_relation_type for
+       non-Iceberg models. Unlike session-scoped temporary tables, transient
+       tables are visible to Snowflake's lineage tracking. Note that transient
+       tables share the regular schema namespace; use the
+       snowflake__resolve_incremental_tmp_relation dispatch macro to redirect
+       tmp relations to a dedicated schema to avoid name collisions when
+       multiple runs share the same target schema.
   #} */
 
-  {#-- Always use table for catalog-linked databases (Iceberg) --#}
-  {% if snowflake__is_catalog_linked_database(relation=config.model) %}
-    {{ return("table") }}
-  {% endif %}
-
   {% if language == "python" and tmp_relation_type is not none %}
     {% do exceptions.raise_compiler_error(
       "Python models currently only support 'table' for tmp_relation_type but "
        ~ tmp_relation_type ~ " was specified."
     ) %}
   {% endif %}
 
-  {% if strategy in ["delete+insert", "microbatch"] and tmp_relation_type is not none and tmp_relation_type != "table" and unique_key is not none %}
+  {#-- Python always uses a temporary table, regardless of other conditions --#}
+  {% if language != "sql" %}
+    {{ return("table") }}
+  {% endif %}
+
+  {#-- CLD schemas only support Iceberg tables; use table (not transient) --#}
+  {% if snowflake__is_catalog_linked_database(relation=config.model) %}
+    {{ return("table") }}
+  {% endif %}
+
+  {% if strategy in ["delete+insert", "microbatch"] and tmp_relation_type is not none and tmp_relation_type not in ("table", "transient") and unique_key is not none %}
     {% do exceptions.raise_compiler_error(
       "In order to maintain consistent results when `unique_key` is not none,
-      the `" ~ strategy ~ "` strategy only supports `table` for `tmp_relation_type` but "
+      the `" ~ strategy ~ "` strategy only supports `table` or `transient` for `tmp_relation_type` but "
       ~ tmp_relation_type ~ " was specified."
       )
   %}
   {% endif %}
 
-  {% if language != "sql" %}
-    {{ return("table") }}
-  {% elif tmp_relation_type == "table" %}
+  {% if tmp_relation_type == "table" %}
     {{ return("table") }}
   {% elif tmp_relation_type == "view" %}
     {{ return("view") }}
+  {% elif tmp_relation_type == "transient" %}
+    {{ return("transient") }}
   {% elif strategy in ("default", "merge", "append", "insert_overwrite") %}
     {{ return("view") }}
   {% elif strategy in ["delete+insert", "microbatch"] and unique_key is none %}
@@ -63,6 +79,27 @@
   {% endif %}
 {% endmacro %}
 
+
+{% macro resolve_incremental_tmp_relation(tmp_relation) %}
+  {{ return(adapter.dispatch('resolve_incremental_tmp_relation', 'dbt')(tmp_relation)) }}
+{% endmacro %}
+
+
+{% macro snowflake__resolve_incremental_tmp_relation(tmp_relation) %}
+  {#--
+    Override this macro in your project to control where the incremental
+    tmp relation is created. Useful for redirecting to a dedicated scratch
+    schema to avoid name collisions when multiple runs share the same
+    target schema.
+
+    Example:
+      {% macro snowflake__resolve_incremental_tmp_relation(tmp_relation) %}
+        {{ return(tmp_relation.incorporate(schema='scratch')) }}
+      {% endmacro %}
+  --#}
+  {{ return(tmp_relation) }}
+{% endmacro %}
+
 {% materialization incremental, adapter='snowflake', supported_languages=['sql', 'python'] -%}
 
   {% set original_query_tag = set_query_tag() %}
@@ -94,8 +131,11 @@
   {% if is_catalog_linked_db %}
     {% set tmp_relation = make_temp_relation(this).incorporate(type=tmp_relation_type, catalog=catalog_relation.catalog_name, is_table=true) %}
   {% else %}
-    {% set tmp_relation = make_temp_relation(this).incorporate(type=tmp_relation_type) %}
+    {#-- Transient tables are dropped with DROP TABLE, so the relation type must be 'table' --#}
+    {% set tmp_relation_object_type = 'table' if tmp_relation_type == 'transient' else tmp_relation_type %}
+    {% set tmp_relation = make_temp_relation(this).incorporate(type=tmp_relation_object_type) %}
   {% endif %}
+  {% set tmp_relation = resolve_incremental_tmp_relation(tmp_relation) %}
 
   {% set grant_config = config.get('grants') %}
 
@@ -131,7 +171,7 @@
     %}
 
   {% else %}
-    {#-- Create the temp relation, either as a view or as a temp table --#}
+    {#-- Create the temp relation as a view, temp table, or transient table --#}
     {% if is_catalog_linked_db %}
         {%- call statement('create_tmp_relation', language=language) -%}
           {{ create_table_as(False, tmp_relation, compiled_code, language) }}
@@ -140,6 +180,10 @@
         {%- call statement('create_tmp_relation') -%}
           {{ snowflake__create_view_as_with_temp_flag(tmp_relation, compiled_code, True) }}
         {%- endcall -%}
+    {% elif tmp_relation_type == 'transient' %}
+        {%- call statement('create_tmp_relation', language=language) -%}
+          {{ snowflake__create_table_transient_sql(tmp_relation, compiled_code) }}
+        {%- endcall -%}
     {% else %}
         {%- call statement('create_tmp_relation', language=language) -%}
           {{ create_table_as(True, tmp_relation, compiled_code, language) }}

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

@@ -35,6 +35,36 @@
 {% endmacro %}
 
 
+{% macro snowflake__create_table_transient_sql(relation, compiled_code) -%}
+{#-
+    Implements CREATE TRANSIENT TABLE ... AS SELECT for use as an incremental
+    tmp relation. Unlike session-scoped temporary tables, transient tables
+    persist in the catalog (enabling Snowflake lineage tracking) but have no
+    fail-safe period, avoiding the storage costs of permanent tables.
+    https://docs.snowflake.com/en/sql-reference/sql/create-table
+-#}
+
+{%- set contract_config = config.get('contract') -%}
+{%- if contract_config.enforced -%}
+    {{- get_assert_columns_equivalent(compiled_code) -}}
+    {%- set compiled_code = get_select_subquery(compiled_code) -%}
+{%- endif -%}
+
+{%- set sql_header = config.get('sql_header', none) -%}
+{{ sql_header if sql_header is not none }}
+
+create or replace transient table {{ relation }}
+    {%- if contract_config.enforced %}
+    {{ get_table_columns_and_constraints() }}
+    {%- endif %}
+as (
+    {{ compiled_code }}
+    )
+;
+
+{%- endmacro %}
+
+
 {% macro snowflake__create_table_temporary_sql(relation, compiled_code) -%}
 {#-
     Implements CREATE TEMPORARY TABLE and CREATE TEMPORARY TABLE ... AS SELECT:

dbt-snowflake/tests/functional/adapter/incremental/test_incremental_transient.py

@@ -0,0 +1,78 @@
+import pytest
+from dbt.tests.util import run_dbt, run_dbt_and_capture
+
+
+_MODEL_TRANSIENT_TMP = """
+{{ config(
+    materialized='incremental',
+    unique_key='id',
+    tmp_relation_type='transient',
+) }}
+
+select 1 as id, 'alice' as name
+{% if is_incremental() %}
+union all
+select 2 as id, 'bob' as name
+{% endif %}
+"""
+
+_MODEL_TRANSIENT_TMP_DELETE_INSERT = """
+{{ config(
+    materialized='incremental',
+    incremental_strategy='delete+insert',
+    unique_key='id',
+    tmp_relation_type='transient',
+) }}
+
+select 1 as id, 'alice' as name
+{% if is_incremental() %}
+union all
+select 2 as id, 'bob' as name
+{% endif %}
+"""
+
+
+class TestIncrementalTransientTmpRelation:
+    """tmp_relation_type='transient' creates a transient (not session-scoped) staging
+    table, enabling Snowflake lineage tracking while avoiding permanent-table
+    fail-safe storage costs."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"transient_incremental.sql": _MODEL_TRANSIENT_TMP}
+
+    def test_incremental_transient(self, project):
+        run_dbt(["run"])
+        result = project.run_sql(
+            "select count(*) as cnt from {database}.{schema}.transient_incremental",
+            fetch="one",
+        )
+        assert result[0] == 1
+
+        _, logs = run_dbt_and_capture(["--debug", "run"])
+        assert "create or replace transient table" in logs.lower()
+
+        result = project.run_sql(
+            "select count(*) as cnt from {database}.{schema}.transient_incremental",
+            fetch="one",
+        )
+        assert result[0] == 2
+
+        run_dbt(["test"])
+
+
+class TestIncrementalTransientTmpRelationDeleteInsert:
+    """transient tmp_relation_type is allowed for delete+insert strategy since
+    transient tables are stable across multiple statements, unlike views."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"transient_delete_insert.sql": _MODEL_TRANSIENT_TMP_DELETE_INSERT}
+
+    def test_incremental_transient_delete_insert_runs(self, project):
+        run_dbt(["run"])
+
+        _, logs = run_dbt_and_capture(["--debug", "run"])
+        assert "create or replace transient table" in logs.lower()
+
+        run_dbt(["test"])