fix: Do not set UniForm properties on managed iceberg (#1256)

<!-- Please review our pull request review process in CONTRIBUTING.md
before your proceed. -->

Resolves #

<!---
Include the number of the issue addressed by this PR above if
applicable.

  Example:
    resolves #1234

Please review our pull request review process in CONTRIBUTING.md before
your proceed.
-->

### Description

There is a bug in `TblPropertiesProcessor` that is assuming that all
tables with iceberg format need UniForm properties to be set. When
managed iceberg behavior flag is enabled, this causes the
materialization to fail

The fix is to only apply these properties when working with a UniForm
table. Unfortunately, we cannot use the `file_format` to distinguish
because we do not actually require the user to set `file_format` to
`parquet` for managed iceberg. We infer it in the adapter based on the
`table_format` and behavior flag. To work around this fact, I am setting
`model.config.extra["_databricks_use_managed_iceberg"]` to pass the
information to the config processor

### Checklist

- [ ] I have run this code in development and it appears to resolve the
stated issue
- [ ] This PR includes tests, or tests are not required/relevant for
this PR
- [ ] I have updated the `CHANGELOG.md` and added information about my
change to the "dbt-databricks next" section.

Author: ericj-db

CHANGELOG.md

@@ -1,5 +1,9 @@
 ## dbt-databricks 1.11.1 (TBD)
 
+### Fixes
+
+- Fix bug that was applying UniForm tblproperties on managed Iceberg tables causing materializations to fail
+
 ### Under the hood
 
 - Add validation for query tag value length and auto-escape special characters

dbt/adapters/databricks/global_state.py

@@ -50,3 +50,15 @@ def get_connector_log_level(cls) -> str:
                 "DBT_DATABRICKS_CONNECTOR_LOG_LEVEL", "WARN"
             ).upper()
         return cls.__connector_log_level
+
+    __use_managed_iceberg: ClassVar[Optional[bool]] = None
+
+    @classmethod
+    def set_use_managed_iceberg(cls, value: bool) -> None:
+        """Set the use_managed_iceberg flag for the session."""
+        cls.__use_managed_iceberg = value
+
+    @classmethod
+    def get_use_managed_iceberg(cls) -> Optional[bool]:
+        """Get the use_managed_iceberg flag for the session."""
+        return cls.__use_managed_iceberg

dbt/adapters/databricks/impl.py

@@ -236,6 +236,9 @@ def __init__(self, config: Any, mp_context: SpawnContext) -> None:
         self.add_catalog_integration(constants.DEFAULT_UNITY_CATALOG)
         self.add_catalog_integration(constants.DEFAULT_HIVE_METASTORE_CATALOG)
 
+        # Store the use_managed_iceberg flag in GlobalState for the session
+        GlobalState.set_use_managed_iceberg(bool(self.behavior.use_managed_iceberg))
+
     @property
     def _behavior_flags(self) -> list[BehaviorFlag]:
         return [
@@ -920,6 +923,7 @@ def get_relation_config(self, relation: DatabricksRelation) -> DatabricksRelatio
     @available.parse(lambda *a, **k: {})
     def get_config_from_model(self, model: RelationConfig) -> DatabricksRelationConfigBase:
         assert model.config, "Config was missing from relation"
+
         if model.config.materialized == "materialized_view":
             return MaterializedViewAPI.get_from_relation_config(model)
         elif model.config.materialized == "streaming_table":

dbt/adapters/databricks/relation_configs/tblproperties.py

@@ -4,6 +4,8 @@
 from dbt.adapters.relation_configs.config_base import RelationResults
 from dbt_common.exceptions import DbtRuntimeError
 
+from dbt.adapters.databricks import constants
+from dbt.adapters.databricks.global_state import GlobalState
 from dbt.adapters.databricks.relation_configs import base
 from dbt.adapters.databricks.relation_configs.base import (
     DatabricksComponentConfig,
@@ -81,18 +83,22 @@ def from_relation_results(cls, results: RelationResults) -> TblPropertiesConfig:
     @classmethod
     def from_relation_config(cls, relation_config: RelationConfig) -> TblPropertiesConfig:
         tblproperties = base.get_config_value(relation_config, "tblproperties") or {}
-        is_iceberg = base.get_config_value(relation_config, "table_format") == "iceberg"
 
         if not isinstance(tblproperties, dict):
             raise DbtRuntimeError("tblproperties must be a dictionary")
 
-        # If the table format is Iceberg, we need to set the iceberg-specific tblproperties
-        if is_iceberg:
+        table_format = base.get_config_value(relation_config, "table_format")
+        use_managed_iceberg = GlobalState.get_use_managed_iceberg()
+
+        is_uniform = table_format == constants.ICEBERG_TABLE_FORMAT and use_managed_iceberg is False
+
+        if is_uniform:
             tblproperties.update(
                 {
                     "delta.enableIcebergCompatV2": "true",
-                    "delta.universalFormat.enabledFormats": "iceberg",
+                    "delta.universalFormat.enabledFormats": constants.ICEBERG_TABLE_FORMAT,
                 }
             )
+
         tblproperties = {str(k): str(v) for k, v in tblproperties.items()}
         return TblPropertiesConfig(tblproperties=tblproperties)

tests/functional/adapter/iceberg/fixtures.py

@@ -69,3 +69,29 @@
 }}
 select 1 as id
 """
+
+incremental_iceberg_base = """
+{{
+  config(
+    materialized = "incremental",
+    table_format = "iceberg",
+    incremental_strategy = "merge",
+    unique_key = "id",
+  )
+}}
+select 1 as id, 'initial' as status
+"""
+
+incremental_iceberg_update = """
+{{
+  config(
+    materialized = "incremental",
+    table_format = "iceberg",
+    incremental_strategy = "merge",
+    unique_key = "id",
+  )
+}}
+select 1 as id, 'updated' as status
+union all
+select 2 as id, 'new' as status
+"""

tests/functional/adapter/iceberg/test_iceberg_support.py

@@ -65,3 +65,33 @@ class TestManagedIcebergTables(TestIcebergTables, ManagedIcebergMixin):
 
 class TestManagedIcebergSwap(TestIcebergSwap, ManagedIcebergMixin):
     pass
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIcebergIncrementalMerge(ManagedIcebergMixin):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"iceberg_incremental.sql": fixtures.incremental_iceberg_base}
+
+    def test_iceberg_incremental_merge(self, project):
+        results = util.run_dbt()
+        assert len(results) == 1
+
+        result = project.run_sql("select * from iceberg_incremental order by id", fetch="all")
+        assert len(result) == 1
+        assert result[0][0] == 1  # id
+        assert result[0][1] == "initial"  # status
+
+        util.write_file(fixtures.incremental_iceberg_update, "models", "iceberg_incremental.sql")
+
+        # Second run should merge successfully without UniForm property errors
+        results = util.run_dbt()
+        assert len(results) == 1
+
+        # Verify merged data: id=1 should be updated, id=2 should be new
+        result = project.run_sql("select * from iceberg_incremental order by id", fetch="all")
+        assert len(result) == 2
+        assert result[0][0] == 1
+        assert result[0][1] == "updated"  # Updated via merge
+        assert result[1][0] == 2
+        assert result[1][1] == "new"  # New row

tests/unit/relation_configs/test_tblproperties.py

@@ -3,6 +3,8 @@
 import pytest
 from dbt.exceptions import DbtRuntimeError
 
+from dbt.adapters.databricks import constants
+from dbt.adapters.databricks.global_state import GlobalState
 from dbt.adapters.databricks.relation_configs.tblproperties import (
     TblPropertiesConfig,
     TblPropertiesProcessor,
@@ -56,3 +58,42 @@ def test_from_model_node__with_incorrect_tblproperties(self):
             match="tblproperties must be a dictionary",
         ):
             _ = TblPropertiesProcessor.from_relation_config(model)
+
+    def test_from_model_node__with_uniform_iceberg_adds_properties(self):
+        GlobalState.set_use_managed_iceberg(False)
+        model = Mock()
+        model.config.extra = {
+            "table_format": constants.ICEBERG_TABLE_FORMAT,
+            "tblproperties": {"custom_prop": "value"},
+        }
+        spec = TblPropertiesProcessor.from_relation_config(model)
+        # Should have both custom property AND UniForm properties
+        assert spec == TblPropertiesConfig(
+            tblproperties={
+                "custom_prop": "value",
+                "delta.enableIcebergCompatV2": "true",
+                "delta.universalFormat.enabledFormats": constants.ICEBERG_TABLE_FORMAT,
+            }
+        )
+
+    def test_from_model_node__with_managed_iceberg_no_uniform_properties(self):
+        GlobalState.set_use_managed_iceberg(True)
+        model = Mock()
+        model.config.extra = {
+            "table_format": constants.ICEBERG_TABLE_FORMAT,
+            "tblproperties": {"custom_prop": "value"},
+        }
+        spec = TblPropertiesProcessor.from_relation_config(model)
+        # Should only have the custom property, NOT the UniForm properties
+        assert spec == TblPropertiesConfig(tblproperties={"custom_prop": "value"})
+
+    def test_from_model_node__with_iceberg_no_flag_no_properties(self):
+        GlobalState.set_use_managed_iceberg(None)
+        model = Mock()
+        model.config.extra = {
+            "table_format": constants.ICEBERG_TABLE_FORMAT,
+            "tblproperties": {},
+        }
+        spec = TblPropertiesProcessor.from_relation_config(model)
+        # Should not have UniForm properties without explicit use_managed_iceberg=False
+        assert spec == TblPropertiesConfig(tblproperties={})