feat(sea): forward queryTags through executeStatement to the napi binding

Threads the public `ExecuteStatementOptions.queryTags` through the
SEA backend into the kernel's per-statement Spark conf overlay.

JS-side adapter path:
- Serialises the `Record<string, string | null | undefined>` tag map
  via the existing `serializeQueryTags` util — same wire shape the
  Thrift backend produces (`k1:v1,k2:v2`, null values as bare keys).
- Writes the serialised string into `statementConf["query_tags"]`
  on the napi `ExecuteOptions` (defined on the kernel side in
  `napi/src/connection.rs`).
- Pre-serialising at the JS layer (instead of via the napi binding's
  own `queryTags` field) preserves null-valued tags, which a
  `HashMap<String, String>` over the FFI boundary cannot carry.

Why match Thrift's exact wire shape: SEA and Thrift hit the same
server with the same `confOverlay.query_tags` key. A consumer
porting from Thrift to SEA gets byte-equivalent tagging behaviour
without any code change on their side.

Plumbing changes:
- `SeaNativeLoader.ts` — `SeaNativeConnection.executeStatement` now
  accepts an optional `SeaNativeExecuteOptions` second arg with
  `statementConf` + `queryTags` fields, mirroring the napi-generated
  `ExecuteOptions` interface.
- `SeaSessionBackend.ts` — adapter call site builds `nativeOptions`
  from the public `queryTags` and forwards it on the napi call.
  Existing `namedParameters`/`ordinalParameters`/`queryTimeout`
  deferred errors stay in place.
- `native/sea/index.d.ts` — adds `ExecuteOptions` and updates
  `Connection.executeStatement` signature to take the optional
  second arg. Mirrors the napi-rs codegen from kernel branch
  `msrathore/krn-statement-options`.

Tests:
- `tests/unit/sea/execution-query-tags.test.ts` — 8 cases covering
  omission/empty, single + multiple tags, null/undefined-valued
  tags as bare keys, special-char escaping in keys and values.

Deferred (separate PRs):
- Positional / named parameters (TypedValue mapping over napi)
- Row limit, wait timeout (need kernel `StatementSpec` extension)
- Async execute (`submit()` + `await_result()` — needs new
  ExecutedAsyncStatement napi wrapper)

Cross-PR dependency: requires kernel branch
`msrathore/krn-statement-options` (commit `79bba34`), which is
stacked on `msrathore/krn-max-connections`. This branch stacks
on `msrathore/sea-max-connections`.

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

Author: msrathore-db

lib/sea/SeaNativeLoader.ts

@@ -64,16 +64,43 @@ export interface SeaNativeStatement {
   close(): Promise<void>;
 }
 
+/**
+ * Per-statement options for the napi `Connection.executeStatement`.
+ * Mirrors the napi-rs-generated `ExecuteOptions` in
+ * `native/sea/index.d.ts`. Declared locally to avoid coupling the
+ * JS-side adapter to the auto-generated file.
+ *
+ * - `statementConf` — per-statement Spark conf overlay merged on top
+ *   of the session-level `sessionConf` at execute time. The map wins
+ *   on key collisions.
+ * - `queryTags` — the napi binding accepts a `Record<string, string>`
+ *   and serialises it into `statementConf["query_tags"]` matching
+ *   NodeJS Thrift's `serializeQueryTags` wire shape. The JS-side
+ *   adapter today pre-serialises via the existing
+ *   `serializeQueryTags` helper and writes the result into
+ *   `statementConf` directly (so null-valued tags carry through),
+ *   so this field is not used by the SEA backend's adapter call
+ *   site; it is declared because the napi binding exports it and
+ *   alternate consumers may use it.
+ */
+export interface SeaNativeExecuteOptions {
+  statementConf?: Record<string, string>;
+  queryTags?: Record<string, string>;
+}
+
 /**
  * Typed surface for the opaque napi `Connection` handle.
  */
 export interface SeaNativeConnection {
   /**
    * Execute a SQL statement. Catalog / schema / sessionConf are
    * session-level — set on `openSession`, applied to every statement
-   * executed on the resulting `Connection`. No per-statement options.
+   * executed on the resulting `Connection`.
+   *
+   * `options` is optional; today carries `statementConf`
+   * (per-statement Spark conf overlay) and `queryTags`.
    */
-  executeStatement(sql: string): Promise<SeaNativeStatement>;
+  executeStatement(sql: string, options?: SeaNativeExecuteOptions): Promise<SeaNativeStatement>;
   close(): Promise<void>;
 }
 

lib/sea/SeaSessionBackend.ts

@@ -34,6 +34,7 @@ import HiveDriverError from '../errors/HiveDriverError';
 import { SeaNativeConnection } from './SeaNativeLoader';
 import { decodeNapiKernelError } from './SeaErrorMapping';
 import SeaOperationBackend from './SeaOperationBackend';
+import { serializeQueryTags } from '../utils';
 
 export interface SeaSessionBackendOptions {
   /** The opaque napi `Connection` handle returned by `openSession`. */
@@ -116,9 +117,22 @@ export default class SeaSessionBackend implements ISessionBackend {
       );
     }
 
+    // Build the per-statement conf overlay. Today only `queryTags` is
+    // surfaced on the public `ExecuteStatementOptions` (mirrors Thrift);
+    // pre-serialise on the JS side via the existing
+    // `serializeQueryTags` helper so the kernel-side conf overlay
+    // shape exactly matches the Thrift wire bytes for the same input.
+    // Doing the serialisation here (instead of inside the napi `queryTags`
+    // field) is what carries null-valued tags through correctly — napi's
+    // `HashMap<String, String>` can't represent nulls.
+    const serializedQueryTags = serializeQueryTags(options.queryTags);
+    const statementConf =
+      serializedQueryTags !== undefined ? { query_tags: serializedQueryTags } : undefined;
+    const nativeOptions = statementConf !== undefined ? { statementConf } : undefined;
+
     let nativeStatement;
     try {
-      nativeStatement = await this.connection.executeStatement(statement);
+      nativeStatement = await this.connection.executeStatement(statement, nativeOptions);
     } catch (err) {
       throw decodeNapiKernelError(err);
     }

native/sea/index.d.ts

@@ -71,6 +71,22 @@ export interface ConnectionOptions {
  * to camelCase for free functions).
  */
 export declare function openSession(options: ConnectionOptions): Promise<Connection>
+/**
+ * Per-statement options for `Connection.executeStatement`.
+ * Mirrors the napi-rs-generated `ExecuteOptions` in
+ * `napi/src/connection.rs`. Today carries:
+ * - `statementConf` — per-statement Spark conf overlay merged on top
+ *   of the session-level `sessionConf` at execute time. Map wins on
+ *   key collisions.
+ * - `queryTags` — JSON-encoded into `statementConf["query_tags"]`
+ *   matching NodeJS Thrift's `serializeQueryTags` shape. Passing both
+ *   `queryTags` and an explicit `statementConf["query_tags"]` raises
+ *   `InvalidArgument`.
+ */
+export interface ExecuteOptions {
+  statementConf?: Record<string, string>
+  queryTags?: Record<string, string>
+}
 /**
  * A single Arrow IPC stream payload encoding one record batch (plus
  * the schema header so the JS-side reader is stateless).
@@ -108,11 +124,13 @@ export declare class Connection {
    * Execute a SQL statement and return a Statement handle that
    * streams batches via `fetchNextBatch()`.
    *
-   * No per-statement options: catalog / schema / sessionConf are
-   * session-level (`openSession`). Positional / named parameters
-   * land in M1 via `Statement::spec().param(…)` on the kernel.
+   * Catalog / schema / sessionConf are session-level (`openSession`).
+   * `options` carries per-statement knobs: `statementConf`
+   * (per-statement Spark conf overlay) and `queryTags` (serialised
+   * into `statementConf["query_tags"]` matching NodeJS Thrift's
+   * `serializeQueryTags` wire shape).
    */
-  executeStatement(sql: string): Promise<Statement>
+  executeStatement(sql: string, options?: ExecuteOptions | undefined | null): Promise<Statement>
   /**
    * Explicit close. Marks the connection wrapper as closed so
    * subsequent calls on this `Connection` return `InvalidArg`, then

tests/unit/sea/execution-query-tags.test.ts

@@ -0,0 +1,163 @@
+// Copyright (c) 2026 Databricks, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/**
+ * Unit tests for `SeaSessionBackend.executeStatement` query-tags
+ * threading. The JS-side adapter pre-serialises the public
+ * `queryTags: Record<string, string | null | undefined>` map via
+ * the existing `serializeQueryTags` util (so null-valued tags carry
+ * through) and writes the result into the napi
+ * `statementConf["query_tags"]`. The kernel then forwards the conf
+ * overlay verbatim onto the SEA wire.
+ *
+ * These tests verify that the JS adapter constructs the napi options
+ * shape correctly. End-to-end behaviour against a live warehouse is
+ * exercised separately in `tests/e2e/sea/`.
+ */
+
+import { expect } from 'chai';
+import sinon from 'sinon';
+import SeaSessionBackend from '../../../lib/sea/SeaSessionBackend';
+import {
+  SeaNativeConnection,
+  SeaNativeStatement,
+  SeaNativeExecuteOptions,
+} from '../../../lib/sea/SeaNativeLoader';
+import IClientContext, { ClientConfig } from '../../../lib/contracts/IClientContext';
+import IDBSQLLogger, { LogLevel } from '../../../lib/contracts/IDBSQLLogger';
+
+class FakeNativeStatement implements SeaNativeStatement {
+  public async fetchNextBatch() {
+    return null;
+  }
+
+  public async schema() {
+    return { ipcBytes: Buffer.alloc(0) };
+  }
+
+  public async cancel() {
+    // no-op
+  }
+
+  public async close() {
+    // no-op
+  }
+}
+
+function makeFakeContext(): IClientContext {
+  const logger: IDBSQLLogger = {
+    log(_level: LogLevel, _message: string): void {
+      // no-op
+    },
+  };
+  const config = {} as ClientConfig;
+  return {
+    getConfig: () => config,
+    getLogger: () => logger,
+    getConnectionProvider: () => {
+      throw new Error('not used');
+    },
+    getClient: () => {
+      throw new Error('not used');
+    },
+    getDriver: () => {
+      throw new Error('not used');
+    },
+  } as unknown as IClientContext;
+}
+
+describe('SeaSessionBackend — query tags threading', () => {
+  let executeSpy: sinon.SinonSpy;
+  let connection: SeaNativeConnection;
+  let session: SeaSessionBackend;
+
+  beforeEach(() => {
+    const stmt = new FakeNativeStatement();
+    executeSpy = sinon.spy(async (_sql: string, _options?: SeaNativeExecuteOptions) => stmt);
+    connection = {
+      executeStatement: executeSpy,
+      close: async () => {},
+    } as unknown as SeaNativeConnection;
+    session = new SeaSessionBackend({ connection, context: makeFakeContext() });
+  });
+
+  it('omits the napi options arg when queryTags is not set', async () => {
+    await session.executeStatement('SELECT 1', {});
+    expect(executeSpy.calledOnce).to.equal(true);
+    expect(executeSpy.firstCall.args[0]).to.equal('SELECT 1');
+    expect(executeSpy.firstCall.args[1]).to.equal(undefined);
+  });
+
+  it('omits the napi options arg when queryTags is empty', async () => {
+    await session.executeStatement('SELECT 1', { queryTags: {} });
+    expect(executeSpy.firstCall.args[1]).to.equal(undefined);
+  });
+
+  it('forwards a single tag through statementConf["query_tags"]', async () => {
+    await session.executeStatement('SELECT 1', {
+      queryTags: { team: 'platform' },
+    });
+    const opts = executeSpy.firstCall.args[1] as SeaNativeExecuteOptions;
+    expect(opts.statementConf).to.deep.equal({ query_tags: 'team:platform' });
+  });
+
+  it('forwards multiple tags as comma-separated key:value pairs', async () => {
+    await session.executeStatement('SELECT 1', {
+      queryTags: { team: 'platform', env: 'staging' },
+    });
+    const opts = executeSpy.firstCall.args[1] as SeaNativeExecuteOptions;
+    expect(opts.statementConf!.query_tags).to.match(
+      /^(team:platform,env:staging|env:staging,team:platform)$/,
+    );
+  });
+
+  it('preserves null-valued tags as bare keys (no colon)', async () => {
+    await session.executeStatement('SELECT 1', {
+      queryTags: { highPriority: null, team: 'platform' },
+    });
+    const opts = executeSpy.firstCall.args[1] as SeaNativeExecuteOptions;
+    const encoded = opts.statementConf!.query_tags;
+    // Object iteration is insertion-order for string keys; serializeQueryTags
+    // follows that. Two possible orderings depending on Object.keys order.
+    expect(encoded).to.match(
+      /^(highPriority,team:platform|team:platform,highPriority)$/,
+    );
+  });
+
+  it('preserves undefined-valued tags as bare keys (no colon)', async () => {
+    await session.executeStatement('SELECT 1', {
+      queryTags: { highPriority: undefined, team: 'platform' },
+    });
+    const opts = executeSpy.firstCall.args[1] as SeaNativeExecuteOptions;
+    expect(opts.statementConf!.query_tags).to.contain('highPriority');
+    expect(opts.statementConf!.query_tags).to.contain('team:platform');
+  });
+
+  it('escapes special chars (colon, comma, backslash) in values', async () => {
+    await session.executeStatement('SELECT 1', {
+      queryTags: { k: 'a:b,c\\d' },
+    });
+    const opts = executeSpy.firstCall.args[1] as SeaNativeExecuteOptions;
+    // `:` → `\:`, `,` → `\,`, `\` → `\\`
+    expect(opts.statementConf!.query_tags).to.equal('k:a\\:b\\,c\\\\d');
+  });
+
+  it('escapes backslashes in keys', async () => {
+    await session.executeStatement('SELECT 1', {
+      queryTags: { 'k\\1': 'v' },
+    });
+    const opts = executeSpy.firstCall.args[1] as SeaNativeExecuteOptions;
+    expect(opts.statementConf!.query_tags).to.equal('k\\\\1:v');
+  });
+});