fix: prevent _cache_dbr_capabilities from caching None-version entries (#1398)

When _query_dbr_version() returns None after a successful connection (e.g.
transient failure on `SET spark.databricks.clusterUsageTags.sparkVersion`),
_cache_dbr_capabilities would write DBRCapabilities(dbr_version=None) to the
class-level cache, and the idempotency guard at the top of the method would
block every subsequent re-query — silently disabling every version-gated
feature for the life of the process.

Apply the same None-guard already present in _try_cache_dbr_capabilities, and
switch the downstream read in open() to .get() with a default DBRCapabilities()
so the now-possibly-missing entry no longer KeyErrors.

Author: sd-db

CHANGELOG.md

@@ -1,3 +1,9 @@
+## dbt-databricks next (TBD)
+
+### Fixes
+
+- Fix DBR capability cache being permanently poisoned by a transient version-query failure ([#1398](https://github.com/databricks/dbt-databricks/issues/1398))
+
 ## dbt-databricks 1.11.7 (Apr 17, 2026)
 
 ### Features

dbt/adapters/databricks/connections.py

@@ -202,11 +202,11 @@ def _cache_dbr_capabilities(cls, creds: DatabricksCredentials, http_path: str) -
         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)
-
-            cls._dbr_capabilities_cache[http_path] = DBRCapabilities(
-                dbr_version=dbr_version,
-                is_sql_warehouse=not is_cluster,
-            )
+            if dbr_version is not None:
+                cls._dbr_capabilities_cache[http_path] = DBRCapabilities(
+                    dbr_version=dbr_version,
+                    is_sql_warehouse=not is_cluster,
+                )
 
     @classmethod
     def _try_cache_dbr_capabilities(cls, creds: DatabricksCredentials, http_path: str) -> None:
@@ -506,9 +506,9 @@ def connect() -> DatabricksHandle:
                 if conn:
                     databricks_connection.session_id = conn.session_id
                     cls._cache_dbr_capabilities(creds, databricks_connection.http_path)
-                    databricks_connection.capabilities = cls._dbr_capabilities_cache[
-                        databricks_connection.http_path
-                    ]
+                    databricks_connection.capabilities = cls._dbr_capabilities_cache.get(
+                        databricks_connection.http_path, DBRCapabilities()
+                    )
                     return conn
                 else:
                     raise DbtDatabaseError("Failed to create connection")

tests/unit/test_connection_manager.py

@@ -148,3 +148,70 @@ def test_skips_write_when_already_cached(self, mock_query):
 
         mock_query.assert_not_called()
         assert DatabricksConnectionManager._dbr_capabilities_cache[self.HTTP_PATH] is existing
+
+
+class TestCacheDbr:
+    """Unit tests for _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):
+        """Regression for #1398: a None version query result must not poison the cache."""
+        creds = Mock(spec=DatabricksCredentials)
+        creds.cluster_id = None
+
+        DatabricksConnectionManager._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._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._cache_dbr_capabilities(creds, self.HTTP_PATH)
+
+        mock_query.assert_not_called()
+        assert DatabricksConnectionManager._dbr_capabilities_cache[self.HTTP_PATH] is existing
+
+    def test_retry_succeeds_after_transient_failure(self):
+        """Regression for #1398: after a transient None result, the next call must re-query."""
+        creds = Mock(spec=DatabricksCredentials)
+        creds.cluster_id = None
+
+        with patch.object(DatabricksConnectionManager, "_query_dbr_version", return_value=None):
+            DatabricksConnectionManager._cache_dbr_capabilities(creds, self.HTTP_PATH)
+
+        with patch.object(DatabricksConnectionManager, "_query_dbr_version", return_value=(16, 2)):
+            DatabricksConnectionManager._cache_dbr_capabilities(creds, self.HTTP_PATH)
+
+        caps = DatabricksConnectionManager._dbr_capabilities_cache.get(self.HTTP_PATH)
+        assert caps is not None
+        assert caps.dbr_version == (16, 2)
+        assert caps.has_capability(DBRCapability.ICEBERG)