fix(sea): address #410 review — error fidelity, fetch perf, validation, tests

Addresses the 8 review threads on PR #410. Validated against a live
warehouse (results-e2e parity gate + interval-duration-e2e + execution-e2e
all pass) plus new/updated unit tests.

- SeaOperationLifecycle: rethrowKernelError now delegates to the canonical
  decodeNapiKernelError, so cancel/close errors get the same fidelity as
  fetch errors — the sqlState remap (envelope field is `sqlState`, the old
  code read `sqlstate` and dropped it), the kernelMetadata namespace, and
  the strict `startsWith` sentinel match (was a loose `indexOf >= 0`).

- SeaArrowIpc: replace decodeIpcBatch (full RecordBatchReader
  materialization just to sum row counts) with countRowsInIpc, which reads
  RecordBatch header `length` via MessageReader and skips bodies — no
  vector decode. Removes ~2x Arrow decode CPU + transient allocation on the
  fetch hot path (the converter still re-decodes for values). SeaResultsProvider
  switched to it.

- SeaArrowIpc: hermetic unit tests (tests/unit/sea/SeaArrowIpc.test.ts) for
  the framing walk, no-op/garbage rewrite paths, the row-count path, and the
  empty-schema guard. (The Duration-positive rewrite stays covered by the
  live e2e — apache-arrow@13 can't construct a Duration column hermetically.)

- SeaOperationBackend: on a fetch-error cleanup close() that also fails, log
  the failure at warn (statement may leak) instead of fully swallowing it —
  the original fetch error is still surfaced.

- SeaSessionBackend: reject queryTags / useLZ4Compression /
  stagingAllowedLocalPath with M0-style errors instead of silently ignoring
  them (+ unit tests). Silent no-ops are the worst failure mode for callers.

- SeaArrowIpc: throw a typed HiveDriverError (not a raw TypeError) when an
  IPC payload carries no schema.

- SeaArrowIpcDurationFix.readMessageAt: fail-closed guards for negative
  metadataLength / bodyLength (was relying on subarray clamping).

- Fix stale `tests/integration/sea/...` doc refs → `tests/e2e/sea/`.

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

Author: msrathore-db

lib/sea/SeaArrowIpc.ts

@@ -12,9 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-import { RecordBatchReader, Schema, Field, DataType, TypeMap } from 'apache-arrow';
+import { RecordBatchReader, MessageReader, MessageHeader, Schema, Field, DataType, TypeMap } from 'apache-arrow';
 import { TTableSchema, TTypeId, TPrimitiveTypeEntry } from '../../thrift/TCLIService_types';
 import { rewriteDurationToInt64, DURATION_UNIT_METADATA_KEY } from './SeaArrowIpcDurationFix';
+import HiveDriverError from '../errors/HiveDriverError';
 
 /**
  * Field metadata key used by the kernel to attach the original Databricks
@@ -23,43 +24,46 @@ import { rewriteDurationToInt64, DURATION_UNIT_METADATA_KEY } from './SeaArrowIp
 const DATABRICKS_TYPE_NAME = 'databricks.type_name';
 
 /**
- * Decode an Arrow IPC stream payload (schema header + zero-or-more
- * record-batch messages) into its row count.
+ * Sum the row counts of every RecordBatch message in an Arrow IPC
+ * stream, WITHOUT materializing the Arrow vector tree.
  *
- * Returns `{ schema, rowCount }`. The schema is left intact as the
- * apache-arrow Schema object so callers can reuse it; the rowCount is
- * the sum of `RecordBatch.numRows` across every record-batch message
- * in the stream.
+ * Why this exists: `ArrowResultConverter` consumes `ArrowBatch` objects
+ * that carry an explicit `rowCount`, but the kernel's IPC payload only
+ * carries per-RecordBatch `length` (no separate total). `SeaResultsProvider`
+ * needs that count to build the `ArrowBatch` it hands to the converter —
+ * which then re-decodes the same bytes for the actual values.
  *
- * Why we parse upfront: `ArrowResultConverter` consumes `ArrowBatch`
- * objects which carry an explicit `rowCount`. The kernel's IPC payload
- * does not carry a separate count — only per-RecordBatch numRows. We
- * walk the messages once to sum them so the converter sees the same
- * shape as the thrift path (`ArrowResultHandler.fetchNext` at
- * `lib/result/ArrowResultHandler.ts:55`).
+ * The previous implementation used `RecordBatchReader` and iterated the
+ * batches, which calls `_loadVectors` and materializes the full vector
+ * tree for every batch just to read `numRows` and discard everything
+ * else — ~2x Arrow decode CPU + transient allocation on the fetch hot
+ * path. `MessageReader` instead reads only each message's FlatBuffer
+ * metadata header (where `RecordBatch.length` lives) and skips the body
+ * bytes, so no vectors are decoded here. The converter's later re-decode
+ * is the only real materialization.
  *
- * Re-parsing inside the converter is unavoidable because `RecordBatch`
- * instances created here cannot be passed across the converter's
- * `Buffer[]` boundary without rewriting the converter. Callers that already
- * patched the IPC bytes can set `alreadyPatched` to avoid running the
- * FlatBuffer rewrite twice on the hot fetch path.
+ * `ipcBytes` must already be Duration-patched (row count is unaffected
+ * by the Duration→Int64 rewrite, and the framing is unchanged). Returns
+ * 0 for an empty / schema-only stream.
  */
-export function decodeIpcBatch(
-  ipcBytes: Buffer,
-  options: { alreadyPatched?: boolean } = {},
-): { schema: Schema<TypeMap>; rowCount: number } {
-  const patched = options.alreadyPatched ? ipcBytes : rewriteDurationToInt64(ipcBytes);
-  const reader = RecordBatchReader.from<TypeMap>(patched);
-  // Eagerly open so `schema` is populated.
-  reader.open();
-  const { schema } = reader;
-
+export function countRowsInIpc(ipcBytes: Buffer): number {
+  const reader = new MessageReader(ipcBytes);
   let rowCount = 0;
-  // Iterate all record batches in the stream and sum row counts.
-  for (const batch of reader) {
-    rowCount += batch.numRows;
+  for (const message of reader) {
+    if (message.headerType === MessageHeader.RecordBatch) {
+      // header() for a RecordBatch message carries `length` (the row
+      // count) in the FlatBuffer metadata — no body decode needed.
+      rowCount += Number((message.header() as { length: number | bigint }).length);
+    }
+    // Advance past the (undecoded) body so the next message reads at the
+    // correct offset. readMessageBody returns a view; it does not decode
+    // the body into Arrow vectors.
+    const bodyLength = Number(message.bodyLength);
+    if (bodyLength > 0) {
+      reader.readMessageBody(bodyLength);
+    }
   }
-  return { schema, rowCount };
+  return rowCount;
 }
 
 /**
@@ -70,6 +74,14 @@ export function decodeIpcSchema(ipcBytes: Buffer): Schema<TypeMap> {
   const patched = rewriteDurationToInt64(ipcBytes);
   const reader = RecordBatchReader.from<TypeMap>(patched);
   reader.open();
+  // `RecordBatchReader.from(emptyBuffer).open()` does not throw — it
+  // leaves `schema` undefined. Without this guard a downstream
+  // `arrowSchemaToThriftSchema(undefined)` would hit `undefined.fields`
+  // and surface a raw TypeError instead of a typed driver error. The
+  // real kernel always materialises a schema, so this is defensive.
+  if (!reader.schema) {
+    throw new HiveDriverError('SEA result: Arrow IPC stream carried no schema (empty or truncated payload)');
+  }
   return reader.schema;
 }
 

lib/sea/SeaArrowIpcDurationFix.ts

@@ -196,7 +196,12 @@ function readMessageAt(
     cursor += 4;
   }
 
-  if (metadataLength === 0) {
+  // A zero or negative length (other than the -1 continuation sentinel
+  // handled above) is malformed. Reject by intent rather than relying on
+  // `subarray` clamping: a negative `metadataLength` would make
+  // `metadataEnd < cursor`, which silently passes the `> byteLength`
+  // upper-bound check below.
+  if (metadataLength <= 0) {
     return null;
   }
 
@@ -211,6 +216,12 @@ function readMessageAt(
   const message = Message.getRootAsMessage(bb);
 
   const bodyLength = Number(message.bodyLength());
+  // Same fail-closed intent as metadataLength: a negative body length
+  // (server-controlled) would make `bodyEnd < bodyStart` and slip past
+  // the upper-bound check.
+  if (bodyLength < 0) {
+    return null;
+  }
   const bodyStart = metadataEnd;
   const bodyEnd = bodyStart + bodyLength;
   if (bodyEnd > view.byteLength) {

lib/sea/SeaOperationBackend.ts

@@ -21,7 +21,7 @@
  *   `ArrowResultConverter` (Phase 1 + Phase 2; reused unchanged) →
  *   `ResultSlicer` (chunk-size normalisation; reused unchanged). The M0
  *   row shape is byte-identical to the thrift path for every M0
- *   datatype (parity gate exercised by `tests/integration/sea/results-e2e.test.ts`).
+ *   datatype (parity gate exercised by `tests/e2e/sea/results-e2e.test.ts`).
  *
  * - **Lifecycle (from sea-operation):** `cancel()` / `close()` /
  *   `finished()` (alias of `waitUntilReady`) delegate to the helpers
@@ -42,6 +42,7 @@ import IOperationBackend, { IOperationBackendWaitOptions } from '../contracts/IO
 import { OperationStatus, OperationState } from '../contracts/OperationStatus';
 import { ResultMetadata, ResultFormat } from '../contracts/ResultMetadata';
 import IClientContext from '../contracts/IClientContext';
+import { LogLevel } from '../contracts/IDBSQLLogger';
 import Status from '../dto/Status';
 import HiveDriverError from '../errors/HiveDriverError';
 import ArrowResultConverter from '../result/ArrowResultConverter';
@@ -155,7 +156,22 @@ export default class SeaOperationBackend implements IOperationBackend {
       // error ("call close() and discard"). Close the statement so the server
       // reclaims it promptly — best-effort, so a close failure never masks the
       // original fetch error — then surface a typed kernel error.
-      await seaClose(this.lifecycle, this.statement, this.context, this._id).catch(() => undefined);
+      //
+      // If close() ALSO fails, seaClose has reset isClosed back to false and
+      // the kernel-side statement handle is now leaked (the stream is already
+      // wedged, so nothing downstream forces another close). We still don't
+      // mask the original fetch error, but log the close failure at warn so
+      // the leak is diagnosable rather than completely invisible.
+      await seaClose(this.lifecycle, this.statement, this.context, this._id).catch((closeErr) => {
+        const cause = closeErr instanceof Error ? closeErr.message : String(closeErr);
+        this.context
+          .getLogger()
+          .log(
+            LogLevel.warn,
+            `SEA fetch-error cleanup: close() failed for operation ${this._id}; the server-side ` +
+              `statement may leak until the session is closed. Cause: ${cause}`,
+          );
+      });
       throw decodeNapiKernelError(err);
     }
   }

lib/sea/SeaOperationLifecycle.ts

@@ -47,7 +47,7 @@ import Status from '../dto/Status';
 import { OperationStatus, OperationState } from '../contracts/OperationStatus';
 import { LogLevel } from '../contracts/IDBSQLLogger';
 import IClientContext from '../contracts/IClientContext';
-import { mapKernelErrorToJsError, KernelErrorShape } from './SeaErrorMapping';
+import { decodeNapiKernelError } from './SeaErrorMapping';
 import OperationStateError, { OperationStateErrorCode } from '../errors/OperationStateError';
 
 /**
@@ -89,33 +89,18 @@ export function createLifecycleState(): SeaOperationLifecycleState {
 
 /**
  * Normalise an error thrown by the napi `Statement` into one of the
- * driver's typed error classes. The binding surfaces kernel errors as
- * a JSON envelope on `napi::Error.reason` with the sentinel prefix
- * `__databricks_error__:` (see the napi-binding round 2 findings,
- * section "JSON-envelope error reason"). If we can parse out a kernel
- * payload, we route it through `mapKernelErrorToJsError`; otherwise
- * the original error is rethrown unchanged.
+ * driver's typed error classes, then throw it.
+ *
+ * Delegates to the canonical {@link decodeNapiKernelError} so cancel /
+ * close errors get exactly the same fidelity as fetch errors: the
+ * `sqlState` remap (the envelope field is `sqlState`, not `sqlstate`),
+ * the `kernelMetadata` namespace (vendorCode / httpStatus / retryable /
+ * queryId), and the strict `startsWith` sentinel match. The previous
+ * hand-rolled reimplementation here dropped SQLSTATE and metadata and
+ * used a looser substring match.
  */
 function rethrowKernelError(err: unknown): never {
-  if (err instanceof Error && typeof err.message === 'string') {
-    const sentinel = '__databricks_error__:';
-    const idx = err.message.indexOf(sentinel);
-    if (idx >= 0) {
-      const json = err.message.slice(idx + sentinel.length);
-      let parsed: KernelErrorShape | undefined;
-      try {
-        parsed = JSON.parse(json) as KernelErrorShape;
-      } catch {
-        // Malformed envelope — fall through and rethrow the original
-        // below; we never silently drop a kernel error.
-        parsed = undefined;
-      }
-      if (parsed) {
-        throw mapKernelErrorToJsError(parsed);
-      }
-    }
-  }
-  throw err;
+  throw decodeNapiKernelError(err);
 }
 
 /**

lib/sea/SeaResultsProvider.ts

@@ -14,7 +14,7 @@
 
 import IResultsProvider, { ResultsProviderFetchNextOptions } from '../result/IResultsProvider';
 import { ArrowBatch } from '../result/utils';
-import { decodeIpcBatch, patchIpcBytes } from './SeaArrowIpc';
+import { countRowsInIpc, patchIpcBytes } from './SeaArrowIpc';
 
 /**
  * The minimal slice of the napi-binding `Statement` class that we
@@ -102,11 +102,15 @@ export default class SeaResultsProvider implements IResultsProvider<ArrowBatch>
       // Patch the raw bytes once: rewrite any Arrow `Duration` field to
       // `Int64` with a `databricks.arrow.duration_unit` marker, so that
       // apache-arrow@13 (which predates Duration support) can decode the
-      // stream. `decodeIpcBatch` is told these bytes are already patched;
-      // the downstream `RecordBatchReader.from` inside `ArrowResultConverter`
-      // sees the same patched buffer. See `SeaArrowIpcDurationFix.ts`.
+      // stream. The downstream `RecordBatchReader.from` inside
+      // `ArrowResultConverter` sees the same patched buffer. See
+      // `SeaArrowIpcDurationFix.ts`.
       const ipcBytes = patchIpcBytes(next.ipcBytes);
-      const { rowCount } = decodeIpcBatch(ipcBytes, { alreadyPatched: true });
+      // Row count only — `countRowsInIpc` reads the RecordBatch metadata
+      // headers without materializing vectors (the converter re-decodes
+      // the bytes for the actual values). Avoids a full second Arrow
+      // decode on the fetch hot path.
+      const rowCount = countRowsInIpc(ipcBytes);
       // Skip empty batches — the converter handles them but pre-filtering here
       // avoids a round-trip through the converter's prefetch loop. Continue to
       // find a non-empty batch or hit exhaustion.

lib/sea/SeaSessionBackend.ts

@@ -116,6 +116,25 @@ export default class SeaSessionBackend implements ISessionBackend {
         'SEA executeStatement: useCloudFetch is controlled by the kernel result configuration and is not a per-statement option on SEA',
       );
     }
+    // Reject — rather than silently ignore — the remaining Thrift-path
+    // options the SEA M0 backend does not honor. Silently dropping them
+    // is the worst failure mode for an agent/caller: passing e.g.
+    // `queryTags` or `useLZ4Compression` would no-op with zero signal.
+    // (`maxRows` is intentionally NOT here — the facade applies it at
+    // fetch time.)
+    if (options.queryTags !== undefined) {
+      throw new HiveDriverError('SEA executeStatement: queryTags is not supported in M0 (deferred to M1)');
+    }
+    if (options.useLZ4Compression !== undefined) {
+      throw new HiveDriverError(
+        'SEA executeStatement: useLZ4Compression is not supported on SEA (result compression is governed by the kernel)',
+      );
+    }
+    if (options.stagingAllowedLocalPath !== undefined) {
+      throw new HiveDriverError(
+        'SEA executeStatement: stagingAllowedLocalPath (volume operations) is not supported in M0 (deferred to M1)',
+      );
+    }
 
     let nativeStatement;
     try {

tests/unit/sea/SeaArrowIpc.test.ts

@@ -0,0 +1,102 @@
+// 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.
+
+import { expect } from 'chai';
+import { tableFromArrays, RecordBatchStreamWriter } from 'apache-arrow';
+import { countRowsInIpc, decodeIpcSchema, patchIpcBytes } from '../../../lib/sea/SeaArrowIpc';
+import { rewriteDurationToInt64 } from '../../../lib/sea/SeaArrowIpcDurationFix';
+import HiveDriverError from '../../../lib/errors/HiveDriverError';
+
+// Hermetic coverage for the SEA Arrow-IPC layer. apache-arrow@13 cannot
+// construct a `Duration` column, so the Duration-positive rewrite path is
+// covered by the live e2e (tests/e2e/sea/interval-duration-e2e.test.ts).
+// These tests pin everything reachable without a warehouse: the IPC
+// framing walk, the no-op / malformed-input handling, the cheap
+// row-count path, and the schema-decode guard.
+
+/** Build a multi-message Arrow IPC stream from arrays of per-batch row data. */
+function makeIpcStream(batchRows: number[][]): Buffer {
+  const writer = new RecordBatchStreamWriter();
+  for (const rows of batchRows) {
+    const table = tableFromArrays({ a: Int32Array.from(rows) });
+    writer.write(table.batches[0]);
+  }
+  writer.finish();
+  return Buffer.from(writer.toUint8Array(true));
+}
+
+/** Schema-only IPC stream (no record batches). */
+function makeSchemaOnlyStream(): Buffer {
+  const writer = new RecordBatchStreamWriter();
+  const table = tableFromArrays({ a: Int32Array.from([1]) });
+  // Start the stream (emits the schema) then finish without writing a batch.
+  writer.reset(undefined, table.schema);
+  writer.finish();
+  return Buffer.from(writer.toUint8Array(true));
+}
+
+describe('SeaArrowIpc.countRowsInIpc', () => {
+  it('sums RecordBatch row counts across messages', () => {
+    const ipc = makeIpcStream([
+      [1, 2, 3],
+      [4, 5],
+    ]);
+    expect(countRowsInIpc(ipc)).to.equal(5);
+  });
+
+  it('returns 0 for a schema-only stream (no record batches)', () => {
+    expect(countRowsInIpc(makeSchemaOnlyStream())).to.equal(0);
+  });
+
+  it('counts a single-batch stream', () => {
+    expect(countRowsInIpc(makeIpcStream([[10, 20, 30, 40]]))).to.equal(4);
+  });
+});
+
+describe('SeaArrowIpc.rewriteDurationToInt64 (no-Duration / malformed paths)', () => {
+  it('is a no-op for a stream with no Duration field (returns input unchanged)', () => {
+    const ipc = makeIpcStream([[1, 2, 3]]);
+    const out = rewriteDurationToInt64(ipc);
+    expect(out.equals(ipc)).to.equal(true);
+  });
+
+  it('returns the input unchanged for an empty buffer (no throw)', () => {
+    const empty = Buffer.alloc(0);
+    const out = rewriteDurationToInt64(empty);
+    expect(out.byteLength).to.equal(0);
+  });
+
+  it('does not throw on garbage / truncated bytes (fail-closed framing)', () => {
+    // Random bytes are not a valid IPC stream; readMessageAt must reject
+    // them (the negative-length and bounds guards) rather than crash.
+    const garbage = Buffer.from([0x01, 0x02, 0x03, 0x04, 0xff, 0xff, 0xff, 0xff]);
+    expect(() => rewriteDurationToInt64(garbage)).to.not.throw();
+  });
+
+  it('patchIpcBytes is byte-identical to the input when no Duration is present', () => {
+    const ipc = makeIpcStream([[7, 8]]);
+    expect(patchIpcBytes(ipc).equals(ipc)).to.equal(true);
+  });
+});
+
+describe('SeaArrowIpc.decodeIpcSchema', () => {
+  it('decodes the schema of a normal stream', () => {
+    const schema = decodeIpcSchema(makeIpcStream([[1]]));
+    expect(schema.fields.map((f) => f.name)).to.deep.equal(['a']);
+  });
+
+  it('throws a typed HiveDriverError (not a raw TypeError) on an empty payload', () => {
+    expect(() => decodeIpcSchema(Buffer.alloc(0))).to.throw(HiveDriverError);
+  });
+});