Fix insert overwrite compatibility check for model-level compute override

Author: ericj-db

dbt/adapters/databricks/connections.py

@@ -205,7 +205,13 @@ def api_client(self) -> DatabricksApiClient:
         return self._api_client
 
     def is_cluster(self) -> bool:
-        return self.get_thread_connection().credentials.cluster_id is not None
+        conn = self.get_thread_connection()
+        return (
+            conn.credentials.cluster_id is not None
+            # Credentials field is not updated when overriding the compute at model level.
+            # This secondary check is a workaround for that case
+            or "/warehouses/" not in cast(DatabricksDBTConnection, conn).http_path
+        )
 
     def cancel_open(self) -> list[str]:
         cancelled = super().cancel_open()

tests/conftest.py

@@ -48,6 +48,13 @@ def dbt_profile_data(unique_schema, dbt_profile_target, profiles_config_update):
     }
     target = dbt_profile_target
     target["schema"] = unique_schema
+
+    # For testing model-level compute override
+    target["compute"] = {
+        "alternate_uc_cluster": {
+            "http_path": get_databricks_cluster_target("databricks_uc_cluster")["http_path"]
+        }
+    }
     profile["test"]["outputs"]["default"] = target
 
     alternate_warehouse = target.copy()

tests/functional/adapter/incremental/test_incremental_strategies.py

@@ -106,7 +106,31 @@ def project_config_update(self):
             "models": {
                 "+incremental_strategy": "insert_overwrite",
                 "+partition_by": "id",
-            }
+            },
+        }
+
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {
+            "upsert_expected.csv": fixtures.upsert_expected,
+        }
+
+    def test_incremental(self, project):
+        self.seed_and_run_twice()
+        util.check_relations_equal(project.adapter, ["overwrite_model", "upsert_expected"])
+
+
+# Only runs under SQL warehouse profile, but overrides compute at model level
+@pytest.mark.skip_profile("databricks_uc_cluster", "databricks_cluster")
+class TestInsertOverwriteWithModelComputeOverride(IncrementalBase):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "models": {
+                "+incremental_strategy": "insert_overwrite",
+                "+partition_by": "id",
+                "+databricks_compute": "alternate_uc_cluster",
+            },
         }
 
     @pytest.fixture(scope="class")
@@ -115,6 +139,12 @@ def seeds(self):
             "upsert_expected.csv": fixtures.upsert_expected,
         }
 
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "overwrite_model.sql": fixtures.base_model,
+        }
+
     def test_incremental(self, project):
         self.seed_and_run_twice()
         util.check_relations_equal(project.adapter, ["overwrite_model", "upsert_expected"])