feat(cloudfetch): apply download settings to kernel sessions

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Honor `cloudFetchConcurrentDownloads` when opening kernel sessions, with a default window of 10 (previously the kernel default of 16) and explicit chunk-setting overrides. Values other than positive safe integers raise `HiveDriverError` at session open.
 - Reduce result chunking and `fetchAll()` overhead for large query results.
 - Thrift CloudFetch downloads are now bounded end to end by `retriesTimeout` (default 15 minutes), including every attempt and retry backoff. Previously a stalled download, where the response headers or body never arrived, blocked the fetch forever. A download still running when the deadline expires is aborted and fails with `RetryError` (`TIMEOUT_EXCEEDED`, message `CloudFetch result download failed: Retry timeout exceeded`). (#520)
 - Thrift connections made through `proxy` now apply `checkServerCertificate`, `customCaCert` and `clientCert`/`clientKey`. Previously the proxied path ignored them and verified against Node's process-wide trust store, so a certificate outside the configured trust set could be accepted. **Compatibility:** proxied connections now verify unless `checkServerCertificate: false` is set; as on direct connections, `NODE_TLS_REJECT_UNAUTHORIZED=0` no longer turns verification off. For an `https` proxy these options also apply to the proxy's own certificate. (#485)

CONNECTION_PARAMETERS.md

@@ -130,6 +130,19 @@ still use HTTPS to `host`. Kernel uses `http://host[:port]` for all of its
 requests, including its own OAuth and telemetry calls. CloudFetch downloads use
 the presigned URLs returned by the server on both backends.
 
+## CloudFetch download window
+
+Set `client.getConfig().cloudFetchConcurrentDownloads` before `openSession()`
+to tune downloads on either backend (default 10). Kernel sessions snapshot this
+value at open; changes apply to subsequent sessions. Thrift reads it on each
+fetch. Kernel mode requires a positive safe integer and rejects invalid values
+at session open. Values above 256 are clamped by the kernel.
+
+The kernel bounds downloading and queued chunks ahead of consumption, excluding
+consumer buffers. This is not a byte limit. An explicit
+`cloudfetch_max_chunks_in_memory` in `openSession({ configuration: ... })` wins
+case-insensitively. Invalid raw overrides warn and retain the kernel default 16.
+
 ## Results and type rendering
 
 | Option                        | Type      | Thrift | Kernel | Default Value | Note                                                                                                                                                                               |

lib/contracts/IClientContext.ts

@@ -20,6 +20,9 @@ export interface ClientConfig {
   retryDelayMax: number; // in milliseconds
 
   useCloudFetch: boolean;
+  /** Download window (default 10). Kernel sessions snapshot it at openSession;
+   * it bounds downloading and queued chunks ahead of consumption, excluding
+   * consumer buffers. It is not a byte limit. */
   cloudFetchConcurrentDownloads: number;
   cloudFetchSpeedThresholdMBps: number;
 

lib/kernel/KernelBackend.ts

@@ -152,6 +152,15 @@ export default class KernelBackend implements IBackend {
     if (request.configuration !== undefined) {
       sessionOptions.sessionConf = { ...request.configuration };
     }
+    const configuredWindow = this.context.getConfig().cloudFetchConcurrentDownloads;
+    const downloadWindow = configuredWindow === undefined ? 10 : configuredWindow;
+    if (!Number.isSafeInteger(downloadWindow) || downloadWindow <= 0) {
+      throw new HiveDriverError('cloudFetchConcurrentDownloads must be a positive safe integer');
+    }
+    const chunkKey = 'cloudfetch_max_chunks_in_memory';
+    if (!Object.keys(sessionOptions.sessionConf ?? {}).some((key) => key.toLowerCase() === chunkKey)) {
+      sessionOptions.sessionConf = { ...sessionOptions.sessionConf, [chunkKey]: String(downloadWindow) };
+    }
     if (this.context.getConfig().useCloudFetch === false) {
       sessionOptions.sessionConf = { ...sessionOptions.sessionConf, cloudfetch_enabled: 'false' };
     }

tests/unit/kernel/auth-m2m.test.ts

@@ -381,6 +381,7 @@ describe('KernelAuth + KernelBackend — OAuth M2M auth flow', () => {
       expect(calls).to.have.lengthOf(1);
       expect(calls[0].method).to.equal('openSession');
       expectNativeConnectionOptions(calls[0].args[0], {
+        sessionConf: { cloudfetch_max_chunks_in_memory: '10' },
         hostName: 'example.cloud.databricks.com',
         httpPath: '/sql/1.0/warehouses/abc',
         intervalsAsString: true,

tests/unit/kernel/auth-u2m.test.ts

@@ -223,6 +223,7 @@ describe('KernelAuth + KernelBackend — OAuth U2M auth flow', () => {
       expect(calls).to.have.lengthOf(1);
       expect(calls[0].method).to.equal('openSession');
       expectNativeConnectionOptions(calls[0].args[0], {
+        sessionConf: { cloudfetch_max_chunks_in_memory: '10' },
         hostName: 'example.cloud.databricks.com',
         httpPath: '/sql/1.0/warehouses/abc',
         intervalsAsString: true,

tests/unit/kernel/execution.test.ts

@@ -551,6 +551,7 @@ describe('KernelBackend', () => {
     expectNativeConnectionOptions(args, {
       hostName: 'workspace.example',
       httpPath: '/sql/1.0/warehouses/xyz',
+      sessionConf: { cloudfetch_max_chunks_in_memory: '10' },
       authMode: 'Pat',
       token: 'dapi-token',
       intervalsAsString: true,
@@ -747,12 +748,76 @@ describe('KernelBackend', () => {
       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?.cloudfetch_max_chunks_in_memory).to.equal('10');
       expect(conf?.can_cloud_download).to.equal(undefined);
       expect(conf?.TIMEZONE).to.equal('UTC');
       expect(configuration).to.deep.equal({ TIMEZONE: 'UTC' });
     });
   });
 
+  for (const [window, override] of [
+    [1, undefined],
+    [4, undefined],
+    [1000, undefined],
+    [4, 'cloudfetch_max_chunks_in_memory'],
+    [4, 'CLOUDFETCH_MAX_CHUNKS_IN_MEMORY'],
+  ] as const) {
+    it(`openSession() maps download window ${window}, override=${override}`, async () => {
+      const binding = makeBinding(new FakeNativeConnection());
+      const context = makeContext(undefined, { cloudFetchConcurrentDownloads: window });
+      const backend = new KernelBackend({ context, nativeBinding: binding });
+      await backend.connect({ host: 'h', path: '/p', token: 't' } as ConnectionOptions);
+      const configuration = override ? { [override]: '2' } : {};
+      await backend.openSession({ configuration });
+      const options = binding.openSessionStub.firstCall.args[0] as { sessionConf: Record<string, string> };
+      expect(options.sessionConf).to.deep.equal(
+        override
+          ? { [override]: '2' }
+          : {
+              cloudfetch_max_chunks_in_memory: String(window),
+            },
+      );
+      expect(configuration).to.deep.equal(override ? { [override]: '2' } : {});
+    });
+  }
+
+  for (const window of [0, -1, 1.5, NaN, Infinity, Number.MAX_SAFE_INTEGER + 1, null as unknown as number]) {
+    it(`openSession() rejects invalid download window ${window}`, async () => {
+      const binding = makeBinding(new FakeNativeConnection());
+      const backend = new KernelBackend({
+        context: makeContext(undefined, { cloudFetchConcurrentDownloads: window }),
+        nativeBinding: binding,
+      });
+      await backend.connect({ host: 'h', path: '/p', token: 't' } as ConnectionOptions);
+      try {
+        await backend.openSession({});
+        expect.fail('invalid window accepted');
+      } catch (error) {
+        expect(error).to.be.instanceOf(HiveDriverError);
+        expect((error as Error).message).to.include('positive safe integer');
+      }
+      expect(binding.openSessionStub.called).to.equal(false);
+    });
+  }
+
+  it('snapshots the download window separately for each opened session', async () => {
+    const binding = makeBinding(new FakeNativeConnection());
+    const context = makeContext(undefined, { cloudFetchConcurrentDownloads: 1 });
+    const backend = new KernelBackend({ context, nativeBinding: binding });
+    await backend.connect({ host: 'h', path: '/p', token: 't' } as ConnectionOptions);
+    await backend.openSession({});
+    context.getConfig().cloudFetchConcurrentDownloads = 4;
+    await backend.openSession({});
+    expect(binding.openSessionStub.firstCall.args[0]).to.have.nested.property(
+      'sessionConf.cloudfetch_max_chunks_in_memory',
+      '1',
+    );
+    expect(binding.openSessionStub.secondCall.args[0]).to.have.nested.property(
+      'sessionConf.cloudfetch_max_chunks_in_memory',
+      '4',
+    );
+  });
+
   ['can_cloud_download', 'CLOUDFETCH_ENABLED', 'cloudfetch_enabled'].forEach((key) => {
     it(`openSession() forwards the client disable with ${key}`, async () => {
       const binding = makeBinding(new FakeNativeConnection());
@@ -764,7 +829,12 @@ describe('KernelBackend', () => {
       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(conf).to.deep.equal({
+        [key]: 'true',
+        TIMEZONE: 'UTC',
+        cloudfetch_enabled: 'false',
+        cloudfetch_max_chunks_in_memory: '10',
+      });
       expect(configuration).to.deep.equal({ [key]: 'true', TIMEZONE: 'UTC' });
     });
   });
@@ -841,7 +911,7 @@ describe('KernelBackend', () => {
       token: 't',
       catalog: 'main',
       schema: 'default',
-      sessionConf: { 'spark.sql.execution.arrow.enabled': 'true' },
+      sessionConf: { 'spark.sql.execution.arrow.enabled': 'true', cloudfetch_max_chunks_in_memory: '10' },
     });
 
     // And the SQL still threads through executeStatement (now with no