[DO NOT MERGE (yet)] Avoid Iceberg CTAS for table materialization

dbt/adapters/duckdb/credentials.py

@@ -241,6 +241,7 @@ def __post_init__(self):
 
         # Build set of ducklake database names for efficient lookup
         self._ducklake_dbs = set()
+        self._iceberg_dbs = set()
 
         if self.is_ducklake or "ducklake:" in self.path.lower():
             self._ducklake_dbs.add(self.path_derived_database_name(self.path))
@@ -250,6 +251,7 @@ def __post_init__(self):
                 is_ducklake_flag = getattr(attachment, "is_ducklake", None)
                 path = getattr(attachment, "path", None)
                 alias = getattr(attachment, "alias", None)
+                attachment_type = getattr(attachment, "type", None)
 
                 # Detect ducklake by explicit type, or by path scheme. Be lenient on case.
                 if (isinstance(is_ducklake_flag, bool) and is_ducklake_flag) or (
@@ -259,6 +261,11 @@ def __post_init__(self):
                         self._ducklake_dbs.add(alias)
                     else:
                         self._ducklake_dbs.add(self.path_derived_database_name(path))
+                if isinstance(attachment_type, str) and attachment_type.lower() == "iceberg":
+                    if alias:
+                        self._iceberg_dbs.add(alias)
+                    else:
+                        self._iceberg_dbs.add(self.path_derived_database_name(path))
 
         # Add MotherDuck plugin if the path is a MotherDuck database
         # and plugin was not specified in profile.yml

dbt/adapters/duckdb/impl.py

@@ -151,6 +151,14 @@ def is_ducklake(self, relation: DuckDBRelation) -> bool:
 
         return relation.database in self.config.credentials._ducklake_dbs
 
+    @available
+    def is_iceberg(self, relation: DuckDBRelation) -> bool:
+        """Check if a relation's database is backed by an iceberg attachment."""
+        if not relation or not relation.database:
+            return False
+
+        return relation.database in self.config.credentials._iceberg_dbs
+
     @available
     def convert_datetimes_to_strs(self, table: "agate.Table") -> "agate.Table":
         import agate

dbt/include/duckdb/macros/adapters.sql

@@ -57,6 +57,18 @@
   )
 {% endmacro %}
 
+{% macro duckdb__column_definitions(columns) %}
+  {% for column in columns %}
+    {{ api.Relation.create(identifier=column.name) }} {{ column.data_type }}{{ "," if not loop.last }}
+  {% endfor %}
+{% endmacro %}
+
+{% macro duckdb__column_names(columns) %}
+  {% for column in columns %}
+    {{ api.Relation.create(identifier=column.name) }}{{ "," if not loop.last }}
+  {% endfor %}
+{% endmacro %}
+
 
 {% macro duckdb__get_partitioned_by(relation, temporary) -%}
   {%- if temporary -%}
@@ -111,6 +123,7 @@
 {% macro duckdb__create_table_as(temporary, relation, compiled_code, language='sql', partitioned_by=none) -%}
   {%- if language == 'sql' -%}
     {% set contract_config = config.get('contract') %}
+    {% set is_iceberg = adapter.is_iceberg(relation) %}
     {% if contract_config.enforced %}
       {{ get_assert_columns_equivalent(compiled_code) }}
     {% endif %}
@@ -120,7 +133,25 @@
 
     create {% if temporary: -%}temporary{%- endif %} table
       {{ relation.include(database=(not temporary), schema=(not temporary)) }}
-  {% if contract_config.enforced and not temporary %}
+  {% if is_iceberg and not temporary %}
+    {% if contract_config.enforced %}
+      {{ get_table_columns_and_constraints() }} ;
+      insert into {{ relation }} {{ get_column_names() }} (
+        {{ get_select_subquery(compiled_code) }}
+      );
+    {% else %}
+      {% set columns = adapter.get_column_schema_from_query(compiled_code) %}
+      (
+        {{ duckdb__column_definitions(columns) }}
+      );
+      insert into {{ relation }} (
+        {{ duckdb__column_names(columns) }}
+      )
+      select * from (
+        {{ compiled_code }}
+      ) as __dbt_iceberg_ctas;
+    {% endif %}
+  {% elif contract_config.enforced and not temporary %}
     {#-- DuckDB doesnt support constraints on temp tables --#}
     {{ get_table_columns_and_constraints() }} ;
     {% if partitioned_by %}
@@ -248,7 +279,20 @@ def materialize(df, con):
 {% macro duckdb__rename_relation(from_relation, to_relation) -%}
   {% set target_name = adapter.quote_as_configured(to_relation.identifier, 'identifier') %}
   {% call statement('rename_relation') -%}
-    alter {{ to_relation.type }} {{ from_relation }} rename to {{ target_name }}
+    {% if adapter.is_iceberg(to_relation) %}
+      {% set columns = adapter.get_column_schema_from_query('select * from ' ~ from_relation) %}
+      drop table if exists {{ to_relation }};
+      create table {{ to_relation }} (
+        {{ duckdb__column_definitions(columns) }}
+      );
+      insert into {{ to_relation }} (
+        {{ duckdb__column_names(columns) }}
+      )
+      select * from {{ from_relation }};
+      drop table if exists {{ from_relation }}
+    {% else %}
+      alter {{ to_relation.type }} {{ from_relation }} rename to {{ target_name }}
+    {% endif %}
   {%- endcall %}
 {% endmacro %}
 

dbt/include/duckdb/macros/materializations/table.sql

@@ -4,6 +4,34 @@
 
   {%- set existing_relation = load_cached_relation(this) -%}
   {%- set target_relation = this.incorporate(type='table') %}
+  {%- set is_iceberg = adapter.is_iceberg(target_relation) -%}
+  {% if is_iceberg %}
+    {% set grant_config = config.get('grants') %}
+
+    {{ drop_relation_if_exists(existing_relation) }}
+    {% call statement('drop_existing_relation') -%}
+      drop table if exists {{ target_relation }}
+    {%- endcall %}
+
+    {{ run_hooks(pre_hooks, inside_transaction=False) }}
+    {{ run_hooks(pre_hooks, inside_transaction=True) }}
+
+    {% call statement('main', language=language) -%}
+      {{- create_table_as(False, target_relation, compiled_code, language) }}
+    {%- endcall %}
+
+    {{ run_hooks(post_hooks, inside_transaction=True) }}
+
+    {% set should_revoke = should_revoke(existing_relation, full_refresh_mode=True) %}
+    {% do apply_grants(target_relation, grant_config, should_revoke=should_revoke) %}
+    {% do persist_docs(target_relation, model) %}
+
+    {{ adapter.commit() }}
+    {{ run_hooks(post_hooks, inside_transaction=False) }}
+
+    {{ return({'relations': [target_relation]}) }}
+  {% endif %}
+
   {%- set intermediate_relation =  make_intermediate_relation(target_relation) -%}
   -- the intermediate_relation should not already exist in the database; get_relation
   -- will return None in that case. Otherwise, we get a relation that we can drop

tests/unit/test_duckdb_adapter.py

@@ -340,3 +340,35 @@ def test_is_ducklake_with_type_ducklake_case_insensitive(self):
         result = adapter.is_ducklake(relation)
         self.assertTrue(result)
 
+    def test_is_iceberg_with_attachment_type(self):
+        """Test is_iceberg returns True when relation database matches iceberg attachment."""
+        profile_cfg = self.base_profile_cfg.copy()
+        profile_cfg["outputs"]["test"]["attach"] = [
+            {
+                "alias": "iceberg_db",
+                "path": "http://localhost:8181",
+                "type": "iceberg",
+            }
+        ]
+
+        adapter = self._get_adapter(profile_cfg)
+        relation = DuckDBRelation.create(database="iceberg_db", schema="test_schema", identifier="test_table")
+
+        result = adapter.is_iceberg(relation)
+        self.assertTrue(result)
+
+    def test_is_iceberg_regular_attachment(self):
+        """Test is_iceberg returns False for non-iceberg attachments."""
+        profile_cfg = self.base_profile_cfg.copy()
+        profile_cfg["outputs"]["test"]["attach"] = [
+            {
+                "alias": "regular_db",
+                "path": "/path/to/regular.db",
+            }
+        ]
+
+        adapter = self._get_adapter(profile_cfg)
+        relation = DuckDBRelation.create(database="regular_db", schema="test_schema", identifier="test_table")
+
+        result = adapter.is_iceberg(relation)
+        self.assertFalse(result)