fix(kernel): honor use_cloud_fetch

Signed-off-by: Vu Anh Phung <vu.phung@databricks.com>

Author: vuanhphung

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

@@ -230,6 +230,7 @@ def __init__(
         self._use_arrow_native_complex_types = kwargs.get(
             "_use_arrow_native_complex_types", True
         )
+        self._use_cloud_fetch = kwargs.get("use_cloud_fetch", True)
         # NB: don't call ``kernel_auth_kwargs`` here. That call
         # materialises the bearer token in-process; keeping a
         # cleartext copy on a long-lived connector object that may
@@ -293,12 +294,16 @@ def open_session(
     ) -> SessionId:
         if self._kernel_session is not None:
             raise InterfaceError("KernelDatabricksClient already has an open session.")
-        # ``session_configuration`` flows through to the kernel's
-        # ``session_conf`` map verbatim; the SEA endpoint enforces
-        # its own allow-list and rejects unknown keys.
-        session_conf: Optional[Dict[str, str]] = None
-        if session_configuration:
-            session_conf = {k: str(v) for k, v in session_configuration.items()}
+        # Convert server session confs to strings, then add the kernel's
+        # client-side CloudFetch knob to the same boundary map.
+        session_conf = (
+            {k: str(v) for k, v in session_configuration.items()}
+            if session_configuration
+            else {}
+        )
+        # The kernel consumes this before filtering the server confs and
+        # selects INLINE when CloudFetch is disabled.
+        session_conf["cloudfetch_enabled"] = str(self._use_cloud_fetch).lower()
         # The kwarg builds run INSIDE the try so the ``finally`` scrub
         # below always fires — including when ``kernel_auth_kwargs``
         # itself raises mid-build (e.g. an OAuth token-exchange failure

src/databricks/sql/session.py

@@ -204,6 +204,7 @@ def _create_backend(
                 http_client=self.http_client,
                 catalog=kwargs.get("catalog"),
                 schema=kwargs.get("schema"),
+                use_cloud_fetch=kwargs.get("use_cloud_fetch", True),
                 _use_arrow_native_complex_types=_use_arrow_native_complex_types,
                 auth_options=kernel_auth_options,
                 retry_options=kernel_retry_options,

tests/unit/test_kernel_client.py

@@ -344,6 +344,42 @@ def fake_session(**kw):
     assert captured.get("complex_types_as_json") is expected_flag
 
 
+@pytest.mark.parametrize(
+    "use_cloud_fetch, expected",
+    [(True, "true"), (False, "false")],
+)
+def test_open_session_passes_cloud_fetch_setting_to_kernel(
+    monkeypatch, use_cloud_fetch, expected
+):
+    captured = {}
+
+    def fake_session(**kw):
+        captured.update(kw)
+        sess = MagicMock()
+        sess.session_id = "sess-id"
+        return sess
+
+    monkeypatch.setattr(kernel_client._kernel, "Session", fake_session)
+
+    c = kernel_client.KernelDatabricksClient(
+        server_hostname="example.cloud.databricks.com",
+        http_path="/sql/1.0/warehouses/abc",
+        auth_provider=AccessTokenAuthProvider("dapi-test"),
+        ssl_options=None,
+        use_cloud_fetch=use_cloud_fetch,
+    )
+    c.open_session(
+        session_configuration={"ANSI_MODE": "false"},
+        catalog=None,
+        schema=None,
+    )
+
+    assert captured["session_conf"] == {
+        "ANSI_MODE": "false",
+        "cloudfetch_enabled": expected,
+    }
+
+
 def test_execute_command_forwards_parameters_to_bind_param():
     """``execute_command(parameters=[...])`` routes each parameter
     through ``bind_tspark_params`` onto the kernel statement before

tests/unit/test_session.py

@@ -478,6 +478,44 @@ def test_retry_kwargs_threaded_into_kernel_client(self):
                 conn.close()
 
 
+class TestKernelCloudFetchThreading:
+    def test_use_cloud_fetch_threaded_into_kernel_client(self):
+        import sys
+        import types
+
+        pytest.importorskip(
+            "pyarrow",
+            reason="kernel client module imports pyarrow at load",
+        )
+
+        fake = types.ModuleType("databricks_sql_kernel")
+        fake.KernelError = type("KernelError", (Exception,), {})
+        fake.Session = MagicMock()
+
+        with patch.dict(sys.modules, {"databricks_sql_kernel": fake}), patch(
+            "databricks.sql.backend.kernel.client.KernelDatabricksClient"
+        ) as mock_kernel_client, patch(
+            "databricks.sql.session.get_python_sql_connector_auth_provider"
+        ):
+            instance = mock_kernel_client.return_value
+            instance.open_session.return_value = SessionId(
+                BackendType.SEA, "sess-id", None
+            )
+
+            conn = databricks.sql.connect(
+                server_hostname="foo",
+                http_path="/sql/1.0/warehouses/abc",
+                use_kernel=True,
+                use_cloud_fetch=False,
+                access_token="dapi-xyz",
+                enable_telemetry=False,
+            )
+            try:
+                assert mock_kernel_client.call_args.kwargs["use_cloud_fetch"] is False
+            finally:
+                conn.close()
+
+
 class TestKernelUserAgentForwarding:
     """user_agent_entry must reach the kernel on the use_kernel path —
     session.py folds it into the composed User-Agent and includes it in