fix: Set correct cluster type on databricks connection (#1158)

<!-- 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

We're currently setting the wrong connection type in
`DatabricksHandle.from_connection_args(..)` when establishing
connections with a compute override at the model level

### 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

dbt/adapters/databricks/connections.py

@@ -36,7 +36,7 @@
 from dbt.adapters.databricks.handle import CursorWrapper, DatabricksHandle, SqlUtils
 from dbt.adapters.databricks.logging import logger
 from dbt.adapters.databricks.python_models.run_tracking import PythonRunTracker
-from dbt.adapters.databricks.utils import redact_credentials
+from dbt.adapters.databricks.utils import is_cluster_http_path, redact_credentials
 from dbt.adapters.events.types import (
     ConnectionClosedInCleanup,
     ConnectionReused,
@@ -130,12 +130,8 @@ def api_client(self) -> DatabricksApiClient:
 
     def is_cluster(self) -> bool:
         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
-        )
+        databricks_conn = cast(DatabricksDBTConnection, conn)
+        return is_cluster_http_path(databricks_conn.http_path, conn.credentials.cluster_id)
 
     def cancel_open(self) -> list[str]:
         cancelled = super().cancel_open()
@@ -402,7 +398,8 @@ def connect() -> DatabricksHandle:
             try:
                 # TODO: what is the error when a user specifies a catalog they don't have access to
                 conn = DatabricksHandle.from_connection_args(
-                    conn_args, creds.cluster_id is not None
+                    conn_args,
+                    is_cluster_http_path(databricks_connection.http_path, creds.cluster_id),
                 )
                 if conn:
                     databricks_connection.session_id = conn.session_id

dbt/adapters/databricks/utils.py

@@ -1,6 +1,6 @@
 import re
 from collections.abc import Callable
-from typing import TYPE_CHECKING, Any, TypeVar
+from typing import TYPE_CHECKING, Any, Optional, TypeVar
 
 from dbt_common.exceptions import DbtRuntimeError
 from jinja2 import Undefined
@@ -88,3 +88,12 @@ def handle_exceptions_as_warning(op: Callable[[], None], log_gen: ExceptionToStr
         op()
     except Exception as e:
         logger.warning(log_gen(e))
+
+
+def is_cluster_http_path(http_path: str, cluster_id: Optional[str]) -> bool:
+    return (
+        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 http_path
+    )

tests/unit/test_connection_manager.py

@@ -0,0 +1,88 @@
+from multiprocessing import get_context
+from unittest.mock import Mock, patch
+
+from dbt.adapters.databricks.connections import DatabricksConnectionManager, DatabricksDBTConnection
+from dbt.adapters.databricks.credentials import DatabricksCredentials
+from dbt.adapters.databricks.utils import is_cluster_http_path
+
+
+class TestDatabricksConnectionManager:
+    def test_is_cluster_with_warehouse_path_no_cluster_id(self):
+        """Test is_cluster() returns False for warehouse path with no cluster_id"""
+        # Create a minimal connection manager with mock config
+        mock_config = Mock()
+        connection_manager = DatabricksConnectionManager(mock_config, get_context("spawn"))
+
+        # Mock the connection
+        mock_connection = Mock(spec=DatabricksDBTConnection)
+        mock_connection.credentials = Mock(spec=DatabricksCredentials)
+        mock_connection.credentials.cluster_id = None
+        mock_connection.http_path = "sql/1.0/warehouses/abc123def456"
+
+        with patch.object(
+            connection_manager, "get_thread_connection", return_value=mock_connection
+        ):
+            assert connection_manager.is_cluster() is False
+
+    def test_is_cluster_with_cluster_id_overrides_path(self):
+        """Test is_cluster() returns True when cluster_id is provided, regardless of path"""
+        # Create a minimal connection manager with mock config
+        mock_config = Mock()
+        connection_manager = DatabricksConnectionManager(mock_config, get_context("spawn"))
+
+        # Mock the connection with cluster_id set (overriding warehouse path)
+        mock_connection = Mock(spec=DatabricksDBTConnection)
+        mock_connection.credentials = Mock(spec=DatabricksCredentials)
+        mock_connection.credentials.cluster_id = "cluster-123"
+        mock_connection.http_path = "sql/1.0/warehouses/abc123def456"
+
+        with patch.object(
+            connection_manager, "get_thread_connection", return_value=mock_connection
+        ):
+            assert connection_manager.is_cluster() is True
+
+    def test_is_cluster_http_path_function_warehouse_path(self):
+        assert is_cluster_http_path("sql/1.0/warehouses/abc123def456", None) is False
+
+    def test_is_cluster_http_path_function_cluster_path(self):
+        assert is_cluster_http_path("sql/protocolv1/o/1234567890123456/", None) is True
+
+    def test_is_cluster_http_path_function_cluster_id_overrides(self):
+        assert is_cluster_http_path("sql/1.0/warehouses/abc123def456", "cluster-123") is True
+
+    @patch("dbt.adapters.databricks.connections.DatabricksHandle.from_connection_args")
+    @patch("dbt.adapters.databricks.connections.SqlUtils.prepare_connection_arguments")
+    def test_open_calls_is_cluster_http_path_for_warehouse(
+        self, mock_prepare_args, mock_from_connection_args
+    ):
+        """
+        Test that open() method calls is_cluster_http_path with correct arguments for warehouse
+        """
+        # Create a minimal connection manager with mock config
+        mock_config = Mock()
+        connection_manager = DatabricksConnectionManager(mock_config, get_context("spawn"))
+
+        # Mock the connection with proper timeout values
+        mock_connection = Mock(spec=DatabricksDBTConnection)
+        mock_connection.credentials = Mock(spec=DatabricksCredentials)
+        mock_connection.credentials.cluster_id = None
+        mock_connection.credentials.connect_retries = 1
+        mock_connection.credentials.connect_timeout = 10
+        mock_connection.http_path = "sql/protocolv1/o/abc123def456"
+        mock_connection.credentials.authenticate.return_value = Mock()
+
+        # Mock the handle creation
+        mock_handle = Mock()
+        mock_handle.session_id = "test_session"
+        mock_from_connection_args.return_value = mock_handle
+
+        mock_prepare_args.return_value = {}
+
+        # Call open method
+        connection_manager.open(mock_connection)
+
+        # Verify that from_connection_args was called with is_cluster=False (warehouse path)
+        mock_from_connection_args.assert_called_once()
+        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

tests/unit/test_utils.py

@@ -1,4 +1,9 @@
-from dbt.adapters.databricks.utils import quote, redact_credentials, remove_ansi
+from dbt.adapters.databricks.utils import (
+    is_cluster_http_path,
+    quote,
+    redact_credentials,
+    remove_ansi,
+)
 
 
 class TestDatabricksUtils:
@@ -67,3 +72,12 @@ def test_remove_ansi(self):
 
     def test_quote(self):
         assert quote("table") == "`table`"
+
+    def test_is_cluster_http_path_with_cluster_id(self):
+        assert is_cluster_http_path("/sql/1.0/warehouses/abc", "cluster-123") is True
+
+    def test_is_cluster_http_path_without_cluster_id_and_warehouses(self):
+        assert is_cluster_http_path("/sql/1.0/endpoints/abc", None) is True
+
+    def test_is_cluster_http_path_without_cluster_id_and_with_warehouses(self):
+        assert is_cluster_http_path("/sql/1.0/warehouses/abc", None) is False