fix(volume): preserve OS errors and exact stream token matching

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

Author: aakash-saravanan-db

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

@@ -17,7 +17,8 @@
   ``OperationalError`` here.
 
 Local missing transfer files use the built-in ``FileNotFoundError`` class;
-the OS ``errno`` and ``filename`` attributes are unavailable.
+the original OS error code is preserved when available. The ``filename``
+attribute is unavailable.
 
 These primitives live in their own module so both ``client.py``
 (which orchestrates PyO3 calls) and ``result_set.py`` (which calls
@@ -41,6 +42,7 @@
 from __future__ import annotations
 
 import logging
+import os
 from typing import Union
 
 from databricks.sql.exc import (
@@ -160,19 +162,26 @@ def reraise_kernel_error(
             "diagnostic-info": getattr(exc, "diagnostic_info", None),
         }
     message = getattr(exc, "message", str(exc))
+    os_error_code = getattr(exc, "os_error_code", None)
+    if os_error_code is None:
+        file_error_args = (message,)
+    elif os.name == "nt":
+        file_error_args = (0, message, None, os_error_code)
+    else:
+        file_error_args = (os_error_code, message)
     new: Union[Error, OSError]
     # Local missing files have no HTTP status; remote missing files carry 404.
     if (
         error_code == "FILE_TRANSFER_PERMISSION_ERROR"
         and getattr(exc, "http_status", None) is None
     ):
-        new = PermissionError(message)
+        new = PermissionError(*file_error_args)
     elif (
         file_transfer
         and code == "NotFound"
         and getattr(exc, "http_status", None) is None
     ):
-        new = FileNotFoundError(message)
+        new = FileNotFoundError(*file_error_args)
     else:
         new = cls(message, context)
 
@@ -181,6 +190,7 @@ def reraise_kernel_error(
         "sql_state",
         "error_code",
         "vendor_code",
+        "os_error_code",
         "http_status",
         "retryable",
         "query_id",
@@ -199,8 +209,8 @@ def wrap_kernel_exception(what: str, exc: BaseException) -> "Union[Error, OSErro
     """Map a PyO3 exception to the connector's public error contract.
 
     - ``KernelError`` → mapped class with structured attrs forwarded.
-      A local missing transfer file maps to ``FileNotFoundError`` without
-      the OS ``errno`` or ``filename`` attributes.
+      A local missing transfer file maps to ``FileNotFoundError`` with
+      its original OS error code when available, without ``filename``.
     - 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

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

@@ -575,7 +575,6 @@ def execute_command(
                 if staging_verb == "PUT" and re.match(
                     r"""(['"])__input_stream__\1""",
                     _strip_leading_sql_comments(stripped_sql[3:]),
-                    re.IGNORECASE,
                 ):
                     raise NotSupportedError(
                         "Streaming PUT is not supported by the kernel backend."

tests/unit/test_kernel_client.py

@@ -13,6 +13,8 @@
 
 from __future__ import annotations
 
+import errno
+import os
 import sys
 import types
 from pathlib import Path
@@ -820,6 +822,7 @@ def test_execute_command_forwards_row_limit(row_limit):
     "operation, uses_paths",
     [
         ("PUT '/tmp/files/input' INTO '/Volumes/c/s/v/input'", True),
+        ("PUT '__INPUT_STREAM__' INTO '/Volumes/c/s/v/input'", True),
         ("GET '/Volumes/c/s/v/input' TO '/tmp/files/output'", True),
         ("/* download */ GET '/Volumes/c/s/v/input' TO '/tmp/files/output'", True),
         (
@@ -948,6 +951,7 @@ def test_staging_verb_matches_only_the_leading_statement(operation, expected):
     "query",
     [
         "PUT '__input_stream__' INTO '/Volumes/c/s/v/file'",
+        "put '__input_stream__' INTO '/Volumes/c/s/v/file'",
         "PUT '__input_stream__'INTO '/Volumes/c/s/v/file'",
         'PUT"__input_stream__"INTO "/Volumes/c/s/v/file"',
         "/* upload */ PUT '__input_stream__' INTO '/Volumes/c/s/v/file'",
@@ -2750,6 +2754,43 @@ def test_file_transfer_error_matches_thrift_exception(
     assert result.error_code == error_code
     assert result.code == code
     assert result.http_status == http_status
+    if issubclass(expected, OSError):
+        assert result.errno is None
+
+
+@pytest.mark.parametrize(
+    "code,error_code,raw,expected",
+    [
+        ("NotFound", "FILE_TRANSFER_ERROR", errno.ENOENT, FileNotFoundError),
+        (
+            "PermissionDenied",
+            "FILE_TRANSFER_PERMISSION_ERROR",
+            errno.EACCES,
+            PermissionError,
+        ),
+        (
+            "PermissionDenied",
+            "FILE_TRANSFER_PERMISSION_ERROR",
+            errno.EPERM,
+            PermissionError,
+        ),
+    ],
+)
+def test_local_transfer_error_preserves_os_code(code, error_code, raw, expected):
+    error = _FakeKernelError(code=code, message="File transfer failed")
+    error.error_code = error_code
+    if os.name == "nt":
+        raw = 2 if expected is FileNotFoundError else 5
+    error.os_error_code = raw
+    result = kernel_client._reraise_kernel_error(error)
+    assert type(result) is expected
+    assert result.os_error_code == raw
+    assert result.filename is None
+    if os.name == "nt":
+        assert result.winerror == raw
+        assert result.errno == (errno.ENOENT if raw == 2 else errno.EACCES)
+    else:
+        assert result.errno == raw
 
 
 @pytest.mark.parametrize(