fix(kernel): honor the client CloudFetch default

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Honor the client `useCloudFetch: false` default on the kernel backend so complete results can be fetched without cloud-storage downloads. (#574)
 - Thrift HTTP errors (`THTTPException`) now append server detail from the error headers (at most 500 characters). Kernel-backend errors now carry a non-enumerable `cause` summarizing the kernel error code and message (also at most 500 characters). Reyden fallback preserves KP001 after any existing kernel cause. (#467)
 - Add `Symbol.asyncDispose` to clients, sessions, and operations, delegating to `close()`. Use `asyncDisposable()` with interface-typed results in TypeScript; existing interfaces and factory return types remain unchanged. (#331)
 - CloudFetch retry-exhaustion errors now identify the download and the last failure: the `RetryError` message reads `CloudFetch result download failed: <original message> (last error: <cause>)`. `<cause>` is `HTTP <status>: <statusText>` (or `HTTP <status>` when the status text is empty) for HTTP failures, and `network error <code>`, `network error (<type>)` (e.g. `aborted`, when there is no `code`) or a bare `network error` for network failures (the raw network-error message, which can contain the presigned URL, is not copied). If no cause can be determined, the ` (last error: …)` suffix is omitted. The `RetryError` instance, `errorCode` and `payload` (including the node-fetch `Request` / `Response`) are unchanged. (#471)

CONNECTION_PARAMETERS.md

@@ -47,6 +47,11 @@ backend (the default), the **SEA / Kernel** backend, or both.
 | `useKernel` (internal)  | `boolean` |   —    |   —    | `false`       | Selects the backend: `false`/omitted → Thrift, `true` → Kernel.                        |
 | `clientId` (deprecated) | `string`  |   ✅   |   ✅   | —             | Runtime-only alias for `userAgentEntry`. `userAgentEntry` wins when both are supplied. |
 
+Set `client.getConfig().useCloudFetch = false` (default `true`) before opening a session to fetch
+complete results without cloud-storage downloads. On the kernel backend this
+connection setting takes precedence over conflicting session configuration.
+Per-statement `useCloudFetch` overrides remain unsupported on the kernel backend.
+
 ## Authentication
 
 | Option                                        | Type                                                         | Thrift | Kernel | Default Value                 | Note                                                                                                                                                                                                                       |

KERNEL_REV

@@ -1 +1 @@
-6f4b6d607a61bdde61dcda2cf82bc8e80fd32cb9
+0b5d0edfed32c0745d5abc2468fc3070e2eb2a17

lib/kernel/KernelBackend.ts

@@ -152,6 +152,9 @@ export default class KernelBackend implements IBackend {
     if (request.configuration !== undefined) {
       sessionOptions.sessionConf = { ...request.configuration };
     }
+    if (this.context.getConfig().useCloudFetch === false) {
+      sessionOptions.sessionConf = { ...sessionOptions.sessionConf, cloudfetch_enabled: 'false' };
+    }
     // Session-level query tags: serialize into the reserved `QUERY_TAGS`
     // session conf. The kernel allowlists `QUERY_TAGS` (SESSION_CONF_ALLOWLIST)
     // and forwards it onto the SEA `CreateSession` `session_confs`, mirroring

tests/e2e/kernel/cloudfetch-disabled-e2e.test.ts

@@ -0,0 +1,116 @@
+import { createHash } from 'crypto';
+import { createServer } from 'http';
+import { AddressInfo, connect, Socket } from 'net';
+import { expect } from 'chai';
+import { DBSQLClient } from '../../../lib';
+import { ConnectionOptions } from '../../../lib/contracts/IDBSQLClient';
+import { InternalConnectionOptions } from '../../../lib/contracts/InternalConnectionOptions';
+import { tryGetKernelNative } from '../../../lib/kernel/KernelNativeLoader';
+
+describe('kernel CloudFetch configuration', function suite() {
+  this.timeout(120_000);
+
+  const host = process.env.DATABRICKS_PECOTESTING_SERVER_HOSTNAME;
+  const path = process.env.DATABRICKS_PECOTESTING_HTTP_PATH;
+  const token = process.env.DATABRICKS_PECOTESTING_TOKEN_PERSONAL;
+
+  before(function gate() {
+    if (!host || !path || !token || !tryGetKernelNative()) {
+      // eslint-disable-next-line no-invalid-this
+      this.skip();
+    }
+  });
+
+  for (const useCloudFetch of [false, true]) {
+    it(
+      useCloudFetch ? 'requires storage access when enabled' : 'returns complete large results when disabled',
+      async () => {
+        let workspaceConnections = 0;
+        let blockedConnections = 0;
+        const sockets = new Set<Socket>();
+        const proxy = createServer();
+        proxy.on('connection', (socket) => {
+          sockets.add(socket);
+          socket.on('close', () => sockets.delete(socket));
+        });
+        proxy.on('connect', (request, socket, head) => {
+          if (request.url !== `${host}:443`) {
+            blockedConnections += 1;
+            socket.end('HTTP/1.1 403 Forbidden\r\nConnection: close\r\n\r\n');
+            return;
+          }
+          const upstream = connect(443, host as string, () => {
+            workspaceConnections += 1;
+            socket.write('HTTP/1.1 200 Connection Established\r\n\r\n');
+            upstream.write(head);
+            upstream.pipe(socket);
+            socket.pipe(upstream);
+          });
+          socket.on('close', () => upstream.destroy());
+          socket.on('error', () => upstream.destroy());
+          upstream.on('error', () => socket.destroy());
+        });
+        await new Promise<void>((resolve) => {
+          proxy.listen(0, '127.0.0.1', resolve);
+        });
+        const { port } = proxy.address() as AddressInfo;
+        const client = new DBSQLClient({ logger: { log() {} } });
+        client.getConfig().useCloudFetch = useCloudFetch;
+        try {
+          await client.connect({
+            host: host as string,
+            path: path as string,
+            token: token as string,
+            useKernel: true,
+            proxy: { protocol: 'http', host: '127.0.0.1', port },
+          } as ConnectionOptions & InternalConnectionOptions);
+          const session = await client.openSession();
+          try {
+            let failure: unknown;
+            try {
+              const operation = await session.executeStatement(
+                'SELECT id, sha2(CAST(id AS STRING), 512) AS payload FROM range(250000) ORDER BY id',
+              );
+              try {
+                const rows = (await operation.fetchAll()) as Array<{ id: number | bigint; payload: string }>;
+                const digest = createHash('sha256');
+                let payloadBytes = 0;
+                rows.forEach((row, index) => {
+                  expect(Number(row.id)).to.equal(index);
+                  expect(row.payload).to.equal(createHash('sha512').update(String(index)).digest('hex'));
+                  digest.update(`${index}\t${row.payload}\n`);
+                  payloadBytes += row.payload.length;
+                });
+                expect(rows.length).to.equal(250000);
+                expect(payloadBytes).to.equal(32000000);
+                expect(digest.digest('hex')).to.equal(
+                  'a25ada4c6c476b53e97eb6e8a945e9ce4c14cc432c7ad672dc6fb607d660c86a',
+                );
+              } finally {
+                await operation.close();
+              }
+            } catch (error) {
+              failure = error;
+            }
+            expect(workspaceConnections).to.be.greaterThan(0);
+            if (useCloudFetch) {
+              expect(failure).to.not.equal(undefined);
+              expect(blockedConnections).to.be.greaterThan(0);
+            } else {
+              if (failure) throw failure;
+              expect(blockedConnections).to.equal(0);
+            }
+          } finally {
+            await session.close();
+          }
+        } finally {
+          await client.close();
+          sockets.forEach((socket) => socket.destroy());
+          await new Promise<void>((resolve, reject) => {
+            proxy.close((error) => (error ? reject(error) : resolve()));
+          });
+        }
+      },
+    );
+  }
+});

tests/unit/kernel/execution.test.ts

@@ -738,6 +738,37 @@ describe('KernelBackend', () => {
     }
   });
 
+  [undefined, true, false].forEach((useCloudFetch) => {
+    it(`openSession() honors useCloudFetch=${useCloudFetch}`, async () => {
+      const binding = makeBinding(new FakeNativeConnection());
+      const backend = new KernelBackend({ context: makeContext(undefined, { useCloudFetch }), nativeBinding: binding });
+      await backend.connect({ host: 'h', path: '/p', token: 't' } as ConnectionOptions);
+      const configuration = { TIMEZONE: 'UTC' };
+      await backend.openSession({ configuration });
+      const conf = (binding.openSessionStub.firstCall.args[0] as { sessionConf?: Record<string, string> }).sessionConf;
+      expect(conf?.cloudfetch_enabled).to.equal(useCloudFetch === false ? 'false' : undefined);
+      expect(conf?.can_cloud_download).to.equal(undefined);
+      expect(conf?.TIMEZONE).to.equal('UTC');
+      expect(configuration).to.deep.equal({ TIMEZONE: 'UTC' });
+    });
+  });
+
+  ['can_cloud_download', 'CLOUDFETCH_ENABLED', 'cloudfetch_enabled'].forEach((key) => {
+    it(`openSession() forwards the client disable with ${key}`, async () => {
+      const binding = makeBinding(new FakeNativeConnection());
+      const backend = new KernelBackend({
+        context: makeContext(undefined, { useCloudFetch: false }),
+        nativeBinding: binding,
+      });
+      await backend.connect({ host: 'h', path: '/p', token: 't' } as ConnectionOptions);
+      const configuration = { [key]: 'true', TIMEZONE: 'UTC' };
+      await backend.openSession({ configuration });
+      const conf = (binding.openSessionStub.firstCall.args[0] as { sessionConf?: Record<string, string> }).sessionConf;
+      expect(conf).to.deep.equal({ [key]: 'true', TIMEZONE: 'UTC', cloudfetch_enabled: 'false' });
+      expect(configuration).to.deep.equal({ [key]: 'true', TIMEZONE: 'UTC' });
+    });
+  });
+
   it('openSession() serializes session-level queryTags into sessionConf.QUERY_TAGS', async () => {
     const connection = new FakeNativeConnection();
     const binding = makeBinding(connection);