refactor(sea)!: move catalog/schema/sessionConf from per-statement forwarding to openSession

msrathore-db · msrathore-db · commit 43893dac0214 · 2026-05-31T00:01:43.000Z
The napi binding moved initialCatalog/initialSchema/sessionConfig from
ExecuteOptions onto ConnectionOptions (matching pyo3) because the
kernel does not read StatementSpec::statement_conf — they were
silently no-op'd in the per-statement path. Adapter follows.

- SeaAuth.ts: extend SeaNativeConnectionOptions with optional
  catalog / schema / sessionConf (intersection with each auth-mode
  variant). New SeaSessionDefaults interface for the shared shape.
- SeaBackend.ts::openSession: fold OpenSessionRequest.initialCatalog /
  initialSchema / configuration into the napi options before the
  openSession call. Drop the SeaSessionBackend.defaults forwarding.
- SeaNativeLoader.ts: drop SeaExecuteOptions; executeStatement now
  takes only the SQL.
- SeaSessionBackend.ts: drop SeaSessionDefaults and defaults field;
  drop per-statement overlay logic. useCloudFetch becomes a no-op on
  the SEA path (kernel hardcodes disposition to
  INLINE_OR_EXTERNAL_LINKS; ResultConfig exposure is M1 work).
- tests: replace per-statement-forwarding assertions with
  openSession-arg assertions. 23/23 SEA execution tests pass
  (143/143 across the SEA suite).

Signed-off-by: Madhavendra Rathore &lt;madhavendra.rathore@databricks.com&gt;
diff --git a/lib/sea/SeaAuth.ts b/lib/sea/SeaAuth.ts
@@ -49,26 +49,47 @@ const U2M_DEFAULT_REDIRECT_PORT = 8030;
  * incompatible with `isolatedModules` and a runtime-coupling hazard.
  * The Rust source of truth lives at `native/sea/src/database.rs`.
  */
-export type SeaNativeConnectionOptions =
-  | {
-      hostName: string;
-      httpPath: string;
-      authMode: 'Pat';
-      token: string;
-    }
-  | {
-      hostName: string;
-      httpPath: string;
-      authMode: 'OAuthM2m';
-      oauthClientId: string;
-      oauthClientSecret: string;
-    }
-  | {
-      hostName: string;
-      httpPath: string;
-      authMode: 'OAuthU2m';
-      oauthRedirectPort: number;
-    };
+/**
+ * Session-level defaults shared across all auth-mode variants.
+ *
+ * Mirrors `ConnectionOptions.catalog` / `.schema` / `.sessionConf` on
+ * the napi binding (kernel `Session::builder().defaults(DefaultOpts)`
+ * and `.session_conf(HashMap)` — the routes that actually populate SEA
+ * `CreateSession.catalog` / `.schema` / `.session_confs`).
+ *
+ * Per-statement overrides do not exist on the kernel surface; both
+ * pyo3 and napi expose catalog / schema / sessionConf only at session
+ * creation. Mirror that here so the adapter doesn't promise a
+ * capability the binding can't honour.
+ */
+export interface SeaSessionDefaults {
+  catalog?: string;
+  schema?: string;
+  sessionConf?: Record<string, string>;
+}
+
+export type SeaNativeConnectionOptions = SeaSessionDefaults &
+  (
+    | {
+        hostName: string;
+        httpPath: string;
+        authMode: 'Pat';
+        token: string;
+      }
+    | {
+        hostName: string;
+        httpPath: string;
+        authMode: 'OAuthM2m';
+        oauthClientId: string;
+        oauthClientSecret: string;
+      }
+    | {
+        hostName: string;
+        httpPath: string;
+        authMode: 'OAuthU2m';
+        oauthRedirectPort: number;
+      }
+  );
 
 function prependSlash(str: string): string {
   if (str.length > 0 && str.charAt(0) !== '/') {
diff --git a/lib/sea/SeaBackend.ts b/lib/sea/SeaBackend.ts
@@ -89,28 +89,36 @@ export default class SeaBackend implements IBackend {
       throw new HiveDriverError('SeaBackend: not connected. Call connect() first.');
     }
 
+    // Fold session-level defaults from the OpenSessionRequest into the
+    // napi `ConnectionOptions`. The kernel routes these through
+    // `Session::builder().defaults(DefaultOpts)` + `.session_conf(...)`
+    // so they land on the SEA `CreateSession` wire fields, not on each
+    // per-statement request. Matches pyo3's `Session.__new__` shape.
+    //
+    // Only set the optional keys when present so the napi call shape
+    // stays minimal — keeps wire snapshots / test assertions stable
+    // for callers who pass no defaults.
+    const sessionOptions: SeaNativeConnectionOptions = { ...this.nativeOptions };
+    if (request.initialCatalog !== undefined) {
+      sessionOptions.catalog = request.initialCatalog;
+    }
+    if (request.initialSchema !== undefined) {
+      sessionOptions.schema = request.initialSchema;
+    }
+    if (request.configuration !== undefined) {
+      sessionOptions.sessionConf = { ...request.configuration };
+    }
+
     let nativeConnection: SeaNativeConnection;
     try {
-      nativeConnection = (await this.binding.openSession(this.nativeOptions)) as SeaNativeConnection;
+      nativeConnection = (await this.binding.openSession(sessionOptions)) as SeaNativeConnection;
     } catch (err) {
       throw decodeNapiKernelError(err);
     }
 
-    // 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,
-      },
     });
   }
 
diff --git a/lib/sea/SeaSessionBackend.ts b/lib/sea/SeaSessionBackend.ts
@@ -31,33 +31,14 @@ import {
 import Status from '../dto/Status';
 import InfoValue from '../dto/InfoValue';
 import HiveDriverError from '../errors/HiveDriverError';
-import { SeaNativeConnection, SeaExecuteOptions } from './SeaNativeLoader';
+import { SeaNativeConnection } from './SeaNativeLoader';
 import { decodeNapiKernelError } from './SeaErrorMapping';
 import SeaOperationBackend from './SeaOperationBackend';
 
-/**
- * Per-session defaults that apply to every `executeStatement` issued
- * through this backend. Captured at `SeaBackend.openSession()` time from
- * the `OpenSessionRequest` — `initialCatalog` / `initialSchema` /
- * `sessionConfig`.
- *
- * The napi binding routes these to the kernel's `statement_conf` map,
- * which the SEA wire treats as session-scoped parameters. They are
- * forwarded with every `executeStatement` call so the JDBC-style
- * "session config" semantics are preserved even though SEA's wire
- * protocol is statement-scoped.
- */
-export interface SeaSessionDefaults {
-  initialCatalog?: string;
-  initialSchema?: string;
-  sessionConfig?: Record<string, string>;
-}
-
 export interface SeaSessionBackendOptions {
   /** The opaque napi `Connection` handle returned by `openSession`. */
   connection: SeaNativeConnection;
   context: IClientContext;
-  defaults?: SeaSessionDefaults;
   /** Optional override for `id`. Defaults to a fresh UUIDv4. */
   id?: string;
 }
@@ -72,30 +53,26 @@ export interface SeaSessionBackendOptions {
  * backend continues to handle the metadata path by default (callers
  * opt into SEA via `ConnectionOptions.useSEA`).
  *
- * **Session config flow:** the SEA wire protocol is statement-scoped,
- * so "session config" semantics (Spark conf, `initialCatalog`,
- * `initialSchema`) are emulated by forwarding the same defaults with
- * every `executeStatement` call. Per-statement overrides on
- * `ExecuteStatementOptions` are reserved for M1; M0 carries only the
- * defaults captured at session-open time plus the `useCloudFetch`
- * boolean projected onto `sessionConfig.use_cloud_fetch` for the
- * kernel.
+ * **Session config flow:** catalog / schema / sessionConf are applied
+ * once at session creation (kernel `Session::builder().defaults()` +
+ * `.session_conf()` → SEA `CreateSession.catalog` / `.schema` /
+ * `.session_confs`) and remain in effect for every statement run on
+ * the resulting napi `Connection`. No per-statement forwarding is
+ * needed — that pattern was removed when the napi binding moved these
+ * onto `openSession` to match pyo3.
  */
 export default class SeaSessionBackend implements ISessionBackend {
   private readonly connection: SeaNativeConnection;
 
   private readonly context: IClientContext;
 
-  private readonly defaults: SeaSessionDefaults;
-
   private readonly _id: string;
 
   private closed = false;
 
-  constructor({ connection, context, defaults, id }: SeaSessionBackendOptions) {
+  constructor({ connection, context, id }: SeaSessionBackendOptions) {
     this.connection = connection;
     this.context = context;
-    this.defaults = defaults ?? {};
     this._id = id ?? uuidv4();
   }
 
@@ -108,13 +85,21 @@ export default class SeaSessionBackend implements ISessionBackend {
   }
 
   /**
-   * Execute a SQL statement through the napi binding. Merges the
-   * session-level defaults (`initialCatalog` / `initialSchema` /
-   * `sessionConfig`) with the per-call `useCloudFetch` override.
+   * Execute a SQL statement through the napi binding.
+   *
+   * Catalog / schema / sessionConf were applied at session open, so
+   * there are no per-statement options to thread through.
    *
    * M0 intentionally rejects `queryTimeout`, `namedParameters`, and
-   * `ordinalParameters` with explicit deferred-to-M1 errors. The Thrift
-   * backend remains the path for consumers that need any of those today.
+   * `ordinalParameters` with explicit deferred-to-M1 errors. `useCloudFetch`
+   * is a no-op on the SEA path — the kernel hardcodes the SEA
+   * `disposition` to `INLINE_OR_EXTERNAL_LINKS`, and per-statement
+   * conf overrides have no reader on the kernel; cloud-fetch behaviour
+   * is governed entirely by the kernel's `ResultConfig` (M1 binding
+   * surface).
+   *
+   * The Thrift backend remains the path for consumers that need any
+   * of those today.
    */
   public async executeStatement(statement: string, options: ExecuteStatementOptions): Promise<IOperationBackend> {
     this.failIfClosed();
@@ -131,24 +116,9 @@ export default class SeaSessionBackend implements ISessionBackend {
       );
     }
 
-    // Merge session-level sessionConfig with per-statement useCloudFetch.
-    // The kernel accepts only string-valued conf values; booleans are
-    // String()'d to "true"/"false" matching the existing Thrift conf
-    // convention.
-    const sessionConfig: Record<string, string> = { ...(this.defaults.sessionConfig ?? {}) };
-    if (options.useCloudFetch !== undefined) {
-      sessionConfig.use_cloud_fetch = String(options.useCloudFetch);
-    }
-
-    const executeOptions: SeaExecuteOptions = {
-      initialCatalog: this.defaults.initialCatalog,
-      initialSchema: this.defaults.initialSchema,
-      sessionConfig: Object.keys(sessionConfig).length > 0 ? sessionConfig : undefined,
-    };
-
     let nativeStatement;
     try {
-      nativeStatement = await this.connection.executeStatement(statement, executeOptions);
+      nativeStatement = await this.connection.executeStatement(statement);
     } catch (err) {
       throw decodeNapiKernelError(err);
     }
diff --git a/tests/unit/sea/execution.test.ts b/tests/unit/sea/execution.test.ts
@@ -21,7 +21,6 @@ import {
   SeaNativeBinding,
   SeaNativeConnection,
   SeaNativeStatement,
-  SeaExecuteOptions,
 } from '../../../lib/sea/SeaNativeLoader';
 import IClientContext, { ClientConfig } from '../../../lib/contracts/IClientContext';
 import IDBSQLLogger, { LogLevel } from '../../../lib/contracts/IDBSQLLogger';
@@ -61,18 +60,15 @@ class FakeNativeConnection implements SeaNativeConnection {
 
   public lastSql?: string;
 
-  public lastOptions?: SeaExecuteOptions;
-
   public throwOnExecute: Error | null = null;
 
   public statementToReturn: FakeNativeStatement = new FakeNativeStatement();
 
-  public async executeStatement(sql: string, options: SeaExecuteOptions): Promise<SeaNativeStatement> {
+  public async executeStatement(sql: string): Promise<SeaNativeStatement> {
     if (this.throwOnExecute) {
       throw this.throwOnExecute;
     }
     this.lastSql = sql;
-    this.lastOptions = options;
     return this.statementToReturn;
   }
 
@@ -240,7 +236,7 @@ describe('SeaBackend', () => {
     expect(sessionBackend.id).to.be.a('string').and.have.length.greaterThan(0);
   });
 
-  it('openSession() propagates initialCatalog / initialSchema / sessionConfig through to executeStatement', async () => {
+  it('openSession() forwards initialCatalog / initialSchema / configuration to the napi openSession call (not per-statement)', async () => {
     const connection = new FakeNativeConnection();
     const binding = makeBinding(connection);
     const backend = new SeaBackend({ context: makeContext(), nativeBinding: binding });
@@ -257,14 +253,22 @@ describe('SeaBackend', () => {
       configuration: { 'spark.sql.execution.arrow.enabled': 'true' },
     });
 
-    await session.executeStatement('SELECT 1', {});
+    // The defaults reach the kernel via `Session::builder().defaults()` +
+    // `.session_conf()`, applied on `CreateSession`. Assert they were
+    // folded into the napi `openSession` arg.
+    expect(binding.openSessionStub.calledOnce).to.equal(true);
+    expect(binding.openSessionStub.firstCall.args[0]).to.deep.include({
+      authMode: 'Pat',
+      token: 't',
+      catalog: 'main',
+      schema: 'default',
+      sessionConf: { 'spark.sql.execution.arrow.enabled': 'true' },
+    });
 
+    // And the SQL still threads through executeStatement (now with no
+    // per-statement options).
+    await session.executeStatement('SELECT 1', {});
     expect(connection.lastSql).to.equal('SELECT 1');
-    expect(connection.lastOptions).to.deep.equal({
-      initialCatalog: 'main',
-      initialSchema: 'default',
-      sessionConfig: { 'spark.sql.execution.arrow.enabled': 'true' },
-    });
   });
 
   it('close() clears connection state without throwing', async () => {
@@ -285,8 +289,8 @@ describe('SeaBackend', () => {
 });
 
 describe('SeaSessionBackend', () => {
-  function makeSession(connection: SeaNativeConnection, defaults = {}) {
-    return new SeaSessionBackend({ connection, context: makeContext(), defaults });
+  function makeSession(connection: SeaNativeConnection) {
+    return new SeaSessionBackend({ connection, context: makeContext() });
   }
 
   it('executeStatement passes sql through verbatim', async () => {
@@ -304,21 +308,6 @@ describe('SeaSessionBackend', () => {
     expect(op.id).to.be.a('string').and.have.length.greaterThan(0);
   });
 
-  it('executeStatement merges session defaults into ExecuteOptions', async () => {
-    const connection = new FakeNativeConnection();
-    const session = makeSession(connection, {
-      initialCatalog: 'main',
-      initialSchema: 'default',
-      sessionConfig: { foo: 'bar' },
-    });
-    await session.executeStatement('SELECT 1', {});
-    expect(connection.lastOptions).to.deep.equal({
-      initialCatalog: 'main',
-      initialSchema: 'default',
-      sessionConfig: { foo: 'bar' },
-    });
-  });
-
   it('executeStatement rejects namedParameters (M1)', async () => {
     const connection = new FakeNativeConnection();
     const session = makeSession(connection);
