update incremental sql to support iceberg REST w CLD (#1287)

Co-authored-by: Zeeshan Qureshi <[email protected]>
Co-authored-by: tnk-ysk <[email protected]>
Co-authored-by: Mila Page <[email protected]>
Co-authored-by: sandeepmullangi2 <[email protected]>
Co-authored-by: Daisuke Taniwaki <[email protected]>
Co-authored-by: Doug Beatty <[email protected]>

Author: 7 people

dbt-snowflake/.changes/unreleased/Features-20250902-125917.yaml

@@ -0,0 +1,7 @@
+kind: Features
+body: 'Add support for incremental materializations using iceberg rest catalog / catalog
+  linked db '
+time: 2025-09-02T12:59:17.884349-07:00
+custom:
+  Author: colin-rogers-dbt
+  Issue: "1123"

dbt-snowflake/src/dbt/adapters/snowflake/catalogs/_iceberg_rest.py

@@ -39,6 +39,7 @@ class IcebergRestCatalogIntegration(CatalogIntegration):
     def __init__(self, config: CatalogIntegrationConfig) -> None:
         # we overwrite this because the base provides too much config
         self.name: str = config.name
+        self.catalog_name: Optional[str] = config.catalog_name
         self.external_volume: Optional[str] = config.external_volume
         if adapter_properties := config.adapter_properties:
             self.catalog_linked_database = adapter_properties.get("catalog_linked_database")
@@ -75,7 +76,7 @@ def build_relation(self, model: RelationConfig) -> IcebergRestCatalogRelation:
             )
 
         return IcebergRestCatalogRelation(
-            catalog_name=self.catalog_name,
+            catalog_name=self.name,
             external_volume=None,
             catalog_linked_database=self.catalog_linked_database,
             auto_refresh=parse_model.auto_refresh(model) or self.auto_refresh,

dbt-snowflake/src/dbt/include/snowflake/macros/adapters.sql

@@ -229,6 +229,35 @@
 
 
 
+{% macro snowflake__is_catalog_linked_database(relation=none, catalog_relation=none) -%}
+    {#-- Helper macro to detect if we're in a catalog-linked database context --#}
+    {%- if catalog_relation is not none -%}
+        {#-- Direct catalog_relation object provided --#}
+        {%- if catalog_relation|attr('catalog_linked_database') -%}
+            {{ return(true) }}
+        {%- else -%}
+            {{ return(false) }}
+        {%- endif -%}
+    {%- elif relation and relation.config -%}
+        {%- set catalog_relation = adapter.build_catalog_relation(relation) -%}
+        {%- if catalog_relation is not none and catalog_relation|attr('catalog_linked_database') -%}
+            {{ return(true) }}
+        {%- else -%}
+            {{ return(false) }}
+        {%- endif -%}
+    {%- elif relation and relation.catalog -%}
+        {#-- Relation with catalog attribute --#}
+        {%- set catalog_integration = adapter.get_catalog_integration(relation.catalog) -%}
+        {%- if catalog_integration is not none and catalog_integration|attr('catalog_linked_database') -%}
+            {{ return(true) }}
+        {%- else -%}
+            {{ return(false) }}
+        {%- endif -%}
+    {%- else -%}
+        {{ return(false) }}
+    {%- endif -%}
+{%- endmacro %}
+
 {% macro snowflake_dml_explicit_transaction(dml) %}
   {#
     Use this macro to wrap all INSERT, MERGE, UPDATE, DELETE, and TRUNCATE
@@ -251,6 +280,10 @@
     truncate table {{ relation.render() }}
   {% endset %}
   {% call statement('truncate_relation') -%}
-    {{ snowflake_dml_explicit_transaction(truncate_dml) }}
+    {% if snowflake__is_catalog_linked_database(relation=config.model) %}
+        {{ truncate_dml }}
+    {% else %}
+      {{ snowflake_dml_explicit_transaction(truncate_dml) }}
+    {% endif %}
   {%- endcall %}
 {% endmacro %}

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

@@ -23,8 +23,15 @@
        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.
   #} */
 
+  {#-- 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 "
@@ -67,12 +74,15 @@
   {%- set identifier = this.name -%}
   {%- set catalog_relation = adapter.build_catalog_relation(config.model) -%}
 
+  {%- set is_catalog_linked_db = snowflake__is_catalog_linked_database(relation=none, catalog_relation=catalog_relation) -%}
+
   {%- set target_relation = api.Relation.create(
 	identifier=identifier,
 	schema=schema,
 	database=database,
 	type='table',
-	table_format=catalog_relation.table_format
+	table_format=catalog_relation.table_format,
+  catalog=config.model.catalog,
   ) -%}
 
   {% set existing_relation = load_relation(this) %}
@@ -81,7 +91,12 @@
   {%- set unique_key = config.get('unique_key') -%}
   {% set incremental_strategy = config.get('incremental_strategy') or 'default' %}
   {% set tmp_relation_type = dbt_snowflake_get_tmp_relation_type(incremental_strategy, unique_key, language) %}
-  {% set tmp_relation = make_temp_relation(this).incorporate(type=tmp_relation_type) %}
+
+  {% 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) %}
+  {% endif %}
 
   {% set grant_config = config.get('grants') %}
 
@@ -118,7 +133,11 @@
 
   {% else %}
     {#-- Create the temp relation, either as a view or as a temp table --#}
-    {% if tmp_relation_type == 'view' %}
+    {% if is_catalog_linked_db %}
+        {%- call statement('create_tmp_relation', language=language) -%}
+          {{ create_table_as(False, tmp_relation, compiled_code, language) }}
+        {%- endcall -%}
+    {% elif tmp_relation_type == 'view' %}
         {%- call statement('create_tmp_relation') -%}
           {{ snowflake__create_view_as_with_temp_flag(tmp_relation, compiled_code, True) }}
         {%- endcall -%}
@@ -140,13 +159,14 @@
     {#-- Get the incremental_strategy, the macro to use for the strategy, and build the sql --#}
     {% set incremental_predicates = config.get('predicates', none) or config.get('incremental_predicates', none) %}
     {% set strategy_sql_macro_func = adapter.get_incremental_strategy_macro(context, incremental_strategy) %}
-    {% set strategy_arg_dict = ({'target_relation': target_relation, 'temp_relation': tmp_relation, 'unique_key': unique_key, 'dest_columns': dest_columns, 'incremental_predicates': incremental_predicates }) %}
+    {% set strategy_arg_dict = ({'target_relation': target_relation, 'temp_relation': tmp_relation, 'unique_key': unique_key, 'dest_columns': dest_columns, 'incremental_predicates': incremental_predicates, 'catalog_relation': catalog_relation }) %}
 
     {%- call statement('main') -%}
       {{ strategy_sql_macro_func(strategy_arg_dict) }}
     {%- endcall -%}
   {% endif %}
 
+
   {% do drop_relation_if_exists(tmp_relation) %}
 
   {{ run_hooks(post_hooks) }}

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

@@ -32,6 +32,11 @@
 
     {%- endset -%}
 
-    {% do return(snowflake_dml_explicit_transaction(dml)) %}
+    {#-- Skip transaction wrapping for catalog-linked databases --#}
+    {% if snowflake__is_catalog_linked_database(relation=config.model) %}
+        {% do return(dml) %}
+    {% else %}
+        {% do return(snowflake_dml_explicit_transaction(dml)) %}
+    {% endif %}
 
 {% endmacro %}

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

@@ -1,3 +1,4 @@
+
 {% macro snowflake__get_merge_sql(target, source_sql, unique_key, dest_columns, incremental_predicates) -%}
 
     {#
@@ -27,26 +28,46 @@
     {%- endif -%}
     {%- endset -%}
 
-    {% do return(snowflake_dml_explicit_transaction(dml)) %}
+    {#-- Skip transaction wrapping for catalog-linked databases --#}
+    {% if snowflake__is_catalog_linked_database(relation=config.model) %}
+        {% do return(dml) %}
+    {% else %}
+        {% do return(snowflake_dml_explicit_transaction(dml)) %}
+    {% endif %}
 
 {% endmacro %}
 
 
 {% macro snowflake__get_delete_insert_merge_sql(target, source, unique_key, dest_columns, incremental_predicates) %}
     {% set dml = default__get_delete_insert_merge_sql(target, source, unique_key, dest_columns, incremental_predicates) %}
-    {% do return(snowflake_dml_explicit_transaction(dml)) %}
+    {#-- Skip transaction wrapping for catalog-linked databases --#}
+    {% if snowflake__is_catalog_linked_database(relation=config.model) %}
+        {% do return(dml) %}
+    {% else %}
+        {% do return(snowflake_dml_explicit_transaction(dml)) %}
+    {% endif %}
 {% endmacro %}
 
 
 {% macro snowflake__snapshot_merge_sql(target, source, insert_cols) %}
     {% set dml = default__snapshot_merge_sql(target, source, insert_cols) %}
-    {% do return(snowflake_dml_explicit_transaction(dml)) %}
+    {#-- Skip transaction wrapping for catalog-linked databases --#}
+    {% if snowflake__is_catalog_linked_database(relation=target) %}
+        {% do return(dml) %}
+    {% else %}
+        {% do return(snowflake_dml_explicit_transaction(dml)) %}
+    {% endif %}
 {% endmacro %}
 
 
 {% macro snowflake__get_incremental_append_sql(get_incremental_append_sql) %}
     {% set dml = default__get_incremental_append_sql(get_incremental_append_sql) %}
-    {% do return(snowflake_dml_explicit_transaction(dml)) %}
+    {#-- Skip transaction wrapping for catalog-linked databases --#}
+    {% if snowflake__is_catalog_linked_database(config.model) %}
+        {% do return(dml) %}
+    {% else %}
+        {% do return(snowflake_dml_explicit_transaction(dml)) %}
+    {% endif %}
 {% endmacro %}
 
 

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

@@ -1,3 +1,9 @@
-{% macro snowflake__get_drop_table_sql(relation) %}
-    drop table if exists {{ relation }} cascade
+{% macro snowflake__drop_table(relation) %}
+    {#-- CASCADE is not supported in catalog-linked databases --#}
+
+    {% if snowflake__is_catalog_linked_database(relation=relation) %}
+        drop table if exists {{ relation }}
+    {% else %}
+        drop table if exists {{ relation }} cascade
+    {% endif %}
 {% endmacro %}

dbt-snowflake/tests/functional/adapter/catalog_integrations/test_iceberg_rest_catalog_integrations.py

@@ -17,6 +17,38 @@
                             select 1 as id
                             """
 
+MODEL__INCREMENTAL_ICEBERG_REST = """
+{{
+  config(
+    materialized='incremental',
+    catalog_name='basic_iceberg_rest_catalog',
+    incremental_strategy='merge',
+    unique_key="id",
+  )
+}}
+select * from {{ ref('basic_iceberg_table') }}
+
+{% if is_incremental() %}
+where id > 2
+{% endif %}
+"""
+
+MODEL__INCREMENTAL_ICEBERG_REST_INSERT_OVERWRITE = """
+{{
+  config(
+    materialized='incremental',
+    catalog_name='basic_iceberg_rest_catalog',
+    incremental_strategy='insert_overwrite',
+    unique_key="id",
+  )
+}}
+select * from {{ ref('basic_iceberg_table') }}
+
+{% if is_incremental() %}
+where id > 2
+{% endif %}
+"""
+
 
 class TestSnowflakeIcebergRestCatalogIntegration(BaseCatalogIntegrationValidation):
 
@@ -62,10 +94,12 @@ def models(self):
             "models": {
                 "basic_iceberg_table.sql": MODEL__BASIC_ICEBERG_TABLE,
                 "iceberg_table_with_catalog_config.sql": MODEL__ICEBERG_TABLE_WITH_CATALOG_CONFIG,
+                "incremental_iceberg_rest.sql": MODEL__INCREMENTAL_ICEBERG_REST,
+                "incremental_iceberg_rest_insert_overwrite.sql": MODEL__INCREMENTAL_ICEBERG_REST_INSERT_OVERWRITE,
             }
         }
 
     def test_basic_iceberg_rest_catalog_integration(self, project):
         result = run_dbt(["run"])
-        assert len(result) == 2
+        assert len(result) == 4
         run_dbt(["run"])

dbt-snowflake/tests/unit/test_iceberg_rest_catalog_integration.py

@@ -128,14 +128,6 @@ def test_integration_with_environment_variable_for_macro(self):
         # Simulate the macro check
         assert hasattr(relation, "catalog_linked_database")
 
-        # The macro would use catalog_name when catalog_linked_database is available
-        if relation is not None and hasattr(relation, "catalog_linked_database"):
-            result = relation.catalog_name
-        else:
-            result = "target_database"
-
-        assert result == "POLARIS"
-
     def test_catalog_relation_all_attributes_present(self):
         """Test that all expected attributes are present on the catalog relation."""
         relation = IcebergRestCatalogRelation(