fix(databricks): use UC-compatible access mode when job paths resolve to Volumes

CHANGELOG.md

@@ -2,6 +2,22 @@
 
 All notable changes to spark-kindling are documented here.
 
+## Unreleased
+
+### Fixed
+
+- **Databricks job clusters now get a UC-compatible access mode whenever the
+  resolved job paths point at Unity Catalog Volumes** (gh#217). `_build_job_spec`
+  previously only set `data_security_mode=SINGLE_USER` when the opt-in
+  `cluster_logs_volume` was provided, so any other new-cluster submission
+  whose `artifacts_storage_path` or `python_file` resolved to `/Volumes/...`
+  (directly, via an explicit `python_file`/`bootstrap_script_root` override,
+  or via classic-mode env vars) still got a legacy "No Isolation Shared"
+  cluster, which cannot read Unity Catalog Volumes regardless of grants.
+  Detection now also checks the shape of the resolved `artifacts_storage_path`
+  and `python_file`; cluster log delivery (`cluster_log_conf`) remains a
+  separate, still-opt-in decision.
+
 ## [0.12.3] - 2026-07-29
 
 ### Added

packages/kindling_sdk/kindling_sdk/platform_databricks.py

@@ -212,6 +212,11 @@ def _join_storage_path(root: str, *parts: str) -> str:
             return root.rstrip("/")
         return "/".join([root.rstrip("/"), *cleaned_parts])
 
+    @staticmethod
+    def _is_uc_volume_path(path: str) -> bool:
+        candidate = str(path or "").strip()
+        return candidate == "/Volumes" or candidate.startswith("/Volumes/")
+
     def _resolve_python_file(
         self,
         main_file: str,
@@ -431,6 +436,12 @@ def _build_job_spec(self, job_name: str, job_config: Dict[str, Any]) -> Dict[str
         main_file = job_config.get("main_file", "kindling_bootstrap.py")
         system_test_mode = self._resolve_system_test_mode(job_config)
         artifacts_storage_path = self._resolve_artifacts_storage_path(job_config, system_test_mode)
+        python_file = self._resolve_python_file(
+            main_file=main_file,
+            job_config=job_config,
+            mode=system_test_mode,
+            artifacts_storage_path=artifacts_storage_path,
+        )
 
         bootstrap_params = {
             "app_name": app_name,
@@ -479,15 +490,24 @@ def flatten_dict(d, parent_key=""):
         spark_conf = job_config.get("spark_config", {}).copy()
 
         uc_volume_path = job_config.get("cluster_logs_volume")
+        needs_uc_mode = (
+            bool(uc_volume_path)
+            or self._is_uc_volume_path(artifacts_storage_path)
+            or self._is_uc_volume_path(python_file)
+        )
+
         if uc_volume_path:
             from databricks.sdk.service.compute import VolumesStorageInfo
 
             log_path = f"{uc_volume_path}/cluster-logs/kindling-jobs/{job_name}"
             cluster_log_conf = ClusterLogConf(volumes=VolumesStorageInfo(destination=log_path))
+        else:
+            cluster_log_conf = None
+
+        if needs_uc_mode:
             spark_conf.pop("spark.databricks.cluster.profile", None)
             data_security_mode = DataSecurityMode.SINGLE_USER
         else:
-            cluster_log_conf = None
             data_security_mode = None
 
         cluster_spec = ClusterSpec(
@@ -499,13 +519,6 @@ def flatten_dict(d, parent_key=""):
             spark_conf=spark_conf if spark_conf else None,
         )
 
-        python_file = self._resolve_python_file(
-            main_file=main_file,
-            job_config=job_config,
-            mode=system_test_mode,
-            artifacts_storage_path=artifacts_storage_path,
-        )
-
         existing_cluster_id = (
             job_config.get("existing_cluster_id")
             or job_config.get("cluster_id")

tests/unit/test_platform_databricks_sdk_job_config.py

@@ -116,6 +116,19 @@ def test_resolve_python_file_uses_abfss_for_uc():
     )
 
 
+def test_is_uc_volume_path_matches_volumes_root_and_prefix():
+    api = _make_api()
+
+    assert api._is_uc_volume_path("/Volumes")
+    assert api._is_uc_volume_path("/Volumes/cat/schema/vol")
+    assert not api._is_uc_volume_path("/Volume/cat")
+    assert not api._is_uc_volume_path(
+        "abfss://artifacts@mystorageaccount.dfs.core.windows.net/artifacts"
+    )
+    assert not api._is_uc_volume_path("")
+    assert not api._is_uc_volume_path(None)
+
+
 # --- Bug fix: DATABRICKS_TOKEN respected in _create_sdk_client ---
 
 
@@ -221,6 +234,97 @@ def test_create_job_sets_cluster_log_conf_when_uc_volume_provided():
     assert cluster.data_security_mode == DataSecurityMode.SINGLE_USER
 
 
+# --- gh#217: UC-compatible access mode is driven by resolved path shape, ---
+# --- not only the cluster_logs_volume opt-in ---
+
+
+def test_create_job_sets_single_user_mode_for_uc_volume_artifacts_path():
+    """artifacts_storage_path under /Volumes/... → SINGLE_USER even with no cluster_logs_volume."""
+    api = _make_api_for_create_job()
+
+    api.create_job(
+        "test-job",
+        {
+            "artifacts_storage_path": "/Volumes/cat/schema/vol",
+            "spark_config": {
+                "spark.databricks.cluster.profile": "singleNode",
+                "spark.other.setting": "keep-me",
+            },
+        },
+    )
+
+    from databricks.sdk.service.compute import DataSecurityMode
+
+    create_call = api._client.jobs.create.call_args
+    task = create_call.kwargs["tasks"][0]
+    cluster = task.new_cluster
+    assert cluster.data_security_mode == DataSecurityMode.SINGLE_USER
+    assert cluster.cluster_log_conf is None
+    assert cluster.spark_conf == {"spark.other.setting": "keep-me"}
+
+
+def test_create_job_sets_single_user_mode_for_explicit_python_file_volume_override():
+    """python_file explicitly overridden to /Volumes/... → SINGLE_USER even when
+    artifacts_storage_path itself is a non-Volumes abfss path."""
+    api = _make_api_for_create_job()
+
+    api.create_job(
+        "test-job",
+        {
+            "artifacts_storage_path": "abfss://c@sa.dfs.core.windows.net/artifacts",
+            "python_file": "/Volumes/cat/schema/vol/boot.py",
+        },
+    )
+
+    from databricks.sdk.service.compute import DataSecurityMode
+
+    create_call = api._client.jobs.create.call_args
+    task = create_call.kwargs["tasks"][0]
+    cluster = task.new_cluster
+    assert cluster.data_security_mode == DataSecurityMode.SINGLE_USER
+
+
+def test_create_job_sets_single_user_mode_for_classic_bootstrap_root_volume(monkeypatch):
+    """Classic-mode bootstrap_script_root resolving under /Volumes/... → SINGLE_USER,
+    even though artifacts_storage_path itself is not a Volumes path."""
+    api = _make_api_for_create_job()
+    monkeypatch.setenv("KINDLING_DATABRICKS_CLASSIC_BOOTSTRAP_ROOT", "/Volumes/cat/schema/vol")
+
+    api.create_job(
+        "test-job",
+        {
+            "system_test_mode": "classic",
+            "artifacts_storage_path": "dbfs:/mnt/artifacts",
+        },
+    )
+
+    from databricks.sdk.service.compute import DataSecurityMode
+
+    create_call = api._client.jobs.create.call_args
+    task = create_call.kwargs["tasks"][0]
+    cluster = task.new_cluster
+    assert cluster.data_security_mode == DataSecurityMode.SINGLE_USER
+
+
+def test_build_job_spec_existing_cluster_id_with_uc_volume_path_is_harmless():
+    """existing_cluster_id set alongside a /Volumes/... path: cluster_spec still picks up
+    SINGLE_USER (harmless, unused), and existing_cluster_id is what actually gets used."""
+    api = _make_api_for_create_job()
+
+    spec = api._build_job_spec(
+        "test-job",
+        {
+            "artifacts_storage_path": "/Volumes/cat/schema/vol",
+            "existing_cluster_id": "1234-567890-abcde123",
+        },
+    )
+
+    from databricks.sdk.service.compute import DataSecurityMode
+
+    assert spec["existing_cluster_id"] == "1234-567890-abcde123"
+    assert spec["cluster_spec"].data_security_mode == DataSecurityMode.SINGLE_USER
+
+
 # --- Bug fix: _submit_one_time_run calls jobs.submit, not jobs.runs.submit ---