Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion sqlmesh/core/engine_adapter/databricks.py
Original file line number Diff line number Diff line change
Expand Up @@ -154,7 +154,7 @@ def _set_spark_engine_adapter_if_needed(self) -> None:
host=self._extra_config["databricks_connect_server_hostname"],
token=self._extra_config.get("databricks_connect_access_token"),
)
if "databricks_connect_use_serverless" in self._extra_config:
if self._extra_config.get("databricks_connect_use_serverless"):
connect_kwargs["serverless"] = True
else:
connect_kwargs["cluster_id"] = self._extra_config["databricks_connect_cluster_id"]
Expand Down
105 changes: 105 additions & 0 deletions tests/core/engine_adapter/test_databricks.py
Original file line number Diff line number Diff line change
Expand Up @@ -732,3 +732,108 @@ def test_columns(mocker: MockFixture, make_mocked_engine_adapter: t.Callable):
adapter.cursor.execute.assert_called_once_with(
"""SELECT columns.column_name, columns.full_data_type FROM system.information_schema.columns WHERE table_name = 'test_table' AND table_schema = 'test_db' AND table_catalog = 'test_catalog' ORDER BY ordinal_position ASC"""
)


def _make_databricks_connect_adapter(
mocker: MockFixture,
make_mocked_engine_adapter: t.Callable,
extra_config: t.Dict[str, t.Any],
) -> t.Tuple[DatabricksEngineAdapter, t.Any]:
"""Helper that creates a DatabricksEngineAdapter with Databricks Connect mocked out."""
import sys
import types

mock_session = mocker.MagicMock()
mock_builder = mocker.MagicMock()
mock_builder.remote.return_value = mock_builder
mock_builder.userAgent.return_value = mock_builder
mock_builder.getOrCreate.return_value = mock_session
mock_databricks_session_cls = mocker.MagicMock()
mock_databricks_session_cls.builder = mock_builder

# databricks.connect is a local import inside the method, so inject via sys.modules
mock_connect_module = types.ModuleType("databricks.connect")
mock_connect_module.DatabricksSession = mock_databricks_session_cls # type: ignore
mock_databricks_module = types.ModuleType("databricks")
mocker.patch.dict(
sys.modules,
{"databricks": mock_databricks_module, "databricks.connect": mock_connect_module},
)

mocker.patch(
"sqlmesh.core.engine_adapter.databricks.DatabricksEngineAdapter.can_access_spark_session",
return_value=False,
)
mocker.patch(
"sqlmesh.core.engine_adapter.databricks.DatabricksEngineAdapter.can_access_databricks_connect",
return_value=True,
)

adapter = make_mocked_engine_adapter(
DatabricksEngineAdapter,
default_catalog="test_catalog",
**extra_config,
)
return adapter, mock_builder


def test_databricks_connect_routes_to_cluster_id(
mocker: MockFixture, make_mocked_engine_adapter: t.Callable
) -> None:
"""cluster_id is used when databricks_connect_use_serverless is absent."""
extra_config = {
"databricks_connect_server_hostname": "myhost.azuredatabricks.net",
"databricks_connect_access_token": "mytoken",
"databricks_connect_cluster_id": "0123-456789-mycluster",
}
_, mock_builder = _make_databricks_connect_adapter(
mocker, make_mocked_engine_adapter, extra_config
)

mock_builder.remote.assert_called_once_with(
host="myhost.azuredatabricks.net",
token="mytoken",
cluster_id="0123-456789-mycluster",
)


def test_databricks_connect_routes_to_serverless(
mocker: MockFixture, make_mocked_engine_adapter: t.Callable
) -> None:
"""serverless=True is used when databricks_connect_use_serverless is truthy."""
extra_config = {
"databricks_connect_server_hostname": "myhost.azuredatabricks.net",
"databricks_connect_access_token": "mytoken",
"databricks_connect_cluster_id": "0123-456789-mycluster",
"databricks_connect_use_serverless": True,
}
_, mock_builder = _make_databricks_connect_adapter(
mocker, make_mocked_engine_adapter, extra_config
)

mock_builder.remote.assert_called_once_with(
host="myhost.azuredatabricks.net",
token="mytoken",
serverless=True,
)


def test_databricks_connect_cluster_id_not_overridden_by_falsy_serverless(
mocker: MockFixture, make_mocked_engine_adapter: t.Callable
) -> None:
"""cluster_id is used when databricks_connect_use_serverless is present but False."""
extra_config = {
"databricks_connect_server_hostname": "myhost.azuredatabricks.net",
"databricks_connect_access_token": "mytoken",
"databricks_connect_cluster_id": "0123-456789-mycluster",
"databricks_connect_use_serverless": False,
}
_, mock_builder = _make_databricks_connect_adapter(
mocker, make_mocked_engine_adapter, extra_config
)

mock_builder.remote.assert_called_once_with(
host="myhost.azuredatabricks.net",
token="mytoken",
cluster_id="0123-456789-mycluster",
)
Loading