[SEA-NodeJS] Sync execute via directResults (executeStatementDirect)

The default sync path (`runAsync: false`) now calls the kernel's additive
`Connection.executeStatementDirect` instead of `executeStatementCancellable`.
The kernel runs ExecuteStatement with a bounded server inline wait and returns
WITHOUT polling past it; the session feature-detects the arm via `awaitResult`:
a fast query comes back as a terminal `Statement` (result inline) → wrapped with
the operation backend's `statement` arm; a slow one as an `AsyncStatement` →
the `asyncStatement` arm.

Because the returned handle always corresponds to a server-owned statement:
  - fire-and-forget CREATE/INSERT commits (server runs it inline in the POST);
  - `close()` is a clean release, never a drive-to-terminal (no close-drives);
  - a long query stays cancellable via `op.cancel()` (~150-300ms), Thrift parity;
  - errors surface at `executeStatement`, matching Thrift / Python use_kernel.

Requires the kernel's additive directResults execute
(databricks/databricks-sql-kernel#140). `execute()` on the kernel is unchanged,
so Python `use_kernel` needs no change. Regenerated napi types add
`executeStatementDirect`.

Validated e2e (pecotesting): CREATE fire-and-forget commits, 100k read, error
at execute, mid-run cancel, close() cheap (~120ms) on an abandoned long query.
Unit: SEA suite 260 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,34 @@ 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's
+      // `executeStatementDirect` runs ExecuteStatement with a bounded server
+      // inline wait and returns WITHOUT polling past it:
+      //   - a terminal `Statement` (result 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 via `op.cancel()` and `close()` is a clean
+      // release (no client-side drive-to-terminal). Fire-and-forget DDL/DML
+      // commits because the server runs it inline during the POST.
       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` (the Running arm) exposes
+      // `awaitResult`; the terminal `Statement` (Completed arm) does not.
+      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),
+  // 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> {