sea-execution: executeStatement + openSession via SeaSessionBackend

SeaSessionBackend wraps the napi Connection handle. executeStatement
passes through to napi.executeStatement and returns an
IOperationBackend (SeaOperationBackend in sea-results feature).
Session config + initialCatalog/initialSchema flow to napi
openSession. M0 stops at executeStatement; metadata methods +
per-stmt overrides defer to M1.

No new dependencies. Reuses existing ConnectionOptions / Session
config shapes.

Co-authored-by: Isaac

Author: msrathore-db

lib/DBSQLClient.ts

@@ -243,7 +243,7 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
     // pattern (see databricks-sql-python/src/databricks/sql/session.py).
     const internalOptions = options as ConnectionOptions & InternalConnectionOptions;
     this.backend = internalOptions.useSEA
-      ? new SeaBackend()
+      ? new SeaBackend({ context: this })
       : new ThriftBackend({
           context: this,
           onConnectionEvent: (event, payload) => this.forwardConnectionEvent(event, payload),

lib/sea/SeaBackend.ts

@@ -14,156 +14,123 @@
 
 import IBackend from '../contracts/IBackend';
 import ISessionBackend from '../contracts/ISessionBackend';
-import IOperationBackend from '../contracts/IOperationBackend';
+import IClientContext from '../contracts/IClientContext';
 import { ConnectionOptions, OpenSessionRequest } from '../contracts/IDBSQLClient';
-import {
-  ExecuteStatementOptions,
-  TypeInfoRequest,
-  CatalogsRequest,
-  SchemasRequest,
-  TablesRequest,
-  TableTypesRequest,
-  ColumnsRequest,
-  FunctionsRequest,
-  PrimaryKeysRequest,
-  CrossReferenceRequest,
-} from '../contracts/IDBSQLSession';
-import Status from '../dto/Status';
-import InfoValue from '../dto/InfoValue';
 import HiveDriverError from '../errors/HiveDriverError';
-import { getSeaNative, SeaNativeBinding } from './SeaNativeLoader';
+import {
+  getSeaNative,
+  SeaNativeBinding,
+  SeaNativeConnection,
+} from './SeaNativeLoader';
+import { mapKernelErrorToJsError, KernelErrorShape } from './SeaErrorMapping';
 import { buildSeaConnectionOptions, SeaNativeConnectionOptions } from './SeaAuth';
-
-const NOT_IMPLEMENTED_SESSION =
-  'SEA session backend: method not implemented in sea-auth (M0); lands in sea-execution/sea-operation.';
-
-/**
- * Opaque handle to the napi binding's `Connection` class. The exact
- * shape lives in `native/sea/index.d.ts` (auto-generated). We type it as
- * a structural minimum here so the loader's pass-through typing doesn't
- * leak into every call site.
- */
-interface NativeConnection {
-  close(): Promise<void>;
-}
+import SeaSessionBackend from './SeaSessionBackend';
 
 /**
- * Minimal `ISessionBackend` that wraps the napi-binding's `Connection`.
- *
- * For M0 (sea-auth) only `id` and `close()` are functional — they're the
- * subset required to round-trip a connect-open-close cycle. Every other
- * method throws a clear "not implemented in M0" `HiveDriverError`.
- *
- * The `id` field is currently a synthetic counter-based string; the kernel
- * exposes a real session-id through a follow-on getter that
- * `sea-execution` will wire through.
+ * Sentinel string the napi binding uses on `Error.reason` JSON envelopes.
+ * Keep in sync with `native/sea/src/error.rs` (`SENTINEL`).
  */
-export class SeaSessionBackend implements ISessionBackend {
-  private static seq = 0;
-
-  public readonly id: string;
-
-  private readonly connection: NativeConnection;
-
-  constructor(connection: NativeConnection) {
-    this.connection = connection;
-    SeaSessionBackend.seq += 1;
-    this.id = `sea-session-${SeaSessionBackend.seq}`;
-  }
-
-  /* eslint-disable @typescript-eslint/no-unused-vars */
-  public async getInfo(_infoType: number): Promise<InfoValue> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async executeStatement(
-    _statement: string,
-    _options: ExecuteStatementOptions,
-  ): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getTypeInfo(_request: TypeInfoRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getCatalogs(_request: CatalogsRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getSchemas(_request: SchemasRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getTables(_request: TablesRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getTableTypes(_request: TableTypesRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getColumns(_request: ColumnsRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getFunctions(_request: FunctionsRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getPrimaryKeys(_request: PrimaryKeysRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getCrossReference(_request: CrossReferenceRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
+const KERNEL_ERROR_SENTINEL = '__databricks_error__:';
+
+function rethrowKernelError(err: unknown): never {
+  if (err && typeof err === 'object' && 'message' in err) {
+    const reason = (err as { reason?: unknown }).reason;
+    if (typeof reason === 'string' && reason.startsWith(KERNEL_ERROR_SENTINEL)) {
+      try {
+        const payload = JSON.parse(reason.slice(KERNEL_ERROR_SENTINEL.length)) as KernelErrorShape;
+        throw mapKernelErrorToJsError(payload);
+      } catch (parseErr) {
+        if (parseErr !== err) {
+          throw parseErr;
+        }
+      }
+    }
   }
-  /* eslint-enable @typescript-eslint/no-unused-vars */
+  throw err;
+}
 
-  public async close(): Promise<Status> {
-    await this.connection.close();
-    return Status.success();
-  }
+export interface SeaBackendOptions {
+  context: IClientContext;
+  /**
+   * Optional injection seam for unit tests. When provided, replaces the
+   * default `getSeaNative()` call so tests can swap in a mock napi
+   * binding without loading the `.node` artifact.
+   */
+  nativeBinding?: SeaNativeBinding;
 }
 
 /**
- * M0 SeaBackend — wires PAT auth + napi `openSession` end-to-end.
+ * SEA-backed implementation of `IBackend`.
  *
- * Connect is a no-op at this layer (the napi binding has no notion of a
- * standalone "connect"; a session is opened directly). We capture the
- * validated PAT options and hand them to `openSession()` on demand.
+ * **M0 dispatch model:** the napi binding's `openSession()` already
+ * builds a kernel `Session` from PAT + hostname + httpPath, so there is
+ * no "connect" round-trip before `openSession` — `connect()` only
+ * captures the `ConnectionOptions` and validates that PAT auth is in
+ * use. The actual session open happens inside `openSession()`.
  *
- * Subsequent milestones (`sea-execution`, `sea-operation`) replace the
- * stubbed `ISessionBackend` / `IOperationBackend` methods with real
- * napi-binding calls.
+ * **Auth validation:** delegates to `buildSeaConnectionOptions` from
+ * `SeaAuth`, which mirrors the existing DBSQLClient PAT validation
+ * pattern (slash-prepended httpPath, AuthenticationError on missing
+ * token, HiveDriverError on non-PAT authType naming M1 modes).
+ *
+ * **Why we don't use IClientContext's connectionProvider here:** that
+ * provider is the Thrift HTTP transport. The kernel owns its own
+ * reqwest+rustls stack inside the native binding, so there is no
+ * NodeJS-level connection state to manage on the SEA path. The
+ * `IClientContext` is still useful for logger + config access.
  */
 export default class SeaBackend implements IBackend {
-  private nativeOptions?: SeaNativeConnectionOptions;
+  private readonly context: IClientContext;
 
-  private readonly native: SeaNativeBinding;
+  private readonly binding: SeaNativeBinding;
 
-  constructor(native: SeaNativeBinding = getSeaNative()) {
-    this.native = native;
+  private nativeOptions?: SeaNativeConnectionOptions;
+
+  constructor(options?: SeaBackendOptions) {
+    this.context = options?.context as IClientContext;
+    this.binding = options?.nativeBinding ?? getSeaNative();
   }
 
   public async connect(options: ConnectionOptions): Promise<void> {
     // Validate PAT auth + capture the napi-binding option shape.
-    // Any non-PAT mode (or a missing token) throws here, before we ever
-    // touch the native binding.
+    // Any non-PAT mode (or a missing/empty token) throws here, before
+    // we ever touch the native binding.
     this.nativeOptions = buildSeaConnectionOptions(options);
   }
 
-  // eslint-disable-next-line @typescript-eslint/no-unused-vars
-  public async openSession(_request: OpenSessionRequest): Promise<ISessionBackend> {
+  public async openSession(request: OpenSessionRequest): Promise<ISessionBackend> {
     if (!this.nativeOptions) {
-      throw new HiveDriverError('SeaBackend: connect() must be called before openSession().');
+      throw new HiveDriverError('SeaBackend: not connected. Call connect() first.');
+    }
+
+    let nativeConnection: SeaNativeConnection;
+    try {
+      nativeConnection = (await this.binding.openSession(this.nativeOptions)) as SeaNativeConnection;
+    } catch (err) {
+      rethrowKernelError(err);
     }
-    const connection = (await this.native.openSession(this.nativeOptions)) as NativeConnection;
-    return new SeaSessionBackend(connection);
+
+    // Merge `request.configuration` (the existing public field for Spark
+    // conf) with any backend-specific session config. The SEA wire
+    // protocol applies these per-statement, but we capture them at
+    // session-open time and forward with every executeStatement to
+    // preserve session-config semantics.
+    const sessionConfig = request.configuration ? { ...request.configuration } : undefined;
+
+    return new SeaSessionBackend({
+      connection: nativeConnection!,
+      context: this.context,
+      defaults: {
+        initialCatalog: request.initialCatalog,
+        initialSchema: request.initialSchema,
+        sessionConfig,
+      },
+    });
   }
 
   public async close(): Promise<void> {
-    // Connection-level resources are owned by the session wrapper. No-op here.
+    // No backend-level resources to release — each `SeaSessionBackend`
+    // owns its own napi `Connection` lifecycle.
     this.nativeOptions = undefined;
   }
 }

lib/sea/SeaNativeLoader.ts

@@ -35,27 +35,64 @@
 // eslint-disable-next-line @typescript-eslint/no-var-requires, import/no-dynamic-require, global-require
 const native = require('../../native/sea/index.js');
 
+/**
+ * JS-visible per-execute options carried over the napi binding boundary.
+ * Mirrors the `ExecuteOptions` shape generated by napi-rs into
+ * `native/sea/index.d.ts`. Re-declared here so the JS adapter layer
+ * isn't tied to the binding-generated types.
+ */
+export interface SeaExecuteOptions {
+  initialCatalog?: string;
+  initialSchema?: string;
+  sessionConfig?: Record<string, string>;
+}
+
+/**
+ * Arrow IPC payload returned by `Statement.fetchNextBatch()`. Carries a
+ * complete Arrow IPC stream (schema header + 1 record-batch message).
+ */
+export interface SeaArrowBatch {
+  ipcBytes: Buffer;
+}
+
+/**
+ * Arrow IPC payload returned by `Statement.schema()` (schema header only).
+ */
+export interface SeaArrowSchema {
+  ipcBytes: Buffer;
+}
+
+/**
+ * Typed surface for the opaque napi `Statement` handle. Method signatures
+ * match `native/sea/index.d.ts` exactly so the JS-side wrappers can
+ * `await` them without `any` casts.
+ */
+export interface SeaNativeStatement {
+  fetchNextBatch(): Promise<SeaArrowBatch | null>;
+  schema(): Promise<SeaArrowSchema>;
+  cancel(): Promise<void>;
+  close(): Promise<void>;
+}
+
+/**
+ * Typed surface for the opaque napi `Connection` handle.
+ */
+export interface SeaNativeConnection {
+  executeStatement(sql: string, options: SeaExecuteOptions): Promise<SeaNativeStatement>;
+  close(): Promise<void>;
+}
+
 /**
  * Public surface of the native binding exposed to the rest of the
  * NodeJS driver. Round 2 lands `openSession` + opaque `Connection` /
  * `Statement` classes (the binding-generated `.d.ts` is the source of
  * truth for their method signatures — see `native/sea/index.d.ts`).
- *
- * We deliberately keep this typed loosely (`unknown` for the class
- * shapes) so the loader layer doesn't have to import the binding's
- * generated types and the JS adapter layer can introduce its own
- * higher-level wrappers without conflicting with the binding's TS
- * declarations.
  */
 export interface SeaNativeBinding {
   /** Returns the native crate version (smoke test for the binding's load path). */
   version(): string;
   /** Open a session over PAT auth. Returns an opaque Connection. */
-  openSession(opts: {
-    hostName: string;
-    httpPath: string;
-    token: string;
-  }): Promise<unknown>;
+  openSession(opts: { hostName: string; httpPath: string; token: string }): Promise<SeaNativeConnection>;
   /** Opaque Connection class — instance methods on the binding-generated d.ts. */
   Connection: Function;
   /** Opaque Statement class — instance methods on the binding-generated d.ts. */