diff --git a/CHANGELOG.md b/CHANGELOG.md index 248457155..707c60d07 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,10 @@ ## 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)) +- Use `create or replace table` instead of dropping the table first when the `table` materialization rebuilds a Unity Catalog managed Iceberg table ([#1674](https://github.com/databricks/dbt-databricks/pull/1674) 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)) diff --git a/dbt/include/databricks/macros/materializations/incremental/incremental.sql b/dbt/include/databricks/macros/materializations/incremental/incremental.sql index e79859822..3f8527bdd 100644 --- a/dbt/include/databricks/macros/materializations/incremental/incremental.sql +++ b/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') %} diff --git a/dbt/include/databricks/macros/materializations/incremental/replaceable_format.sql b/dbt/include/databricks/macros/materializations/incremental/replaceable_format.sql new file mode 100644 index 000000000..23342bd2d --- /dev/null +++ b/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 %} diff --git a/dbt/include/databricks/macros/materializations/table.sql b/dbt/include/databricks/macros/materializations/table.sql index eee612c36..47bdf6e87 100644 --- a/dbt/include/databricks/macros/materializations/table.sql +++ b/dbt/include/databricks/macros/materializations/table.sql @@ -9,6 +9,7 @@ {% set existing_relation = adapter.get_relation(database=database, schema=schema, identifier=identifier, needs_information=True) %} {% set target_relation = this.incorporate(type='table') %} {% set compiled_code = adapter.clean_sql(compiled_code) %} + {%- set catalog_relation = adapter.build_catalog_relation(config.model) -%} {% if adapter.get_behavior_flag_no_warn('use_materialization_v2') %} {% set intermediate_relation = make_intermediate_relation(target_relation) %} @@ -25,7 +26,7 @@ {% if safe_create and existing_relation.can_be_renamed %} {{ safe_relation_replace(existing_relation, staging_relation, intermediate_relation, compiled_code) }} {% else %} - {% if existing_relation and (existing_relation.is_shallow_clone or existing_relation.type != 'table' or not (existing_relation.can_be_replaced and adapter.resolve_file_format(config) in ('delta', 'iceberg'))) -%} + {% if existing_relation and (existing_relation.is_shallow_clone or existing_relation.type != 'table' or not (existing_relation.can_be_replaced and format_allows_create_or_replace(catalog_relation, existing_relation))) -%} {{ adapter.drop_relation(existing_relation) }} {%- endif %} {{ create_table_at(target_relation, intermediate_relation, compiled_code) }} @@ -46,7 +47,7 @@ -- setup: if the target relation already exists, drop it -- in case if the existing and future table is delta or iceberg, we want to do a -- create or replace table instead of dropping, so we don't have the table unavailable - {% if existing_relation and (existing_relation.is_shallow_clone or existing_relation.type != 'table' or not (existing_relation.can_be_replaced and adapter.resolve_file_format(config) in ('delta', 'iceberg'))) -%} + {% if existing_relation and (existing_relation.is_shallow_clone or existing_relation.type != 'table' or not (existing_relation.can_be_replaced and format_allows_create_or_replace(catalog_relation, existing_relation))) -%} {{ adapter.drop_relation(existing_relation) }} {%- endif %} diff --git a/tests/functional/adapter/iceberg/test_iceberg_support.py b/tests/functional/adapter/iceberg/test_iceberg_support.py index 5bfb77b9f..ceae7fd99 100644 --- a/tests/functional/adapter/iceberg/test_iceberg_support.py +++ b/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,96 @@ 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 + + +class TestManagedIcebergTableRebuild(ManagedIcebergMixin): + """Rebuilding a `table` model 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_table_rebuild.sql": fixtures.basic_iceberg_swap} + + def test_rebuild_keeps_the_table(self, project): + util.run_dbt() + created = get_version_zero_timestamp(project, "iceberg_table_rebuild") + assert created is not None, "expected history on the managed Iceberg table" + + util.run_dbt() + + assert get_version_zero_timestamp(project, "iceberg_table_rebuild") == created, ( + "history restarted, so the rebuild dropped and recreated the table" + ) + + +@pytest.mark.skip_profile("databricks_cluster") +class TestManagedIcebergTableOverExistingDelta(ManagedIcebergMixin): + """`table` counterpart of TestManagedIcebergOverExistingDelta: switching `use_managed_iceberg` + on over an existing Delta table must drop and recreate it on the next run, since + `create or replace` cannot change a table's provider (issue #1662).""" + + @pytest.fixture(scope="class") + def models(self): + return {"iceberg_table_over_delta.sql": fixtures.basic_iceberg_swap} + + def test_run_converts_the_delta_table(self, project): + project.run_sql( + "create or replace table {database}.{schema}.iceberg_table_over_delta using delta " + "as select 1 as id" + ) + assert get_provider(project, "iceberg_table_over_delta") == "delta" + + util.run_dbt() + + assert get_provider(project, "iceberg_table_over_delta") == "iceberg" + rows = project.run_sql( + "select id from {database}.{schema}.iceberg_table_over_delta", fetch="all" + ) + assert len(rows) == 1 diff --git a/tests/unit/macros/materializations/incremental/test_replaceable_format.py b/tests/unit/macros/materializations/incremental/test_replaceable_format.py new file mode 100644 index 000000000..0b6b70efa --- /dev/null +++ b/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"