fix: capability detection for named compute (databricks_compute) (#1355)

Resolves #1354 

### Description

Fixes a timing bug where `table_format='iceberg'` (and other
capability-gated features) fail with "iceberg requires DBR 14.3+" when
the model targets a named compute via `databricks_compute` model config,
even though the compute supports the required DBR version.

### Fix

One-line change: call `_cache_dbr_capabilities()` eagerly in
`_create_fresh_connection()` before reading the cache. This ensures the
DBR version is queried and cached at connection creation time rather
than waiting for the lazy open(). The call is idempotent (guarded by if
`http_path` not in cache), so for default compute with a warm cache it's
a no-op.

### Key changes
- `dbt/adapters/databricks/connections.py`: Added
`self._cache_dbr_capabilities(creds, conn.http_path)` before
`conn.capabilities =
self._get_capabilities_for_http_path(conn.http_path)` in
`_create_fresh_connection()`
- `tests/unit/test_connection_manager.py`: Added 3 unit tests validating
eager caching for named compute, fallback behavior when cache is empty,
and idempotency for default compute with a warm cache

### Checklist

- [x] I have run this code in development and it appears to resolve the
stated issue
- [x] 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.

---------

Signed-off-by: trouze <tyler@tylerrouze.com>
Co-authored-by: Shubham Dhal <shubham.dhal@databricks.com>

Author: trouze

dbt/adapters/databricks/connections.py

@@ -208,6 +208,23 @@ def _cache_dbr_capabilities(cls, creds: DatabricksCredentials, http_path: str) -
                 is_sql_warehouse=not is_cluster,
             )
 
+    @classmethod
+    def _try_cache_dbr_capabilities(cls, creds: DatabricksCredentials, http_path: str) -> None:
+        """Like _cache_dbr_capabilities, but only writes to the cache when the version query
+        actually succeeds. This prevents a failed eager lookup (e.g. credentials_manager not yet
+        set, cluster still spinning up) from storing a None-version entry that the idempotency
+        guard in open() would later treat as authoritative, causing all capability checks to
+        return False.
+        """
+        if http_path not in cls._dbr_capabilities_cache:
+            is_cluster = is_cluster_http_path(http_path, creds.cluster_id)
+            dbr_version = cls._query_dbr_version(creds, http_path)
+            if dbr_version is not None:
+                cls._dbr_capabilities_cache[http_path] = DBRCapabilities(
+                    dbr_version=dbr_version,
+                    is_sql_warehouse=not is_cluster,
+                )
+
     def cancel_open(self) -> list[str]:
         cancelled = super().cancel_open()
         logger.info("Cancelling open python jobs")
@@ -288,6 +305,7 @@ def _create_fresh_connection(
         conn.http_path = QueryConfigUtils.get_http_path(query_header_context, creds)
         conn.thread_identifier = cast(tuple[int, int], self.get_thread_identifier())
         conn._query_header_context = query_header_context
+        self._try_cache_dbr_capabilities(creds, conn.http_path)
         conn.capabilities = self._get_capabilities_for_http_path(conn.http_path)
         conn.handle = LazyHandle(self.open)
 

tests/unit/test_connection_manager.py

@@ -1,8 +1,14 @@
 from multiprocessing import get_context
 from unittest.mock import Mock, patch
 
-from dbt.adapters.databricks.connections import DatabricksConnectionManager, DatabricksDBTConnection
+import pytest
+
+from dbt.adapters.databricks.connections import (
+    DatabricksConnectionManager,
+    DatabricksDBTConnection,
+)
 from dbt.adapters.databricks.credentials import DatabricksCredentials
+from dbt.adapters.databricks.dbr_capabilities import DBRCapabilities, DBRCapability
 from dbt.adapters.databricks.utils import is_cluster_http_path
 
 
@@ -88,3 +94,57 @@ def test_open_calls_is_cluster_http_path_for_warehouse(
         args, kwargs = mock_from_connection_args.call_args
         # Second argument (is_cluster) should be True for warehouse path with cluster_id
         assert args[1] is True
+
+
+class TestTryCacheDbr:
+    """Unit tests for _try_cache_dbr_capabilities."""
+
+    HTTP_PATH = "sql/protocolv1/o/1234567890123456/cluster-abc"
+
+    @pytest.fixture(autouse=True)
+    def clear_cache(self):
+        DatabricksConnectionManager._dbr_capabilities_cache = {}
+        yield
+        DatabricksConnectionManager._dbr_capabilities_cache = {}
+
+    @patch.object(DatabricksConnectionManager, "_query_dbr_version", return_value=None)
+    def test_does_not_write_to_cache_when_version_is_none(self, mock_query):
+        """When the version query returns None, the cache must not be written.
+
+        This prevents a poisoned None entry from blocking the authoritative write in open().
+        """
+        creds = Mock(spec=DatabricksCredentials)
+        creds.cluster_id = None
+
+        DatabricksConnectionManager._try_cache_dbr_capabilities(creds, self.HTTP_PATH)
+
+        assert self.HTTP_PATH not in DatabricksConnectionManager._dbr_capabilities_cache
+        mock_query.assert_called_once_with(creds, self.HTTP_PATH)
+
+    @patch.object(DatabricksConnectionManager, "_query_dbr_version", return_value=(15, 4))
+    def test_writes_to_cache_when_version_is_known(self, mock_query):
+        """When the version query succeeds, capabilities are cached correctly."""
+        creds = Mock(spec=DatabricksCredentials)
+        creds.cluster_id = None
+
+        DatabricksConnectionManager._try_cache_dbr_capabilities(creds, self.HTTP_PATH)
+
+        mock_query.assert_called_once_with(creds, self.HTTP_PATH)
+        caps = DatabricksConnectionManager._dbr_capabilities_cache.get(self.HTTP_PATH)
+        assert caps is not None
+        assert caps.dbr_version == (15, 4)
+        assert not caps.is_sql_warehouse
+        assert caps.has_capability(DBRCapability.ICEBERG)
+
+    @patch.object(DatabricksConnectionManager, "_query_dbr_version", return_value=(15, 4))
+    def test_skips_write_when_already_cached(self, mock_query):
+        """If the path is already in cache, the version query is never made."""
+        creds = Mock(spec=DatabricksCredentials)
+        creds.cluster_id = None
+        existing = DBRCapabilities(dbr_version=(14, 3), is_sql_warehouse=False)
+        DatabricksConnectionManager._dbr_capabilities_cache[self.HTTP_PATH] = existing
+
+        DatabricksConnectionManager._try_cache_dbr_capabilities(creds, self.HTTP_PATH)
+
+        mock_query.assert_not_called()
+        assert DatabricksConnectionManager._dbr_capabilities_cache[self.HTTP_PATH] is existing