feat(volume): enable native SQL file transfers

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Release History
 
 # Unreleased
+- Support SQL PUT, GET, and REMOVE file transfers on the kernel backend with local-path permission checks and native filesystem exceptions. Streaming PUT remains unsupported in kernel mode.
 - 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`. `REMOVE` requires no local path. Kernel mode does not support streaming PUT. |
 
 ## Telemetry
 
@@ -206,13 +206,12 @@ 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`,
+6. 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`.
-8. `force_enable_telemetry`.
-9. `_port` and `_connection_uri`.
-10. `_pool_connections` and `_proxy_auth_method` do not configure the
+7. `force_enable_telemetry`.
+8. `_port` and `_connection_uri`.
+9. `_pool_connections` and `_proxy_auth_method` do not configure the
     kernel-owned network stack.
 
 ### Supported on Kernel, missing / ignored on Thrift

KERNEL_REV

@@ -1 +1 @@
-6f4b6d607a61bdde61dcda2cf82bc8e80fd32cb9
+37c7a3ec1b54f4690d9325d65f2220e0bb64b83e

examples/staging_ingestion.py

@@ -30,7 +30,8 @@
 2. Set the FILEPATH constant to the path of a file that will be uploaded (this example assumes its a CSV file)
 3. Run this file
 
-Note: staging_allowed_local_path can be either a Pathlike object or a list of Pathlike objects.
+Note: staging_allowed_local_path accepts a path string or a list of strings or
+os.PathLike objects that resolve to strings.
 """
 
 INGESTION_USER = "some.user@example.com"

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

@@ -13,10 +13,13 @@
 - Anything else — ``TypeError`` / ``OverflowError`` /
   ``ValueError`` from PyO3 argument conversion, or arbitrary
   extension-internal Python errors. These would otherwise propagate
-  raw to connector callers, breaking the DB-API contract that says
-  "only PEP 249 exception types cross the boundary". Wrapped in
+  raw to connector callers. Wrapped in
   ``OperationalError`` here.
 
+Local transfer failures use built-in ``OSError`` classes, including
+``FileNotFoundError``, ``PermissionError`` and other platform subclasses.
+The original OS error code and requested filename are preserved when available.
+
 These primitives live in their own module so both ``client.py``
 (which orchestrates PyO3 calls) and ``result_set.py`` (which calls
 ``fetch_next_batch`` on the same kernel handles) can share them
@@ -39,10 +42,13 @@
 from __future__ import annotations
 
 import logging
+import os
+from typing import Union
 
 from databricks.sql.exc import (
     DatabaseError,
     Error,
+    NotSupportedError,
     OperationalError,
     ProgrammingError,
     ServerOperationError,
@@ -115,17 +121,35 @@
 }
 
 
-def reraise_kernel_error(exc: "_kernel.KernelError") -> "Error":
-    """Convert a ``databricks_sql_kernel.KernelError`` to a PEP 249
-    exception with the kernel's structured attributes forwarded onto
-    the new instance.
+def reraise_kernel_error(
+    exc: "_kernel.KernelError",
+) -> "Union[Error, OSError]":
+    """Map a kernel error and forward its structured attributes.
+
+    Local transfer failures use built-in ``OSError`` classes, including
+    missing-file, permission, directory and disk-full errors. Non-local
+    errors use PEP 249 classes.
 
     The returned exception is raised by callers with ``raise ... from
     exc``; the ``from`` clause is what sets ``__cause__``, so we don't
     touch it here.
     """
     code = getattr(exc, "code", "Unknown")
     cls = _CODE_TO_EXCEPTION.get(code, DatabaseError)
+    error_code = getattr(exc, "error_code", None)
+    if error_code == "FILE_TRANSFER_UNSUPPORTED_STREAM":
+        cls = NotSupportedError
+    file_transfer = error_code in (
+        "FILE_TRANSFER_ERROR",
+        "FILE_TRANSFER_PERMISSION_ERROR",
+    )
+    if file_transfer:
+        cls = (
+            ProgrammingError
+            if code in ("PermissionDenied", "InvalidArgument")
+            and getattr(exc, "http_status", None) is None
+            else OperationalError
+        )
 
     # For ServerOperationError, reproduce the Thrift backend's
     # ``context`` dict so callers that read
@@ -141,13 +165,49 @@ def reraise_kernel_error(exc: "_kernel.KernelError") -> "Error":
             "operation-id": getattr(exc, "query_id", None),
             "diagnostic-info": getattr(exc, "diagnostic_info", None),
         }
-    new = cls(getattr(exc, "message", str(exc)), context)
+    message = getattr(exc, "message", str(exc))
+    os_error_code = getattr(exc, "os_error_code", None)
+    os_error_filename = getattr(exc, "os_error_filename", None)
+    file_error_args: tuple
+    if os_error_code is None:
+        file_error_args = (
+            (None, message, os_error_filename)
+            if os_error_filename is not None
+            else (message,)
+        )
+    elif os.name == "nt":
+        file_error_args = (0, message, os_error_filename, os_error_code)
+    else:
+        file_error_args = (os_error_code, message, os_error_filename)
+    new: Union[Error, OSError]
+    # Local missing files have no HTTP status; remote missing files carry 404.
+    if (
+        file_transfer
+        and os_error_code is not None
+        and getattr(exc, "http_status", None) is None
+    ):
+        new = OSError(*file_error_args)
+    elif (
+        error_code == "FILE_TRANSFER_PERMISSION_ERROR"
+        and getattr(exc, "http_status", None) is None
+    ):
+        new = PermissionError(*file_error_args)
+    elif (
+        file_transfer
+        and code == "NotFound"
+        and getattr(exc, "http_status", None) is None
+    ):
+        new = FileNotFoundError(*file_error_args)
+    else:
+        new = cls(message, context)
 
     for attr in (
         "code",
         "sql_state",
         "error_code",
         "vendor_code",
+        "os_error_code",
+        "os_error_filename",
         "http_status",
         "retryable",
         "query_id",
@@ -162,12 +222,14 @@ def reraise_kernel_error(exc: "_kernel.KernelError") -> "Error":
     return new
 
 
-def wrap_kernel_exception(what: str, exc: BaseException) -> "Error":
-    """Map any exception from a PyO3 call site to a PEP 249 exception.
+def wrap_kernel_exception(what: str, exc: BaseException) -> "Union[Error, OSError]":
+    """Map a PyO3 exception to the connector's public error contract.
 
     - ``KernelError`` → mapped class with structured attrs forwarded.
-    - Already-PEP-249 ``Error`` (e.g. raised by an inner caller that
-      already mapped) → passed through unchanged.
+      Local transfer failures map to built-in ``OSError`` classes with
+      their original OS error code and requested filename when available.
+    - Already-mapped PEP 249 or local transfer errors → passed through
+      unchanged so nested callers preserve their class and attributes.
     - Anything else (``TypeError`` / ``ValueError`` / etc. from PyO3
       argument conversion, extension-internal errors) → wrapped in
       ``OperationalError``.
@@ -180,6 +242,11 @@ def wrap_kernel_exception(what: str, exc: BaseException) -> "Error":
         return reraise_kernel_error(exc)
     if isinstance(exc, Error):
         return exc
+    if isinstance(exc, OSError) and getattr(exc, "error_code", None) in (
+        "FILE_TRANSFER_ERROR",
+        "FILE_TRANSFER_PERMISSION_ERROR",
+    ):
+        return exc
     return OperationalError(
         f"Unexpected error from databricks_sql_kernel during {what}: {exc!r}"
     )

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

@@ -16,14 +16,14 @@
 
 - ``query_tags`` on execute is not supported (kernel exposes
   ``statement_conf`` but PyO3 doesn't surface it).
-- Volume PUT/GET (staging operations): kernel has no Volume API
-  yet. Users on Thrift-only paths.
 """
 
 from __future__ import annotations
 
 import inspect
 import logging
+import os
+import re
 import threading
 import uuid
 from typing import Any, Dict, List, Optional, Set, TYPE_CHECKING, Union
@@ -73,24 +73,19 @@
 # http_path). Forwarding either is redundant and trips the kernel's
 # 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``.
-_STAGING_VERBS = ("PUT", "GET", "REMOVE")
+_STAGING_PATHS_TYPE_ERROR = (
+    "staging_allowed_local_path must be a string or a list of "
+    "strings or os.PathLike objects"
+)
 
 
 def _strip_leading_sql_comments(sql: str) -> str:
     """Strip leading whitespace and SQL comments (``-- …`` line and
     ``/* … */`` block, possibly several) from ``sql``, returning the
     remainder.
 
-    Needed so staging detection sees the real leading verb: a
-    comment-prefixed staging op (``-- upload\\nPUT …`` or
-    ``/* c */ PUT …``, common in ETL scripts) must still be classified
-    as staging, or it would slip past the guard into the silent-no-op
-    bug. Block comments do not nest in Databricks SQL, so a simple
-    scan-to-``*/`` is correct.
+    Comment-prefixed PUT/GET statements need the permitted local paths
+    just as statements that begin with the verb do.
     """
     i = 0
     n = len(sql)
@@ -102,9 +97,15 @@ def _strip_leading_sql_comments(sql: str) -> str:
             nl = sql.find("\n", i)
             i = n if nl == -1 else nl + 1
         elif sql.startswith("/*", i):
-            # Block comment: skip to closing */ (or end if unterminated).
-            close = sql.find("*/", i + 2)
-            i = n if close == -1 else close + 2
+            depth = 1
+            i += 2
+            for delimiter in re.finditer(r"/\*|\*/", sql[i:]):
+                depth += 1 if delimiter.group() == "/*" else -1
+                if depth == 0:
+                    i += delimiter.end()
+                    break
+            else:
+                i = n
         else:
             break
     return sql[i:]
@@ -145,19 +146,10 @@ def _catalog_or_none(value: Optional[str]) -> Optional[str]:
     return value
 
 
-def _is_staging_statement(operation: str) -> bool:
-    """True iff ``operation`` is a volume/staging statement (PUT / GET /
-    REMOVE).
-
-    Strips leading whitespace + SQL comments first (so a comment-
-    prefixed staging op is still caught), then matches the leading token
-    only — so a normal query that merely *contains* the word (e.g.
-    ``SELECT 'GET' AS x``) isn't misflagged.
-    """
-    stripped = _strip_leading_sql_comments(operation)
-    # First whitespace-delimited token, uppercased.
-    verb = stripped.split(None, 1)[0].upper() if stripped.strip() else ""
-    return verb in _STAGING_VERBS
+def _staging_verb(stripped_sql: str) -> Optional[str]:
+    """Match a leading verb so SELECT 'GET' cannot enable local file access."""
+    match = re.match(r"(PUT|GET|REMOVE)\b", stripped_sql, re.IGNORECASE)
+    return match.group(1).upper() if match else None
 
 
 def _kernel_session_accepts_kwarg(name: str) -> bool:
@@ -575,28 +567,32 @@ def execute_command(
         # ``_async_statements`` and closed by ``close_command``); the sync
         # path drops it in finally. ``close_stmt`` is the post-success
         # decision flag — it stays True on sync, flips to False on async.
-        # Volume/staging (PUT/GET/REMOVE) is not supported on the kernel
-        # path: the kernel returns the staging control row as a normal
-        # result set (``KernelResultSet.is_staging_operation`` is always
-        # False), so the connector's ``_handle_staging_operation`` never
-        # fires and NO file is transferred. Rather than silently no-op
-        # (the Thrift path performs the presigned-URL upload/download),
-        # fail loud at the call site so ETL scripts don't ingest
-        # stale/missing data. Detected by the leading SQL verb — the
-        # only signal available pre-execute, since the kernel exposes no
-        # staging marker today.
-        if _is_staging_statement(operation):
-            raise NotSupportedError(
-                "Volume / staging operations (PUT / GET / REMOVE) are not "
-                "supported on the kernel backend (use_kernel=True); the file "
-                "transfer would silently not happen. Use the Thrift backend "
-                "for staging operations."
-            )
-
         close_stmt = True
         try:
             try:
+                stripped_sql = _strip_leading_sql_comments(operation)
+                staging_verb = _staging_verb(stripped_sql)
                 stmt.set_sql(operation)
+                allowed_paths = (
+                    cursor.connection.staging_allowed_local_path
+                    if staging_verb in ("PUT", "GET")
+                    else None
+                )
+                if allowed_paths is not None and not isinstance(
+                    allowed_paths, (str, list)
+                ):
+                    raise ProgrammingError(_STAGING_PATHS_TYPE_ERROR)
+                if isinstance(allowed_paths, str):
+                    allowed_paths = [allowed_paths]
+                try:
+                    normalized_paths = [os.fspath(path) for path in allowed_paths or []]
+                    if any(not isinstance(path, str) for path in normalized_paths):
+                        raise TypeError("Local paths must resolve to strings")
+                except (TypeError, ValueError) as exc:
+                    raise ProgrammingError(_STAGING_PATHS_TYPE_ERROR) from exc
+                stmt.set_staging_allowed_local_paths(
+                    normalized_paths, input_stream_token="__input_stream__"
+                )
                 stmt.set_row_limit(row_limit)
                 if query_tags:
                     # Per-statement query tags. The kernel serialises the

src/databricks/sql/client.py

@@ -1394,7 +1394,7 @@ def _handle_staging_put(
                 session_id_hex=self.connection.get_session_id_hex(),
             )
 
-        # Kernel backends reject staging operations before reaching these handlers.
+        # Kernel backends transfer files natively and never dispatch to these handlers.
         assert self.connection.http_client is not None
         with open(local_file, "rb") as fh:
             r = self.connection.http_client.request(