fix(kernel): classify invalid transfer paths and track release gate

Signed-off-by: Aakash Saravanan <aakash.saravanan@databricks.com>

Author: aakash-saravanan-db

KERNEL_REV

@@ -1 +1 @@
-2f317d97212f00d73b89d947f5261a2145ea7a10
+95f294c8c2c825440191af9f58a9b6071aa68408

pyproject.toml

@@ -70,7 +70,7 @@ requests-kerberos = {version = "^0.15.0", optional = true}
 # Floor is 1.4.0 (``^1.4.0`` == ``>=1.4.0,<2.0.0``), allowing compatible
 # 1.x releases while keeping the next major upgrade deliberate. The kernel's
 # pyarrow constraint remains >=23.0.1,<24.
-# TODO: raise the release floor after databricks/databricks-sql-kernel#439 is published.
+# TODO: gate merge/release on https://github.com/databricks/databricks-sql-kernel/issues/443.
 databricks-sql-kernel = {version = "^1.4.0", optional = true, python = ">=3.10"}
 
 

src/databricks/sql/backend/kernel/client.py

@@ -74,10 +74,7 @@
 # per-request skip-and-warn.
 _KERNEL_MANAGED_HEADERS = frozenset({"authorization", "x-databricks-org-id"})
 
-
-# Leading verbs of SQL volume/staging statements. Detected by the
-# leading token (case-insensitive) so the kernel backend can fail loud
-# on staging ops it can't service — see ``execute_command``.
+# Transfer detection controls local-path forwarding and the older-wheel guard.
 _STAGING_VERBS = ("PUT", "GET", "REMOVE")
 
 
@@ -597,7 +594,15 @@ def execute_command(
                     allowed_paths = [allowed_paths]
                 set_paths = getattr(stmt, "set_staging_allowed_local_paths", None)
                 if callable(set_paths):
-                    set_paths([os.fsdecode(path) for path in allowed_paths or []])
+                    try:
+                        normalized_paths = [
+                            os.fsdecode(path) for path in allowed_paths or []
+                        ]
+                    except (TypeError, ValueError) as exc:
+                        raise ProgrammingError(
+                            "staging_allowed_local_path must be a path or a list of paths"
+                        ) from exc
+                    set_paths(normalized_paths)
                 elif _is_staging_statement(operation):
                     raise NotSupportedError(
                         "Volume transfers require a databricks-sql-kernel wheel "

tests/unit/test_kernel_client.py

@@ -848,7 +848,10 @@ def test_execute_command_forwards_allowed_local_paths(allowed, expected, async_o
 
 
 @pytest.mark.parametrize("allowed", [123, [object()]])
-def test_non_transfer_query_ignores_malformed_local_paths(allowed):
+@pytest.mark.parametrize(
+    "operation", ["SELECT 1", "PUT 'input' INTO '/Volumes/c/s/v/file'"]
+)
+def test_malformed_local_paths_only_reject_transfers(allowed, operation):
     c = _make_client()
     c._kernel_session = MagicMock()
     cursor = MagicMock()
@@ -858,11 +861,21 @@ def test_non_transfer_query_ignores_malformed_local_paths(allowed):
         statement_id="00000000-0000-0000-0000-000000000001",
         arrow_schema=MagicMock(return_value=pa.schema([("x", pa.int64())])),
     )
-    c.execute_command(
-        "SELECT 1", MagicMock(), 1, 1, False, cursor, False, [], False, False
-    )
-    stmt.set_staging_allowed_local_paths.assert_called_once_with([])
-    stmt.execute.assert_called_once()
+
+    def execute():
+        c.execute_command(
+            operation, MagicMock(), 1, 1, False, cursor, False, [], False, False
+        )
+
+    if operation.startswith("PUT"):
+        with pytest.raises(ProgrammingError, match="staging_allowed_local_path"):
+            execute()
+        stmt.execute.assert_not_called()
+        stmt.set_staging_allowed_local_paths.assert_not_called()
+    else:
+        execute()
+        stmt.set_staging_allowed_local_paths.assert_called_once_with([])
+        stmt.execute.assert_called_once()
 
 
 @pytest.mark.parametrize(