diff --git a/dbt/adapters/duckdb/credentials.py b/dbt/adapters/duckdb/credentials.py index d06ba28d..f5876c50 100644 --- a/dbt/adapters/duckdb/credentials.py +++ b/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 diff --git a/dbt/adapters/duckdb/impl.py b/dbt/adapters/duckdb/impl.py index f979e8b3..a1dae52c 100644 --- a/dbt/adapters/duckdb/impl.py +++ b/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 diff --git a/dbt/include/duckdb/macros/adapters.sql b/dbt/include/duckdb/macros/adapters.sql index 17b6b92c..c4822071 100644 --- a/dbt/include/duckdb/macros/adapters.sql +++ b/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 %} diff --git a/dbt/include/duckdb/macros/materializations/table.sql b/dbt/include/duckdb/macros/materializations/table.sql index a85ef359..d92e8fe9 100644 --- a/dbt/include/duckdb/macros/materializations/table.sql +++ b/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 diff --git a/tests/unit/test_duckdb_adapter.py b/tests/unit/test_duckdb_adapter.py index 77da76ab..b026f4a3 100644 --- a/tests/unit/test_duckdb_adapter.py +++ b/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)