Complete configured Volume transfers through the native backend

CHANGELOG.md

@@ -3,6 +3,8 @@
 ## Unreleased
 
 - Reduce result chunking and `fetchAll()` overhead for large query results.
+- Support Volume file transfers on the kernel backend when `stagingAllowedLocalPath` is supplied. Symlink targets must remain within configured allowed roots; in-root symlinks are permitted.
+
 - 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)
 - Honor the client `useCloudFetch: false` default on the kernel backend so complete results can be fetched without cloud-storage downloads. (#574)

CONNECTION_PARAMETERS.md

@@ -197,7 +197,7 @@ telemetry implementation, so only the options explicitly forwarded by
 | `maxRows`                 | `number \| bigint \| Int64 \| null`                     |   ✅   |   ❌   | `100000` / —  | Thrift uses it for the initial Direct Results row count; `null` disables Direct Results. Kernel does not retain or forward the supplied value.                            |
 | `useCloudFetch`           | `boolean`                                               |   ✅   |   ❌   | `true` / —    | Kernel logs and ignores this per-statement hint; Kernel result fetching owns its CloudFetch behavior.                                                                     |
 | `useLZ4Compression`       | `boolean`                                               |   ✅   |   ❌   | `true` / —    | Thrift uses it when supported and when the result is not CloudFetch. Kernel owns and auto-detects result compression.                                                     |
-| `stagingAllowedLocalPath` | `string \| string[]`                                    |   ✅   |   ❌   | —             | Local allowlist for Thrift volume/staging operations. Kernel volume operations are unsupported.                                                                           |
+| `stagingAllowedLocalPath` | `string \| string[]`                                    |   ✅   |   ✅   | —             | Local allowlist for Volume operations. Kernel also requires resolved symlink targets within allowed roots; links within those roots are permitted.                        |
 | `namedParameters`         | `Record<string, DBSQLParameter \| DBSQLParameterValue>` |   ✅   |   ✅   | —             | Named SQL parameters. Cannot be combined with non-empty `ordinalParameters`. Thrift requires a protocol that supports parameterized queries.                              |
 | `ordinalParameters`       | `Array<DBSQLParameter \| DBSQLParameterValue>`          |   ✅   |   ✅   | —             | Positional SQL parameters. Cannot be combined with non-empty `namedParameters`.                                                                                           |
 | `queryTags`               | `Record<string, string \| null \| undefined>`           |   ✅   |   ✅   | —             | Serialized into the per-statement `query_tags` conf overlay on both.                                                                                                      |
@@ -228,11 +228,16 @@ session parameter surface.
 
 Thrift-only capabilities include custom/token-provider authentication, custom
 OAuth persistence, SOCKS proxies, authenticated/driver-buffer telemetry knobs,
-query timeout, Direct Results `maxRows`, per-statement CloudFetch/LZ4 hints, and
-volume operations.
+query timeout, Direct Results `maxRows`, and per-statement CloudFetch/LZ4 hints.
 
 Kernel-only capabilities include connection-pool sizing, proxy bypass hosts,
 independent TLS hostname verification, optional U2M disk caching, and
 per-statement `rowLimit`/`statementConf`.
 
 The Kernel path remains internal and unstable; its behavior may change.
+
+With the kernel backend, supplying `stagingAllowedLocalPath` completes Volume
+transfers before `executeStatement` resolves and consumes instruction rows.
+Omitting the option returns raw instructions without file effects. Unlike the
+Thrift backend’s lexical path check, the kernel also rejects symlink targets
+outside the configured roots while permitting symlinks within them.

KERNEL_REV

@@ -1 +1 @@
-0b5d0edfed32c0745d5abc2468fc3070e2eb2a17
+855847bdca71e07285cfc60332fb2f33f8e47786

lib/DBSQLSession.ts

@@ -109,11 +109,20 @@ export default class DBSQLSession implements IDBSQLSession {
     // operation, everything will work as usual. In a case of staging operation, it will be processed like any
     // other query - it will be possible to get data from it as usual, or use other operation methods.
     if (options.stagingAllowedLocalPath !== undefined) {
+      const allowedLocalPath = Array.isArray(options.stagingAllowedLocalPath)
+        ? options.stagingAllowedLocalPath
+        : [options.stagingAllowedLocalPath];
+      if (opBackend.completeStagingOperation) {
+        try {
+          await this.runBackend(() => opBackend.completeStagingOperation!(allowedLocalPath));
+          return operation;
+        } catch (error) {
+          await operation.close().catch(() => {});
+          throw error;
+        }
+      }
       const metadata = await operation.getResultMetadata();
       if (metadata.isStagingOperation) {
-        const allowedLocalPath = Array.isArray(options.stagingAllowedLocalPath)
-          ? options.stagingAllowedLocalPath
-          : [options.stagingAllowedLocalPath];
         return this.handleStagingOperation(operation, allowedLocalPath, metadata);
       }
     }

lib/contracts/IDBSQLSession.ts

@@ -32,7 +32,8 @@ export type ExecuteStatementOptions = {
    *   `false`/unset (default) runs the blocking direct-results path (faster,
    *   cancellable mid-compute); `true` submits and polls (returns a pending
    *   handle before completion). Default is sync, matching the python
-   *   connector's `cursor.execute()`.
+   *   connector's `cursor.execute()`. With `stagingAllowedLocalPath` configured,
+   *   execution waits for staging completion regardless of this option.
    */
   runAsync?: boolean;
   maxRows?: number | bigint | Int64 | null;

lib/contracts/IOperationBackend.ts

@@ -113,6 +113,8 @@ export default interface IOperationBackend {
    */
   getCachedResultMetadata?(): ResultMetadata | undefined;
 
+  completeStagingOperation?(allowedPaths: string[]): Promise<void>;
+
   /** Cancel the operation. */
   cancel(): Promise<Status>;
 

lib/kernel/KernelErrorMapping.ts

@@ -1,3 +1,5 @@
+import { constants } from 'node:os';
+import { getSystemErrorName } from 'node:util';
 import HiveDriverError from '../errors/HiveDriverError';
 import AuthenticationError from '../errors/AuthenticationError';
 import OperationStateError, { OperationStateErrorCode } from '../errors/OperationStateError';
@@ -311,7 +313,26 @@ export function decodeNapiKernelError(err: unknown): Error {
   const msg = envelope.message as string;
   const sqlState = typeof envelope.sqlState === 'string' ? envelope.sqlState : undefined;
 
-  const jsErr = mapKernelErrorToJsError({ code, message: msg, sqlstate: sqlState });
+  let jsErr = mapKernelErrorToJsError({ code, message: msg, sqlstate: sqlState });
+  if (typeof envelope.localPath === 'string' && typeof envelope.localErrorKind === 'string') {
+    const kindCodes: Record<string, string> = {
+      NotFound: 'ENOENT',
+      PermissionDenied: 'EACCES',
+      AlreadyExists: 'EEXIST',
+      NotADirectory: 'ENOTDIR',
+      IsADirectory: 'EISDIR',
+      DirectoryNotEmpty: 'ENOTEMPTY',
+    };
+    const raw = typeof envelope.localOsError === 'number' ? envelope.localOsError : undefined;
+    const systemCode = process.platform !== 'win32' && raw ? getSystemErrorName(-Math.abs(raw)) : undefined;
+    const localCode =
+      systemCode && /^E[A-Z0-9]+$/.test(systemCode) ? systemCode : kindCodes[envelope.localErrorKind] ?? 'EIO';
+    jsErr = Object.assign(new Error(msg), {
+      code: localCode,
+      errno: -((constants.errno as Record<string, number>)[localCode] ?? Math.abs(raw ?? constants.errno.EIO)),
+      path: envelope.localPath,
+    });
+  }
 
   const meta = buildKernelMetadata(envelope);
   // Skip the namespace attachment entirely when no fields validated

lib/kernel/KernelOperationBackend.ts

@@ -46,12 +46,14 @@ import { LogLevel } from '../contracts/IDBSQLLogger';
 import { RowMode } from '../contracts/IDBSQLSession';
 import Status from '../dto/Status';
 import HiveDriverError from '../errors/HiveDriverError';
+import StagingError from '../errors/StagingError';
+import ParameterError from '../errors/ParameterError';
 import OperationStateError, { OperationStateErrorCode } from '../errors/OperationStateError';
 import ArrowResultConverter from '../result/ArrowResultConverter';
 import ResultSlicer from '../result/ResultSlicer';
 import KernelResultsProvider from './KernelResultsProvider';
 import { arrowSchemaToThriftSchema, decodeIpcSchema, patchIpcBytes } from './KernelArrowIpc';
-import { decodeNapiKernelError } from './KernelErrorMapping';
+import { decodeNapiKernelError, ErrorWithSqlState } from './KernelErrorMapping';
 import {
   KernelStatement,
   KernelNativeAsyncStatement,
@@ -84,7 +86,10 @@ export type KernelOperationStatement = KernelStatementHandle & Partial<KernelSta
  * (`KernelResultsProvider` → `ArrowResultConverter` → `ResultSlicer`) consumes
  * either interchangeably.
  */
-type KernelFetchHandle = Pick<KernelStatement, 'fetchNextBatch' | 'schema'>;
+type KernelFetchHandle = Pick<
+  KernelStatement,
+  'fetchNextBatch' | 'schema' | 'isVolumeOperation' | 'completeVolumeOperation'
+>;
 
 /**
  * The rich operation-status surface the kernel exposes on a terminal sync
@@ -220,6 +225,8 @@ export default class KernelOperationBackend implements IOperationBackend {
   // already-terminal statement. Drives both fetch and result-metadata.
   private fetchHandlePromise?: Promise<KernelFetchHandle>;
 
+  private volumeCompleted = false;
+
   // Async path only: set once the kernel reported the statement Succeeded (a
   // status poll, or `awaitResult()` resolving). The kernel exposes no
   // synchronous terminal-state accessor, so the observed success is retained.
@@ -256,7 +263,7 @@ export default class KernelOperationBackend implements IOperationBackend {
     // drop-guard fires whenever this handle is eventually GC'd.
     this.lifecycleHandle = cancellableExecution
       ? {
-          cancel: () => cancellableExecution.cancel(),
+          cancel: () => (this.blockingStatement ? this.blockingStatement.cancel() : cancellableExecution.cancel()),
           close: () => (this.blockingStatement ? this.blockingStatement.close() : cancellableExecution.cancel()),
         }
       : ((asyncStatement ?? statement) as KernelStatementHandle);
@@ -304,6 +311,28 @@ export default class KernelOperationBackend implements IOperationBackend {
   // Fetch / metadata (owned by the kernel-results pipeline).
   // ---------------------------------------------------------------------------
 
+  public async completeStagingOperation(allowedPaths: string[]): Promise<void> {
+    failIfNotActive(this.lifecycle);
+    const handle = await this.getFetchHandle();
+    try {
+      const outcome = await handle.completeVolumeOperation(allowedPaths, true);
+      if (outcome === 'needs_input_stream') {
+        throw new StagingError('Volume PUT requires a local file path');
+      }
+      this.volumeCompleted = outcome === 'completed';
+    } catch (err) {
+      const mapped = decodeNapiKernelError(err);
+      const metadata = (mapped as ErrorWithSqlState).kernelMetadata;
+      if (
+        (mapped instanceof ParameterError && !metadata?.httpStatus) ||
+        metadata?.errorCode === 'VOLUME_STORAGE_HTTP_ERROR'
+      ) {
+        throw new StagingError(mapped.message);
+      }
+      throw mapped;
+    }
+  }
+
   public async fetchChunk({
     limit,
     disableBuffering,
@@ -316,6 +345,7 @@ export default class KernelOperationBackend implements IOperationBackend {
     // Cancel-mid-fetch propagation: if cancel() has flipped the
     // lifecycle flag, fail locally without a wire round-trip.
     failIfNotActive(this.lifecycle);
+    if (this.volumeCompleted) return [];
     // Cooperative cancel (parity with ThriftOperationBackend): the facade
     // supplies `isClosed` and re-checks `failIfClosed()` after we return, so
     // bailing with `[]` at a yield point is the contract-correct way for a
@@ -356,6 +386,7 @@ export default class KernelOperationBackend implements IOperationBackend {
 
   public async hasMore(): Promise<boolean> {
     failIfNotActive(this.lifecycle);
+    if (this.volumeCompleted) return false;
     const slicer = await this.getResultSlicer();
     return slicer.hasMore();
   }
@@ -394,7 +425,7 @@ export default class KernelOperationBackend implements IOperationBackend {
         // `arrowSchema` doesn't hit apache-arrow@13's `Unrecognized type
         // "Duration" (18)`. Matches what the per-batch fetch path already does.
         arrowSchema: patchIpcBytes(arrowSchemaIpc.ipcBytes),
-        isStagingOperation: false,
+        isStagingOperation: Boolean(handle.isVolumeOperation),
       };
       this.metadata = meta;
       return meta;

lib/kernel/KernelSessionBackend.ts

@@ -151,7 +151,7 @@ export default class KernelSessionBackend implements ISessionBackend {
    * Still rejected (genuinely unsupported on kernel, rather than silently
    * dropped): `useCloudFetch` (governed by the kernel `ResultConfig`, not a
    * per-statement knob), `useLZ4Compression` (kernel owns result compression),
-   * and `stagingAllowedLocalPath` (volume operations). `maxRows` is applied by
+   * while `stagingAllowedLocalPath` is applied by the facade. `maxRows` is applied by
    * the facade at fetch time, so it is intentionally not handled here.
    */
   public async executeStatement(statement: string, options: ExecuteStatementOptions): Promise<IOperationBackend> {
@@ -160,8 +160,7 @@ export default class KernelSessionBackend implements ISessionBackend {
     // Per-statement options the kernel backend doesn't honour are treated as
     // NO-OPs (logged at warn), not errors — so call sites written for the Thrift
     // backend are drop-in on the kernel path. These are perf/format hints the
-    // kernel governs internally (useCloudFetch / useLZ4Compression) or a feature
-    // it doesn't expose yet (staging); ignoring them cannot change query results.
+    // kernel governs internally (useCloudFetch / useLZ4Compression).
     // NB: parameter binding (compound/BINARY) is deliberately NOT no-op'd — a
     // dropped param would silently change results, so it still throws.
     if (options.useCloudFetch !== undefined) {
@@ -180,15 +179,6 @@ export default class KernelSessionBackend implements ISessionBackend {
           'kernel executeStatement: ignoring per-statement `useLZ4Compression` — result compression is governed by the kernel (no-op on kernel).',
         );
     }
-    if (options.stagingAllowedLocalPath !== undefined) {
-      this.context
-        .getLogger()
-        .log(
-          LogLevel.warn,
-          'kernel executeStatement: ignoring `stagingAllowedLocalPath` — volume/staging operations are not yet supported on the kernel backend (no-op).',
-        );
-    }
-
     // `runAsync` selects the kernel execution path. NOTE: this is a kernel-
     // specific use of the option — the Thrift backend hardcodes `runAsync: true`
     // on the wire and never reads `options.runAsync`, so the field is a no-op
@@ -212,7 +202,9 @@ export default class KernelSessionBackend implements ISessionBackend {
     //     the backend polls `status()` to terminal in `waitUntilReady()` and
     //     materialises results via `awaitResult()`. `queryTimeout` is a no-op
     //     here too.
-    const runAsync = options.runAsync ?? false;
+    // Configured staging completes before the facade returns, so retain inline
+    // results even when the caller requests asynchronous execution.
+    const runAsync = (options.runAsync ?? false) && options.stagingAllowedLocalPath === undefined;
     // `queryTimeout` is a NO-OP on the kernel (kernel) backend. It is the JDBC
     // `setQueryTimeout` knob which — per the option's JSDoc — is effective only
     // on Compute clusters; SQL Warehouses (what kernel targets) use the

tests/unit/DBSQLSession.test.ts

@@ -7,6 +7,7 @@ import Status from '../../lib/dto/Status';
 import DBSQLOperation from '../../lib/DBSQLOperation';
 import { PrimaryKeysRequest } from '../../lib/contracts/IDBSQLSession';
 import ISessionBackend from '../../lib/contracts/ISessionBackend';
+import IOperationBackend from '../../lib/contracts/IOperationBackend';
 import ParameterError from '../../lib/errors/ParameterError';
 import { DBSQLParameter } from '../../lib/DBSQLParameter';
 import { TSessionHandle, TProtocolVersion, TOperationState, TStatusCode } from '../../thrift/TCLIService_types';
@@ -883,3 +884,38 @@ describe('DBSQLSession', () => {
     });
   });
 });
+
+describe('DBSQLSession native staging dispatch', () => {
+  it('closes a failed native operation and preserves its error when close also fails', async () => {
+    const error = new Error('completion failed');
+    const close = sinon.stub().rejects(new Error('close failed'));
+    const op = {
+      id: 'volume-operation',
+      completeStagingOperation: sinon.stub().rejects(error),
+      close,
+    } as unknown as IOperationBackend;
+    const session = new DBSQLSession({
+      backend: { id: 'volume-session', executeStatement: async () => op } as unknown as ISessionBackend,
+      context: new ClientContextStub(),
+    });
+    const actual = await session.executeStatement('GET remote TO local', { stagingAllowedLocalPath: [] }).then(
+      () => expect.fail('completion should fail'),
+      (failure) => failure,
+    );
+    expect(actual).to.equal(error);
+    expect(close.callCount).to.equal(1);
+  });
+
+  for (const configured of [false, true]) {
+    it(`calls native completion only when local paths are ${configured ? 'configured' : 'omitted'}`, async () => {
+      const completeStagingOperation = sinon.stub().resolves();
+      const op = { id: 'volume-operation', completeStagingOperation } as unknown as IOperationBackend;
+      const session = new DBSQLSession({
+        backend: { id: 'volume-session', executeStatement: async () => op } as unknown as ISessionBackend,
+        context: new ClientContextStub(),
+      });
+      await session.executeStatement('REMOVE remote', configured ? { stagingAllowedLocalPath: [] } : {});
+      expect(completeStagingOperation.callCount).to.equal(configured ? 1 : 0);
+    });
+  }
+});