[SEA-NodeJS] configurable sync/async execute via runAsync (default sync)

Make the SEA execution path user-configurable between sync and async,
toggled by the EXISTING `runAsync` option — no new public field, exactly
mirroring the Thrift backend's `runAsync` distinction. Default is SYNC
(`runAsync: false`): faster, and with the kernel sync canceller fully
cancellable mid-compute.

SeaSessionBackend.executeStatement now branches on `options.runAsync`:
  - false/undefined (DEFAULT) -> Connection.executeStatementCancellable:
    the kernel blocks on execute() (poll-to-terminal server-side), driven
    lazily in the operation backend's result(). `queryTimeoutSecs` IS
    forwarded (the kernel execute() honours it).
  - true -> Connection.submitStatement (submit + poll), unchanged.
    `queryTimeoutSecs` stays client-side (kernel ignores it on submit).

SeaOperationBackend gains a third dual-mode handle kind,
`cancellableExecution`, alongside `asyncStatement` and `statement`:
  - waitUntilReadyCancellable drives result() to the terminal Statement
    (memoised as the fetch handle + close target);
  - the lifecycle handle is a composite: cancel() routes to the detached
    canceller (lock-free, interrupts a running result() mid-COMPUTE and
    is a no-op once terminal); close() routes to the resolved statement;
  - a cancel-induced result() rejection maps to OperationStateError(
    Canceled) so the DBSQLOperation facade mirrors its cancelled flag,
    matching the Thrift path.

Public API, result shape, schema (TTableSchema), and error classes are
identical across both modes and to Thrift — the only observable
difference is lifecycle timing (when executeStatement resolves).

Built against databricks-sql-kernel napi
639e19ef97decc1c5aa2365c0b3a229c1ccd5b58 (executeStatementCancellable /
CancellableExecution); KERNEL_REV bumped to match. Refreshed the
committed binding surface (native/sea/index.{js,d.ts}).

Co-authored-by: Isaac
Signed-off-by: Madhavendra Rathore <madhavendra.rathore@databricks.com>

Author: msrathore-db

KERNEL_REV

@@ -0,0 +1 @@
+639e19ef97decc1c5aa2365c0b3a229c1ccd5b58

lib/sea/SeaNativeLoader.ts

@@ -38,6 +38,7 @@ import type {
   NamedTypedValueInput as NativeNamedTypedValueInput,
   AsyncStatement as NativeAsyncStatement,
   AsyncResultHandle as NativeAsyncResultHandle,
+  CancellableExecution as NativeCancellableExecution,
 } from '../../native/sea';
 
 // SEA-prefixed re-exports. The kernel-generated `.d.ts` keeps the
@@ -69,6 +70,14 @@ export type SeaNativeNamedTypedValueInput = NativeNamedTypedValueInput;
 export type SeaNativeAsyncStatement = NativeAsyncStatement;
 export type SeaNativeAsyncResultHandle = NativeAsyncResultHandle;
 
+// Cancellable sync-execute surface: `Connection.executeStatementCancellable`
+// returns a `CancellableExecution` that captures a detached StatementCanceller
+// BEFORE dispatching the blocking `execute()`, so a concurrent `cancel()`
+// interrupts a still-running query mid-compute. `result()` drives the blocking
+// execute and resolves to the same terminal `Statement` `executeStatement`
+// returns.
+export type SeaNativeCancellableExecution = NativeCancellableExecution;
+
 /**
  * The full native binding surface, derived from the generated module
  * so it can never drift from the `.d.ts` contract: when the kernel

lib/sea/SeaOperationBackend.ts

@@ -51,7 +51,12 @@ import ResultSlicer from '../result/ResultSlicer';
 import SeaResultsProvider from './SeaResultsProvider';
 import { arrowSchemaToThriftSchema, decodeIpcSchema, patchIpcBytes } from './SeaArrowIpc';
 import { decodeNapiKernelError } from './SeaErrorMapping';
-import { SeaStatement, SeaNativeAsyncStatement, SeaNativeAsyncResultHandle } from './SeaNativeLoader';
+import {
+  SeaStatement,
+  SeaNativeAsyncStatement,
+  SeaNativeAsyncResultHandle,
+  SeaNativeCancellableExecution,
+} from './SeaNativeLoader';
 import {
   SeaStatementHandle,
   SeaOperationLifecycleState,
@@ -116,6 +121,15 @@ export interface SeaOperationBackendOptions {
   asyncStatement?: SeaNativeAsyncStatement;
   /** The terminal napi `Statement` from a metadata call. */
   statement?: SeaOperationStatement;
+  /**
+   * The pending napi `CancellableExecution` from
+   * `Connection.executeStatementCancellable(...)` — the sync (`runAsync: false`)
+   * query path. `result()` drives the blocking `execute()` to a terminal
+   * `Statement` (the fetch handle); `cancel()` fires a detached canceller that
+   * interrupts a still-running `result()` mid-COMPUTE. Exactly one of
+   * `asyncStatement`, `statement`, or `cancellableExecution` must be set.
+   */
+  cancellableExecution?: SeaNativeCancellableExecution;
   context: IClientContext;
   /**
    * Optional override for `id`. Defaults to the napi statement-id when the
@@ -134,15 +148,23 @@ export interface SeaOperationBackendOptions {
 }
 
 export default class SeaOperationBackend implements IOperationBackend {
-  // Query path: pending async statement we poll to terminal. Undefined on the
-  // metadata path.
+  // Async query path: pending async statement we poll to terminal. Undefined on
+  // the metadata / sync-execute paths.
   private readonly asyncStatement?: SeaNativeAsyncStatement;
 
-  // Metadata path: terminal statement. Undefined on the query path.
-  private readonly blockingStatement?: SeaOperationStatement;
+  // Sync query path (`runAsync: false`): pending cancellable execution whose
+  // `result()` drives the blocking `execute()` to a terminal `Statement`.
+  // Undefined on the async / metadata paths.
+  private readonly cancellableExecution?: SeaNativeCancellableExecution;
+
+  // Metadata path: terminal statement. Also the resolved fetch handle on the
+  // sync-execute path once `cancellableExecution.result()` settles.
+  private blockingStatement?: SeaOperationStatement;
 
   // The cancel/close surface — whichever handle backs this operation. Both
-  // `AsyncStatement` and `Statement` expose `cancel()` / `close()`.
+  // `AsyncStatement` and `Statement` expose `cancel()` / `close()`; the
+  // sync-execute path uses a composite that routes `cancel()` to the
+  // cancellable execution (mid-compute) and `close()` to the resolved statement.
   private readonly lifecycleHandle: SeaStatementHandle;
 
   private readonly context: IClientContext;
@@ -168,15 +190,43 @@ export default class SeaOperationBackend implements IOperationBackend {
   // undefined when unset. Enforced in the async poll loop.
   private readonly queryTimeoutMs?: number;
 
-  constructor({ asyncStatement, statement, context, id, queryTimeoutSecs }: SeaOperationBackendOptions) {
-    if ((asyncStatement === undefined) === (statement === undefined)) {
-      throw new HiveDriverError('SeaOperationBackend: exactly one of `asyncStatement` or `statement` must be provided');
+  constructor({
+    asyncStatement,
+    statement,
+    cancellableExecution,
+    context,
+    id,
+    queryTimeoutSecs,
+  }: SeaOperationBackendOptions) {
+    // Exactly one of the three handle kinds must be supplied.
+    const providedCount =
+      (asyncStatement !== undefined ? 1 : 0) +
+      (statement !== undefined ? 1 : 0) +
+      (cancellableExecution !== undefined ? 1 : 0);
+    if (providedCount !== 1) {
+      throw new HiveDriverError(
+        'SeaOperationBackend: exactly one of `asyncStatement`, `statement`, or `cancellableExecution` must be provided',
+      );
     }
     this.asyncStatement = asyncStatement;
+    this.cancellableExecution = cancellableExecution;
     this.blockingStatement = statement;
-    this.lifecycleHandle = (asyncStatement ?? statement) as SeaStatementHandle;
+    // Lifecycle surface. The async/metadata handles expose both cancel/close.
+    // The sync-execute path uses a composite: `cancel()` always routes to the
+    // cancellable execution (lock-free, interrupts a running `result()`
+    // mid-compute and is a no-op once terminal); `close()` routes to the
+    // resolved terminal statement once `result()` has produced it (before that
+    // there is nothing server-side to close, and the kernel's per-execute drop
+    // guard handles an abandoned in-flight execution).
+    this.lifecycleHandle = cancellableExecution
+      ? {
+          cancel: () => cancellableExecution.cancel(),
+          close: () => (this.blockingStatement ? this.blockingStatement.close() : Promise.resolve()),
+        }
+      : ((asyncStatement ?? statement) as SeaStatementHandle);
     this.context = context;
-    this._id = id ?? asyncStatement?.statementId ?? statement?.statementId ?? uuidv4();
+    this._id =
+      id ?? asyncStatement?.statementId ?? statement?.statementId ?? cancellableExecution?.statementId ?? uuidv4();
     this.queryTimeoutMs = queryTimeoutSecs !== undefined && queryTimeoutSecs > 0 ? queryTimeoutSecs * 1000 : undefined;
   }
 
@@ -312,11 +362,20 @@ export default class SeaOperationBackend implements IOperationBackend {
       return { state: OperationState.Closed, hasResultSet: true };
     }
     if (this.asyncStatement) {
-      // Query path: report the real kernel state (single GetStatementStatus
-      // RPC — no polling here; `waitUntilReady` owns the poll loop).
+      // Async query path: report the real kernel state (single
+      // GetStatementStatus RPC — no polling here; `waitUntilReady` owns the
+      // poll loop).
       const state = statusStringToOperationState(await this.asyncStatement.status());
       return { state, hasResultSet: true };
     }
+    if (this.cancellableExecution) {
+      // Sync (`runAsync: false`) path: the kernel `execute()` blocks and polls
+      // server-side; there is no per-status RPC to query while it runs. Report
+      // Running until `result()` has materialised the terminal statement, then
+      // Succeeded — mirroring the kernel's blocking-then-terminal lifecycle.
+      const state = this.fetchHandlePromise ? OperationState.Succeeded : OperationState.Running;
+      return { state, hasResultSet: true };
+    }
     // Metadata path: the kernel statement is already terminal.
     return { state: OperationState.Succeeded, hasResultSet: true };
   }
@@ -325,6 +384,9 @@ export default class SeaOperationBackend implements IOperationBackend {
     if (this.asyncStatement) {
       return this.waitUntilReadyAsync(options);
     }
+    if (this.cancellableExecution) {
+      return this.waitUntilReadyCancellable(options);
+    }
     // Metadata path: the kernel statement has already resolved, so there is
     // nothing to poll. seaFinished fires the progress callback once with a
     // synthesised completion tick, matching the Thrift path's final tick.
@@ -420,6 +482,36 @@ export default class SeaOperationBackend implements IOperationBackend {
     }
   }
 
+  /**
+   * Sync (`runAsync: false`) execute path. Drives the blocking
+   * `CancellableExecution.result()` to a terminal `Statement` (the kernel polls
+   * to completion server-side, honouring `queryTimeoutSecs` on this path). The
+   * await is interruptible: a JS-initiated `cancel()` fires the detached
+   * canceller, the server flips the statement terminal, and the parked
+   * `result()` rejects with `Cancelled` — which we map to the typed
+   * `OperationStateError(Canceled)`.
+   *
+   * Unlike the async path there is no status poll loop (the kernel owns
+   * polling), so the progress callback fires once on completion, matching the
+   * metadata path's single completion tick.
+   */
+  private async waitUntilReadyCancellable(options?: IOperationBackendWaitOptions): Promise<void> {
+    // Already materialised → terminal-and-ready, nothing to wait for.
+    if (this.fetchHandlePromise) {
+      return;
+    }
+    // A JS-initiated cancel/close before we start short-circuits to the typed
+    // state error rather than dispatching the blocking execute.
+    failIfNotActive(this.lifecycle);
+    // `getFetchHandle()` drives `result()` and memoises the resolved Statement
+    // (also stored on `blockingStatement` so `close()` can reach it).
+    await this.getFetchHandle();
+    // Single completion tick, matching the metadata path.
+    if (options?.callback) {
+      await Promise.resolve(options.callback({ state: OperationState.Succeeded, hasResultSet: true }));
+    }
+  }
+
   /**
    * Drive `awaitResult()` on a Failed statement to surface the kernel's typed
    * SQL-error envelope. Falls back to a generic error if `awaitResult()`
@@ -444,6 +536,29 @@ export default class SeaOperationBackend implements IOperationBackend {
         this.fetchHandlePromise = this.asyncStatement.awaitResult().catch((err) => {
           throw decodeNapiKernelError(err);
         }) as Promise<SeaNativeAsyncResultHandle>;
+      } else if (this.cancellableExecution) {
+        // Sync (`runAsync: false`) path: drive the blocking `result()` to the
+        // terminal `Statement`. Store it on `blockingStatement` so `close()` can
+        // reach it post-execute, and so a subsequent fetch uses it directly.
+        this.fetchHandlePromise = this.cancellableExecution
+          .result()
+          .then((stmt) => {
+            this.blockingStatement = stmt as unknown as SeaOperationStatement;
+            return stmt as unknown as SeaFetchHandle;
+          })
+          .catch((err) => {
+            const mapped = decodeNapiKernelError(err);
+            // A cancel-induced rejection surfaces as the kernel's Cancelled
+            // error; map it to the typed `OperationStateError(Canceled)` so the
+            // `DBSQLOperation` facade mirrors its cancelled flag (it only does so
+            // for `OperationStateError`), matching the Thrift path. If the
+            // operation was cancelled client-side, prefer the typed code
+            // regardless of the kernel error text.
+            if (this.lifecycle.isCancelled) {
+              throw new OperationStateError(OperationStateErrorCode.Canceled);
+            }
+            throw mapped;
+          });
       } else {
         const stmt = this.blockingStatement!;
         if (!stmt.fetchNextBatch) {

lib/sea/SeaSessionBackend.ts

@@ -155,14 +155,56 @@ export default class SeaSessionBackend implements ISessionBackend {
       );
     }
 
-    const execOptions = this.buildExecuteOptions(options);
+    // `runAsync` selects the kernel execution path, exactly mirroring the
+    // Thrift backend's `runAsync` distinction (the only observable difference is
+    // WHEN `executeStatement` resolves — the public API, result shape, schema,
+    // and error classes are identical on both paths):
+    //
+    //   - DEFAULT (`runAsync` false/undefined) — SYNC. Route through
+    //     `executeStatementCancellable`: the kernel blocks on `execute()`
+    //     (server-side direct-results / poll-to-terminal), which is faster and,
+    //     with the napi sync canceller, fully cancellable mid-COMPUTE. The
+    //     blocking drive runs in the operation backend's `result()` (inside
+    //     `waitUntilReady`, which the facade invokes lazily at first fetch).
+    //     `queryTimeoutSecs` IS honoured on this path (forwarded to the napi
+    //     options below) since the kernel `execute()` consults it.
+    //
+    //   - `runAsync: true` — ASYNC. Submit (`wait_timeout=0s`): the server
+    //     returns a pending `AsyncStatement` immediately while the query runs;
+    //     the backend polls `status()` to terminal in `waitUntilReady()` and
+    //     materialises results via `awaitResult()`. `queryTimeoutSecs` is
+    //     ignored by the kernel on submit, so it is enforced client-side by the
+    //     operation backend's poll-loop deadline instead.
+    const runAsync = options.runAsync ?? false;
+    const queryTimeoutSecs =
+      options.queryTimeout !== undefined ? numberToInt64(options.queryTimeout).toNumber() : undefined;
+
+    if (!runAsync) {
+      // Sync path: forward `queryTimeoutSecs` to the napi options — the kernel
+      // `execute()` honours it (server statement timeout).
+      const execOptions = this.buildExecuteOptions(options, queryTimeoutSecs);
+      let cancellableExecution;
+      try {
+        cancellableExecution =
+          execOptions === undefined
+            ? await this.connection.executeStatementCancellable(statement)
+            : await this.connection.executeStatementCancellable(statement, execOptions);
+      } catch (err) {
+        throw this.logAndMapError('executeStatement', err);
+      }
+      return new SeaOperationBackend({
+        cancellableExecution: cancellableExecution!,
+        context: this.context,
+        // The kernel honours `queryTimeoutSecs` on the sync `execute` path, so
+        // it is forwarded via the napi options (see `buildExecuteOptions`); the
+        // backend also keeps it as a deadline guard for parity with async.
+        queryTimeoutSecs,
+      });
+    }
 
-    // Submit asynchronously (kernel `wait_timeout=0s`): the server returns a
-    // pending `AsyncStatement` immediately while the query runs, matching the
-    // Thrift backend's always-async (`runAsync: true`) path. The operation
-    // backend polls `status()` to terminal in `waitUntilReady()` and
-    // materialises results via `awaitResult()`, so a long-running query stays
-    // cancellable mid-flight and `status()` reports real Pending/Running states.
+    // Async path: do NOT forward `queryTimeoutSecs` (the kernel ignores it on
+    // submit — `wait_timeout=0s`); it is enforced client-side by the poll loop.
+    const execOptions = this.buildExecuteOptions(options);
     let asyncStatement;
     try {
       asyncStatement =
@@ -181,7 +223,7 @@ export default class SeaSessionBackend implements ISessionBackend {
       context: this.context,
       // `queryTimeout` is typed `number | bigint | Int64`; `numberToInt64(...).toNumber()`
       // coerces all three (a bare `Number(int64)` yields NaN — node-int64 has no valueOf).
-      queryTimeoutSecs: options.queryTimeout !== undefined ? numberToInt64(options.queryTimeout).toNumber() : undefined,
+      queryTimeoutSecs,
     });
   }
 
@@ -190,7 +232,10 @@ export default class SeaSessionBackend implements ISessionBackend {
    * `ExecuteOptions`, returning `undefined` when nothing is set so the
    * no-options call shape (`executeStatement(sql)`) is preserved.
    */
-  private buildExecuteOptions(options: ExecuteStatementOptions): SeaNativeExecuteOptions | undefined {
+  private buildExecuteOptions(
+    options: ExecuteStatementOptions,
+    queryTimeoutSecs?: number,
+  ): SeaNativeExecuteOptions | undefined {
     // Positional (`?`) and named (`:name`) parameters are mutually exclusive —
     // the kernel binds one placeholder style per statement. Use the SAME error
     // type and message as the Thrift backend (`ThriftSessionBackend`) so a
@@ -208,9 +253,14 @@ export default class SeaSessionBackend implements ISessionBackend {
     if (namedParams !== undefined) {
       execOptions.namedParams = namedParams;
     }
-    // `queryTimeout` is intentionally NOT forwarded here — the kernel ignores
-    // `queryTimeoutSecs` on `submitStatement`, so it is enforced client-side by
-    // the operation backend's poll-loop deadline instead (see executeStatement).
+    // `queryTimeoutSecs` is forwarded only on the SYNC path (the caller passes
+    // it in): the kernel `execute()` consults it as the server statement
+    // timeout. On the async submit path the caller omits it (the kernel ignores
+    // it under `wait_timeout=0s`), so it is enforced client-side by the
+    // operation backend's poll-loop deadline instead (see executeStatement).
+    if (queryTimeoutSecs !== undefined) {
+      execOptions.queryTimeoutSecs = queryTimeoutSecs;
+    }
     if (options.rowLimit !== undefined) {
       execOptions.rowLimit = Number(options.rowLimit);
     }