feat(sea): wire rowLimit, statementConf, queryTags + TIMESTAMP_NTZ/LTZ params

Four statement-option / param-type gaps where the kernel + napi were already
ready but the node SEA layer didn't expose or thread them, so they were
silently dropped vs the Thrift backend.

- queryTags: `ExecuteStatementOptions.queryTags` is a public option but
  `SeaSessionBackend.executeStatement` never forwarded it. Now serialised
  JS-side via `serializeQueryTags` into the conf overlay's `query_tags` key
  (the same wire shape the Thrift backend produces), so null-valued tags
  round-trip — the napi `queryTags` field is a `HashMap<String,String>` that
  can't represent nulls, and the kernel rejects setting both.
- rowLimit: new `ExecuteStatementOptions.rowLimit` → napi `rowLimit`
  (SEA `row_limit`). SEA-only server-side cap; Thrift has no execute-time cap.
- statementConf: new `ExecuteStatementOptions.statementConf` → napi
  `statementConf` (SEA `statement_conf`), the Thrift `confOverlay` equivalent;
  queryTags merge into it under `query_tags`.
- TIMESTAMP_NTZ / TIMESTAMP_LTZ: added to `DBSQLParameterType` so callers can
  bind timezone-explicit timestamp params. `toSparkParameter` already honours
  an explicit type, and `SeaPositionalParams` passes the SQL type verbatim to
  the kernel codec (which has the NTZ/LTZ arms). Without these a migrating
  caller silently coerces NTZ/LTZ columns to TIMESTAMP.

231 SEA unit tests pass; verified live against pecotesting (rowLimit caps rows,
NTZ param round-trips, tags/conf accepted).

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

Author: msrathore-db

lib/DBSQLParameter.ts

@@ -8,6 +8,14 @@ export enum DBSQLParameterType {
   STRING = 'STRING',
   DATE = 'DATE',
   TIMESTAMP = 'TIMESTAMP',
+  // Timezone-explicit timestamp variants. A bare `Date` value defaults to
+  // `TIMESTAMP`; set one of these explicitly to bind a TIMESTAMP_NTZ
+  // (no timezone, wall-clock) or TIMESTAMP_LTZ (local timezone) parameter.
+  // The Thrift wire only has `TIMESTAMP`; these are SEA-path types the kernel
+  // param codec accepts — without them a migrating caller silently coerces
+  // NTZ/LTZ columns to TIMESTAMP.
+  TIMESTAMP_NTZ = 'TIMESTAMP_NTZ',
+  TIMESTAMP_LTZ = 'TIMESTAMP_LTZ',
   FLOAT = 'FLOAT',
   DECIMAL = 'DECIMAL',
   DOUBLE = 'DOUBLE',

lib/contracts/IDBSQLSession.ts

@@ -27,6 +27,20 @@ export type ExecuteStatementOptions = {
    * These tags apply only to this statement and do not persist across queries.
    */
   queryTags?: Record<string, string | null | undefined>;
+  /**
+   * Server-side cap on the number of rows the statement returns (SEA path only).
+   * Maps to the kernel's `row_limit` / SEA `row_limit`. The Thrift backend has no
+   * execute-time server cap, so this is a no-op there; use `maxRows` for the
+   * client-side per-fetch chunk size on both backends.
+   */
+  rowLimit?: number;
+  /**
+   * Arbitrary per-statement configuration overlay (SEA path only). Maps to the
+   * kernel's `statement_conf` / SEA `statement_conf`, the same mechanism the
+   * Thrift backend exposes as `confOverlay`. `queryTags` are merged into this map
+   * under the `query_tags` key, mirroring the Thrift wire shape.
+   */
+  statementConf?: Record<string, string>;
 };
 
 export type TypeInfoRequest = {

lib/sea/SeaSessionBackend.ts

@@ -38,6 +38,7 @@ import SeaTableTypeFilter from './SeaTableTypeFilter';
 import { seaServerInfoValue } from './SeaServerInfo';
 import { buildSeaPositionalParams, buildSeaNamedParams } from './SeaPositionalParams';
 import ParameterError from '../errors/ParameterError';
+import { serializeQueryTags } from '../utils';
 
 export interface SeaSessionBackendOptions {
   /** The opaque napi `Connection` handle returned by `openSession`. */
@@ -115,19 +116,17 @@ export default class SeaSessionBackend implements ISessionBackend {
   /**
    * 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.
+   * Catalog / schema / sessionConf were applied at session open. The
+   * per-statement options threaded here mirror the Thrift backend:
+   * `ordinalParameters` / `namedParameters` (bound params), `queryTimeout`
+   * (server wait timeout), `queryTags` (serialised into the conf overlay's
+   * `query_tags` key), `statementConf` (arbitrary conf overlay), and
+   * `rowLimit` (SEA-only server-side row cap).
    *
-   * M0 intentionally rejects `queryTimeout`, `namedParameters`, and
-   * `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.
+   * `useCloudFetch` is a no-op on the SEA path — the kernel hardcodes the
+   * SEA `disposition` to `INLINE_OR_EXTERNAL_LINKS`; cloud-fetch behaviour
+   * is governed by the kernel's `ResultConfig`. `maxRows` is the
+   * client-side per-fetch chunk size, applied by the facade, not here.
    */
   public async executeStatement(statement: string, options: ExecuteStatementOptions): Promise<IOperationBackend> {
     this.failIfClosed();
@@ -155,6 +154,27 @@ export default class SeaSessionBackend implements ISessionBackend {
     if (options.queryTimeout !== undefined) {
       nativeOptions.queryTimeoutSecs = Number(options.queryTimeout);
     }
+    // Server-side row cap (SEA `row_limit`). SEA-only — the Thrift backend has
+    // no execute-time server cap, so there is no parity obligation here.
+    if (options.rowLimit !== undefined) {
+      nativeOptions.rowLimit = Number(options.rowLimit);
+    }
+    // Per-statement conf overlay (`statement_conf`) plus query tags. Tags are
+    // serialised JS-side into the `query_tags` key (the same wire shape the
+    // Thrift backend produces via `serializeQueryTags` → `confOverlay`), rather
+    // than via the napi `queryTags` field: napi's `HashMap<String,String>`
+    // can't represent a null-valued tag, and the kernel rejects setting both
+    // the `queryTags` field and a `query_tags` conf key.
+    const serializedQueryTags = serializeQueryTags(options.queryTags);
+    if (options.statementConf !== undefined || serializedQueryTags !== undefined) {
+      const statementConf: Record<string, string> = { ...(options.statementConf ?? {}) };
+      if (serializedQueryTags !== undefined) {
+        statementConf.query_tags = serializedQueryTags;
+      }
+      if (Object.keys(statementConf).length > 0) {
+        nativeOptions.statementConf = statementConf;
+      }
+    }
     const hasOptions = Object.keys(nativeOptions).length > 0;
 
     // Submit asynchronously (kernel `wait_timeout=0s`): the server

tests/unit/sea/execution.test.ts

@@ -455,6 +455,50 @@ describe('SeaSessionBackend', () => {
     expect((thrown as Error).message).to.match(/both ordinal and named/);
   });
 
+  it('executeStatement forwards rowLimit as napi rowLimit', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    await session.executeStatement('SELECT 1', { rowLimit: 500 });
+    expect(connection.lastOptions?.rowLimit).to.equal(500);
+  });
+
+  it('executeStatement forwards statementConf verbatim as napi statementConf', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    await session.executeStatement('SELECT 1', { statementConf: { 'spark.sql.ansi.enabled': 'true' } });
+    expect(connection.lastOptions?.statementConf).to.deep.equal({ 'spark.sql.ansi.enabled': 'true' });
+  });
+
+  it('executeStatement serialises queryTags into statementConf.query_tags (Thrift wire shape)', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    await session.executeStatement('SELECT 1', { queryTags: { team: 'data', env: 'prod' } });
+    expect(connection.lastOptions?.statementConf?.query_tags).to.equal('team:data,env:prod');
+    // queryTags must NOT use the napi `queryTags` field (can't carry null tags;
+    // kernel rejects both field + conf key).
+    expect(connection.lastOptions?.queryTags).to.equal(undefined);
+  });
+
+  it('executeStatement carries a null-valued queryTag as a key-only segment', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    await session.executeStatement('SELECT 1', { queryTags: { audited: null } });
+    expect(connection.lastOptions?.statementConf?.query_tags).to.equal('audited');
+  });
+
+  it('executeStatement merges queryTags into a provided statementConf', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    await session.executeStatement('SELECT 1', {
+      statementConf: { 'spark.sql.ansi.enabled': 'true' },
+      queryTags: { team: 'data' },
+    });
+    expect(connection.lastOptions?.statementConf).to.deep.equal({
+      'spark.sql.ansi.enabled': 'true',
+      query_tags: 'team:data',
+    });
+  });
+
   it('executeStatement uses the no-options fast path when nothing is bound', async () => {
     const connection = new FakeNativeConnection();
     const session = makeSession(connection);

tests/unit/sea/positionalParams.test.ts

@@ -54,6 +54,25 @@ describe('SeaPositionalParams.buildSeaPositionalParams', () => {
       { sqlType: 'TIMESTAMP', value: '2024-01-15 10:30:00' },
     ]);
   });
+
+  it('honours explicit TIMESTAMP_NTZ / TIMESTAMP_LTZ types (kernel param codec)', () => {
+    expect(
+      buildSeaPositionalParams([
+        new DBSQLParameter({ type: DBSQLParameterType.TIMESTAMP_NTZ, value: '2024-01-15 10:30:00' }),
+        new DBSQLParameter({ type: DBSQLParameterType.TIMESTAMP_LTZ, value: '2024-01-15 10:30:00' }),
+      ]),
+    ).to.deep.equal([
+      { sqlType: 'TIMESTAMP_NTZ', value: '2024-01-15 10:30:00' },
+      { sqlType: 'TIMESTAMP_LTZ', value: '2024-01-15 10:30:00' },
+    ]);
+  });
+
+  it('routes a Date with explicit TIMESTAMP_NTZ type as NTZ (not the default TIMESTAMP)', () => {
+    const d = new Date('2024-01-15T10:30:00.000Z');
+    expect(
+      buildSeaPositionalParams([new DBSQLParameter({ type: DBSQLParameterType.TIMESTAMP_NTZ, value: d })]),
+    ).to.deep.equal([{ sqlType: 'TIMESTAMP_NTZ', value: d.toISOString() }]);
+  });
 });
 
 describe('SeaPositionalParams.buildSeaNamedParams', () => {