fix(kernel): preserve ordinary queries with older native wheels

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

Author: aakash-saravanan-db

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Release History
 
 # Unreleased
+- Support SQL PUT, GET, and REMOVE file transfers with a native kernel exposing `Statement.set_staging_allowed_local_paths`; older wheels reject transfers and continue to support ordinary queries.
 - Fix inline parameters (`use_inline_params=True`) so a sequence directly after `IN` renders as a list again: `x IN %(ids)s` renders as `x IN (1,2,3)` instead of `x IN ARRAY(1,2,3)`, which Spark rejects. Inside the list, tuples render as rows for a multi-column `IN`, and a `set` or `frozenset` is accepted again. Sequences anywhere else still render as `ARRAY(...)` ([#629](https://github.com/databricks/databricks-sql-python/issues/629))
 
 # 4.9.0 (2026-11-04)

CONNECTION_PARAMETERS.md

@@ -171,7 +171,7 @@ to change without notice.
 | `use_inline_params`           | `bool` \| `"silent"`              |   ✅   |   ✅   | `False`       | Render parameters inline (legacy) vs. native bound params (DBR 14.1+). Shared cursor code performs the rendering before dispatch, so both backends honor it identically. |
 | `ignore_transactions`         | `bool`                            |   ✅   |   ✅   | `True`        | When `True`: `commit()` is a no-op, `rollback()` raises `NotSupportedError`, and setting `autocommit` is a no-op.          |
 | `fetch_autocommit_from_server`| `bool`                            |   ✅   |   ✅   | `False`       | Query the server (`SET AUTOCOMMIT`) for autocommit state instead of returning the cached value.                           |
-| `staging_allowed_local_path`  | `str` \| `List[str]`              |   ✅   |   ❌   | `None`        | Local path(s) permitted for Unity Catalog Volume `PUT`/`GET`. **Thrift-only** — the kernel has no Volume API yet.          |
+| `staging_allowed_local_path`  | `str` \| `List[str]`              |   ✅   |   ✅   | `None`        | Local path(s) permitted for Unity Catalog Volume `PUT`/`GET`. Kernel transfers require a wheel exposing `Statement.set_staging_allowed_local_paths` (kernel [#439](https://github.com/databricks/databricks-sql-kernel/pull/439)); older wheels reject transfer commands.          |
 
 ## Telemetry
 
@@ -206,7 +206,6 @@ installed kernel binding. `force_enable_telemetry` remains Python-driver-only.
    `enable_query_result_lz4_compression`.
 5. Arrow-native rendering for `_use_arrow_native_decimals` /
    `_use_arrow_native_timestamps` (complex types **are** forwarded).
-6. `staging_allowed_local_path` (Volume `PUT`/`GET`).
 7. Retry fine-tuning beyond the four forwarded knobs: `_retry_delay_default`,
    `_retry_dangerous_codes`, `_respect_server_retry_after_header`,
    `_retry_max_redirects`, `_enable_v3_retries`.

KERNEL_REV

@@ -1 +1 @@
-d41cc3657748966d8b69a802bc40e8d019022ceb
+bc39c7921acecd95e4eea0740becb883867a834b

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

@@ -148,6 +148,7 @@ def reraise_kernel_error(
             "diagnostic-info": getattr(exc, "diagnostic_info", None),
         }
     message = getattr(exc, "message", str(exc))
+    # Local missing files have no HTTP status; remote missing files carry 404.
     if (
         file_transfer
         and code == "NotFound"

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

@@ -73,6 +73,30 @@
 _KERNEL_MANAGED_HEADERS = frozenset({"authorization", "x-databricks-org-id"})
 
 
+def _strip_leading_sql_comments(sql: str) -> str:
+    """Skip SQL comments so unsupported transfers cannot appear successful."""
+    i = 0
+    n = len(sql)
+    while i < n:
+        if sql[i].isspace():
+            i += 1
+        elif sql.startswith("--", i):
+            nl = sql.find("\n", i)
+            i = n if nl == -1 else nl + 1
+        elif sql.startswith("/*", i):
+            close = sql.find("*/", i + 2)
+            i = n if close == -1 else close + 2
+        else:
+            break
+    return sql[i:]
+
+
+def _is_staging_statement(operation: str) -> bool:
+    stripped = _strip_leading_sql_comments(operation)
+    verb = stripped.split(None, 1)[0].upper() if stripped.strip() else ""
+    return verb in ("PUT", "GET", "REMOVE")
+
+
 def _is_not_found(exc: BaseException) -> bool:
     """True iff ``exc`` is a kernel ``NotFound`` error (HTTP 404 /
     ``STATEMENT_NOT_FOUND``).
@@ -530,7 +554,14 @@ def execute_command(
                 allowed_paths = cursor.connection.staging_allowed_local_path
                 if isinstance(allowed_paths, str):
                     allowed_paths = [allowed_paths]
-                stmt.set_staging_allowed_local_paths(allowed_paths or [])
+                set_paths = getattr(stmt, "set_staging_allowed_local_paths", None)
+                if callable(set_paths):
+                    set_paths(allowed_paths or [])
+                elif _is_staging_statement(operation):
+                    raise NotSupportedError(
+                        "Volume transfers require a databricks-sql-kernel wheel "
+                        "with staging_allowed_local_paths support."
+                    )
                 stmt.set_row_limit(row_limit)
                 if query_tags:
                     # Per-statement query tags. The kernel serialises the

tests/unit/test_kernel_client.py

@@ -844,6 +844,46 @@ def test_execute_command_forwards_allowed_local_paths(allowed, expected, async_o
     (stmt.submit if async_op else stmt.execute).assert_called_once()
 
 
+@pytest.mark.parametrize(
+    "operation",
+    [
+        "SELECT 1",
+        "PUT 'a' INTO 'b'",
+        "-- comment\nGET 'a' TO 'b'",
+        "/* comment */ REMOVE 'a'",
+    ],
+)
+def test_execute_command_with_older_kernel(operation):
+    c = _make_client()
+    c._kernel_session = MagicMock()
+    stmt = c._kernel_session.statement.return_value
+    del stmt.set_staging_allowed_local_paths
+    stmt.execute.return_value = MagicMock(
+        statement_id="00000000-0000-0000-0000-000000000001",
+        arrow_schema=MagicMock(return_value=pa.schema([("x", pa.int64())])),
+    )
+    kwargs = dict(
+        operation=operation,
+        session_id=MagicMock(),
+        max_rows=1,
+        max_bytes=1,
+        lz4_compression=False,
+        cursor=MagicMock(),
+        use_cloud_fetch=False,
+        parameters=[],
+        async_op=False,
+        enforce_embedded_schema_correctness=False,
+    )
+    if operation == "SELECT 1":
+        c.execute_command(**kwargs)
+        stmt.execute.assert_called_once()
+    else:
+        with pytest.raises(NotSupportedError, match="Volume transfers require"):
+            c.execute_command(**kwargs)
+        stmt.execute.assert_not_called()
+        stmt.close.assert_called_once()
+
+
 # ---------------------------------------------------------------------------
 # Sync cancel wiring (cursor.cancel() during a blocking execute())
 # ---------------------------------------------------------------------------