[SEA-NodeJS] Sync execute via directResults: fix CREATE, drop close-drives, keep mid-run cancel

The default sync path (`runAsync: false`) now calls the kernel's directResults
execute (`Connection.executeStatementDirect`) instead of the cancellable-execute
path. The kernel runs ExecuteStatement with a bounded server inline wait and
returns WITHOUT polling past it: a fast query comes back as a terminal
`Statement` (result inline), a slow one as an `AsyncStatement` (a poll/cancel
handle). The session feature-detects the arm via `awaitResult` and wraps it with
the operation backend's existing `statement` / `asyncStatement` arms.

Because the returned handle always corresponds to a server-owned statement:

  - fire-and-forget `CREATE`/`INSERT` (execute then close, no fetch) COMMITS —
    the server runs it inline during the POST (fixes the prior lazy-execute
    "CREATE didn't run" bug);
  - `close()` is a clean release (DELETE), never a drive-to-terminal — the
    eager-handle + close-drives workaround is gone entirely;
  - a long query is cancellable via `op.cancel()` once the handle is held
    (~150-300ms), at parity with the Thrift backend;
  - errors surface at `executeStatement` (not silently at close), matching
    Thrift and Python use_kernel.

Requires the kernel directResults execute (databricks/databricks-sql-kernel#136).
The regenerated napi router types add `executeStatementDirect`; the fallback
package names stay on the driver's canonical `@databricks/sql-kernel-*`.

Validated e2e (pecotesting): CREATE fire-and-forget commits, dependent ordering,
100k read, error-at-execute, mid-run cancel, and close() cheap (~120ms) on an
abandoned long query. Unit: SEA suite 247 passing; eslint clean.

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

Author: msrathore-db

lib/sea/SeaSessionBackend.ts

@@ -33,7 +33,7 @@ import InfoValue from '../dto/InfoValue';
 import HiveDriverError from '../errors/HiveDriverError';
 import ParameterError from '../errors/ParameterError';
 import { LogLevel } from '../contracts/IDBSQLLogger';
-import { SeaConnection, SeaNativeExecuteOptions, SeaStatement } from './SeaNativeLoader';
+import { SeaConnection, SeaNativeExecuteOptions, SeaStatement, SeaNativeAsyncStatement } from './SeaNativeLoader';
 import { decodeNapiKernelError } from './SeaErrorMapping';
 import { numberToInt64 } from '../thrift-backend/ThriftSessionBackend';
 import SeaOperationBackend from './SeaOperationBackend';
@@ -183,24 +183,37 @@ export default class SeaSessionBackend implements ISessionBackend {
       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).
+      // DEFAULT — directResults (the Thrift/JDBC model). The kernel sends
+      // ExecuteStatement with a bounded server-side inline wait (10s +
+      // on_wait_timeout=CONTINUE) and returns WITHOUT polling past it:
+      //   - a terminal `Statement` (results inline) for a fast query, or
+      //   - an `AsyncStatement` (a poll/cancel handle) for a slow one.
+      // Either way the handle is tied to a server-owned statement, so a long
+      // query stays cancellable and `close()` is a clean release — no
+      // eager-handle / close-drives workaround. Fire-and-forget DDL/DML commits
+      // because the server runs it inline during the POST (fast path).
       const execOptions = this.buildExecuteOptions(options, queryTimeoutSecs);
-      let cancellableExecution;
+      let direct;
       try {
-        cancellableExecution =
+        direct =
           execOptions === undefined
-            ? await this.connection.executeStatementCancellable(statement)
-            : await this.connection.executeStatementCancellable(statement, execOptions);
+            ? await this.connection.executeStatementDirect(statement)
+            : await this.connection.executeStatementDirect(statement, execOptions);
       } catch (err) {
         throw this.logAndMapError('executeStatement', err);
       }
+      // Feature-detect the arm: only `AsyncStatement` exposes `awaitResult`.
+      const asAsync = direct as unknown as SeaNativeAsyncStatement;
+      if (direct !== null && typeof asAsync.awaitResult === 'function') {
+        return new SeaOperationBackend({
+          asyncStatement: asAsync,
+          context: this.context,
+          queryTimeoutSecs,
+        });
+      }
       return new SeaOperationBackend({
-        cancellableExecution: cancellableExecution!,
+        statement: direct as unknown as SeaStatement,
         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,
       });
     }

native/sea/index.d.ts

@@ -234,6 +234,26 @@ class FakeNativeConnection implements SeaConnection {
     return this.lastCancellableExecution;
   }
 
+  // directResults (`runAsync: false`, the DEFAULT) query path: records sql +
+  // options and returns either a terminal `Statement` (Completed arm) or — when
+  // `directReturnsRunning` is set — a pending `AsyncStatement` (Running arm).
+  // These are the two arms `SeaSessionBackend.executeStatement` feature-detects
+  // via `awaitResult`.
+  public directReturnsRunning = false;
+
+  public async executeStatementDirect(sql: string, options?: unknown): Promise<any> {
+    if (this.throwOnExecute) {
+      throw this.throwOnExecute;
+    }
+    this.lastSql = sql;
+    this.lastOptions = options;
+    if (this.directReturnsRunning) {
+      this.lastAsyncStatement = new FakeAsyncStatement(this.submitStatusValue);
+      return this.lastAsyncStatement;
+    }
+    return this.statementToReturn;
+  }
+
   // Async-submit path: records sql + per-statement options (for forwarding
   // assertions) and returns a pending AsyncStatement.
   public async submitStatement(sql: string, options?: unknown): Promise<any> {