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

Three statement-option / param-type additions where the kernel + napi were
already ready but the node SEA layer didn't expose them:

- 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.
  Generalises the existing query_tags serialisation so a caller-supplied
  statementConf and queryTags merge into one conf map (queryTags already
  forwarded upstream).
- 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).

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

@@ -117,19 +117,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();
@@ -175,15 +173,26 @@ export default class SeaSessionBackend implements ISessionBackend {
     if (options.queryTimeout !== undefined) {
       nativeOptions.queryTimeoutSecs = Number(options.queryTimeout);
     }
-    // Query tags: serialise JS-side into the conf overlay's `query_tags` key
-    // (the same wire shape the Thrift backend produces via `serializeQueryTags`
-    // → `confOverlay`). Not forwarded via the napi `queryTags` field: that's a
-    // `HashMap<String,String>` which can't represent a null-valued tag, and the
-    // kernel rejects setting both the field and a `query_tags` conf key. A
-    // null-valued tag therefore round-trips as a key-only segment.
+    // 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 (serializedQueryTags !== undefined) {
-      nativeOptions.statementConf = { query_tags: serializedQueryTags };
+    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;
 

tests/unit/sea/execution.test.ts

@@ -506,6 +506,33 @@ describe('SeaSessionBackend', () => {
     expect(connection.lastListSchemasArgs).to.deep.equal([undefined, '%']);
   });
 
+  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 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', () => {