fix: replace instead of drop on managed Iceberg full refresh

CHANGELOG.md

@@ -1,5 +1,9 @@
 ## dbt-databricks 1.12.6 (TBD)
 
+### Fixes
+
+- Use `create or replace table` instead of dropping the table first when a full refresh rebuilds an incremental model on a Unity Catalog managed Iceberg table ([#1669](https://github.com/databricks/dbt-databricks/pull/1669) resolves [#1662](https://github.com/databricks/dbt-databricks/issues/1662))
+
 ### Under the Hood
 
 - Emit only changed `databricks_tags` keys in `ALTER … SET TAGS` ([#1667](https://github.com/databricks/dbt-databricks/pull/1667))

dbt/include/databricks/macros/materializations/incremental/incremental.sql

@@ -11,9 +11,7 @@
   {% set partition_by = config.get('partition_by') %}
   {% set language = model['language'] %}
   {% set on_schema_change = incremental_validate_on_schema_change(config.get('on_schema_change'), default='ignore') %}
-  {% set is_delta = (catalog_relation.file_format == 'delta' and existing_relation.is_delta) %}
-  {% set is_iceberg = (catalog_relation.file_format == 'iceberg' and existing_relation.is_iceberg) %}
-  {% set is_replaceable_format = is_delta or is_iceberg %}
+  {% set is_replaceable_format = format_allows_create_or_replace(catalog_relation, existing_relation) %}
   {% set compiled_code = adapter.clean_sql(model['compiled_code']) %}
 
   {% if adapter.get_behavior_flag_no_warn('use_materialization_v2') %}

dbt/include/databricks/macros/materializations/incremental/replaceable_format.sql

@@ -0,0 +1,29 @@
+{#-- True when `create or replace table` can stand in for drop-then-create on a full refresh.
+
+     Managed Iceberg needs its own arm rather than reusing `file_format`: an Iceberg model keeps
+     `file_format` at delta (`iceberg_table_properties` raises for anything else) and `iceberg` is
+     not an accepted `file_format` at all, so the `file_format == 'iceberg'` test this replaced
+     could never be true. The target is Iceberg exactly when `table_format` is iceberg and the
+     behavior flag is on -- the same condition `file_format_clause` uses to emit `using iceberg`.
+     The flag is read without warning because this runs for every incremental model, including
+     projects that never opt in (issue #1266).
+
+     The two arms are mutually exclusive because `create or replace` cannot change a table's
+     provider: Databricks rejects it with `MANAGED_ICEBERG_OPERATION_NOT_SUPPORTED` and leaves the
+     table as it was. A managed-Iceberg target over a legacy Delta table -- a project that has just
+     switched the flag on -- must therefore drop and recreate, even though `file_format` still reads
+     delta for it. --#}
+{% macro format_allows_create_or_replace(catalog_relation, existing_relation) %}
+  {%- set target_is_managed_iceberg = (
+        catalog_relation.table_format == 'iceberg'
+        and adapter.get_behavior_flag_no_warn('use_managed_iceberg')
+      ) -%}
+  {%- if target_is_managed_iceberg -%}
+    {%- set replaceable = existing_relation.is_iceberg is true -%}
+  {%- else -%}
+    {%- set replaceable = (
+          catalog_relation.file_format == 'delta' and existing_relation.is_delta is true
+        ) -%}
+  {%- endif -%}
+  {{ return(replaceable) }}
+{% endmacro %}

tests/functional/adapter/iceberg/test_iceberg_support.py

@@ -26,6 +26,19 @@ def get_tblproperty(project, identifier, key):
     return values[0] if values else None
 
 
+def get_version_zero_timestamp(project, identifier):
+    """Timestamp of the table's first history entry. A `create or replace` keeps it; dropping
+    and recreating the table starts a new history, so the value changes."""
+    rows = project.run_sql(
+        f"describe history {{database}}.{{schema}}.{identifier}",
+        fetch="all",
+    )
+    for row in rows:
+        if int(row[0]) == 0:
+            return str(row[1])
+    return None
+
+
 @pytest.mark.skip_profile("databricks_cluster")
 class TestIcebergTables:
     @pytest.fixture(scope="class")
@@ -172,3 +185,50 @@ def test_iceberg_incremental_merge(self, project):
         assert result[0][1] == "updated"  # Updated via merge
         assert result[1][0] == 2
         assert result[1][1] == "new"  # New row
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestManagedIcebergFullRefresh(ManagedIcebergMixin):
+    """A full refresh must replace a managed Iceberg table in place rather than dropping it
+    first, so the table stays queryable for the whole rebuild (issue #1662)."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"iceberg_full_refresh.sql": fixtures.incremental_iceberg_base}
+
+    def test_full_refresh_keeps_the_table(self, project):
+        util.run_dbt()
+        created = get_version_zero_timestamp(project, "iceberg_full_refresh")
+        assert created is not None, "expected history on the managed Iceberg table"
+
+        util.run_dbt(["run", "--full-refresh"])
+
+        assert get_version_zero_timestamp(project, "iceberg_full_refresh") == created, (
+            "history restarted, so the full refresh dropped and recreated the table"
+        )
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestManagedIcebergOverExistingDelta(ManagedIcebergMixin):
+    """Switching `use_managed_iceberg` on over tables a project already has as Delta must drop and
+    recreate them. `create or replace` cannot change a table's provider, so replacing here fails
+    with MANAGED_ICEBERG_OPERATION_NOT_SUPPORTED and leaves the table Delta (issue #1662)."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"iceberg_over_delta.sql": fixtures.incremental_iceberg_base}
+
+    def test_full_refresh_converts_the_delta_table(self, project):
+        project.run_sql(
+            "create or replace table {database}.{schema}.iceberg_over_delta using delta "
+            "as select 1 as id, 'initial' as status"
+        )
+        assert get_provider(project, "iceberg_over_delta") == "delta"
+
+        util.run_dbt(["run", "--full-refresh"])
+
+        assert get_provider(project, "iceberg_over_delta") == "iceberg"
+        rows = project.run_sql(
+            "select id, status from {database}.{schema}.iceberg_over_delta", fetch="all"
+        )
+        assert len(rows) == 1

tests/unit/macros/materializations/incremental/test_replaceable_format.py

@@ -0,0 +1,104 @@
+from unittest.mock import Mock
+
+import pytest
+
+from tests.unit.macros.base import MacroTestBase
+
+
+class TestFormatAllowsCreateOrReplace(MacroTestBase):
+    """The predicate that decides whether a full refresh can use `create or replace table`
+    instead of dropping the existing relation first (issue #1662)."""
+
+    @pytest.fixture(scope="class")
+    def template_name(self) -> str:
+        return "replaceable_format.sql"
+
+    @pytest.fixture(scope="class")
+    def macro_folders_to_load(self) -> list:
+        return ["macros/materializations/incremental"]
+
+    def _catalog_relation(self, table_format="default", file_format="delta"):
+        catalog_relation = Mock()
+        catalog_relation.table_format = table_format
+        catalog_relation.file_format = file_format
+        return catalog_relation
+
+    def _existing_relation(self, is_delta=False, is_iceberg=False):
+        existing_relation = Mock()
+        existing_relation.is_delta = is_delta
+        existing_relation.is_iceberg = is_iceberg
+        return existing_relation
+
+    def run_predicate(self, template_bundle, catalog_relation, existing_relation, managed_iceberg):
+        template_bundle.context["adapter"].get_behavior_flag_no_warn = Mock(
+            side_effect=lambda name: managed_iceberg if name == "use_managed_iceberg" else False
+        )
+        return self.run_macro_raw(
+            template_bundle.template,
+            "format_allows_create_or_replace",
+            catalog_relation,
+            existing_relation,
+        ).strip()
+
+    def test_delta_target_on_delta_relation(self, template_bundle):
+        result = self.run_predicate(
+            template_bundle,
+            self._catalog_relation(),
+            self._existing_relation(is_delta=True),
+            managed_iceberg=False,
+        )
+        assert result == "True"
+
+    def test_delta_target_on_iceberg_relation(self, template_bundle):
+        """Provider changed under the model, so the table has to be dropped."""
+        result = self.run_predicate(
+            template_bundle,
+            self._catalog_relation(),
+            self._existing_relation(is_iceberg=True),
+            managed_iceberg=False,
+        )
+        assert result == "False"
+
+    def test_managed_iceberg_target_on_iceberg_relation(self, template_bundle):
+        """The case from #1662: an Iceberg model keeps `file_format` at delta, so keying the
+        Iceberg arm off `file_format` never matched and the full refresh dropped the table."""
+        result = self.run_predicate(
+            template_bundle,
+            self._catalog_relation(table_format="iceberg"),
+            self._existing_relation(is_iceberg=True),
+            managed_iceberg=True,
+        )
+        assert result == "True"
+
+    def test_uniform_target_on_delta_relation(self, template_bundle):
+        """`table_format: iceberg` without the behavior flag writes a Delta table with UniForm
+        properties, so the relation stays Delta and remains replaceable."""
+        result = self.run_predicate(
+            template_bundle,
+            self._catalog_relation(table_format="iceberg"),
+            self._existing_relation(is_delta=True),
+            managed_iceberg=False,
+        )
+        assert result == "True"
+
+    def test_managed_iceberg_target_on_delta_relation(self, template_bundle):
+        """A project that has just switched the flag on still has a Delta table. `create or
+        replace` cannot change a table's provider -- Databricks rejects it with
+        MANAGED_ICEBERG_OPERATION_NOT_SUPPORTED -- so this has to drop and recreate even though
+        `file_format` still reads delta for a managed Iceberg model."""
+        result = self.run_predicate(
+            template_bundle,
+            self._catalog_relation(table_format="iceberg"),
+            self._existing_relation(is_delta=True),
+            managed_iceberg=True,
+        )
+        assert result == "False"
+
+    def test_non_delta_file_format(self, template_bundle):
+        result = self.run_predicate(
+            template_bundle,
+            self._catalog_relation(file_format="parquet"),
+            self._existing_relation(is_delta=True),
+            managed_iceberg=False,
+        )
+        assert result == "False"