fix(volume): reject stream markers and classify remote failures

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

Author: aakash-saravanan-db

CONNECTION_PARAMETERS.md

@@ -206,12 +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).
-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 @@
-bc39c7921acecd95e4eea0740becb883867a834b
+ebb8204b36951b751dd837b2c41ced8c821b2d94

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

@@ -131,7 +131,11 @@ def reraise_kernel_error(
     cls = _CODE_TO_EXCEPTION.get(code, DatabaseError)
     file_transfer = getattr(exc, "error_code", None) == "FILE_TRANSFER_ERROR"
     if file_transfer:
-        cls = ProgrammingError if code == "PermissionDenied" else OperationalError
+        cls = (
+            ProgrammingError
+            if code == "PermissionDenied" and getattr(exc, "http_status", None) is None
+            else OperationalError
+        )
 
     # For ServerOperationError, reproduce the Thrift backend's
     # ``context`` dict so callers that read

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

@@ -22,6 +22,7 @@
 
 import inspect
 import logging
+import re
 import threading
 import uuid
 from typing import Any, Dict, List, Optional, Set, TYPE_CHECKING, Union
@@ -577,6 +578,14 @@ def execute_command(
         close_stmt = True
         try:
             try:
+                if re.match(
+                    r"""PUT\s+(['"])__input_stream__\1(?:\s|$)""",
+                    _strip_leading_sql_comments(operation),
+                    re.IGNORECASE,
+                ):
+                    raise NotSupportedError(
+                        "Streaming PUT is not supported by the kernel backend."
+                    )
                 stmt.set_sql(operation)
                 allowed_paths = cursor.connection.staging_allowed_local_path
                 if isinstance(allowed_paths, str):

tests/unit/test_kernel_client.py

@@ -844,6 +844,25 @@ 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(
+    "query",
+    [
+        "PUT '__input_stream__' INTO '/Volumes/c/s/v/file'",
+        "/* upload */ PUT '__input_stream__' INTO '/Volumes/c/s/v/file'",
+    ],
+)
+def test_kernel_rejects_streaming_put(query):
+    c = _make_client()
+    c._kernel_session = MagicMock()
+    with pytest.raises(NotSupportedError, match="Streaming PUT"):
+        c.execute_command(
+            query, MagicMock(), 1, 1, False, MagicMock(), False, [], False, False
+        )
+    stmt = c._kernel_session.statement.return_value
+    stmt.execute.assert_not_called()
+    stmt.close.assert_called_once()
+
+
 @pytest.mark.parametrize(
     "operation",
     [
@@ -2599,6 +2618,7 @@ def test_only_reserved_headers_omits_kwarg(self, monkeypatch):
     "code, http_status, expected",
     [
         ("PermissionDenied", None, ProgrammingError),
+        ("PermissionDenied", 403, OperationalError),
         ("NotFound", None, FileNotFoundError),
         ("NotFound", 404, OperationalError),
         ("NetworkError", None, OperationalError),