fix(sea): address #410 review — fetch cleanup, cooperative cancel, parity, docs

Validated each finding against a live pecotesting warehouse first; the headline
INTERVAL story turned out to be split-artifact, not breakage.

- F7: getResultMetadata stored the *unpatched* Duration IPC bytes in
  meta.arrowSchema while advertising ArrowBased — store the patched bytes so an
  ArrowBased consumer doesn't hit `Unrecognized type "Duration" (18)`.
- F3: fetchChunk now honors the `isClosed` cooperative-cancel probe (parity with
  ThriftOperationBackend) at its yield points.
- F6: on a fetch error, best-effort close the statement (napi contract: stream
  is unspecified after Err) and surface a typed kernel error via decodeNapiKernelError.
- F9: cancel-after-fetch now throws the canonical OperationStateError(Canceled)
  ("The operation was canceled by a client") — byte-matches the Thrift message.
- F10: typed HiveDriverError (not raw Error) in the schema/fetchNextBatch guards.
- F1: corrected SeaArrowIpcDurationFix docs — on this layer the rewriter only
  makes Duration *decodable* (raw Int64); the duration_unit formatter lands in
  #411 (verified live: byte-identical to Thrift).
- F5: documented that nested Duration is a SHARED apache-arrow@13 limitation —
  verified the Thrift backend throws the identical error, so SEA matches parity.
- F2: added a live e2e that drives a real Arrow Duration column through the
  rewriter (asserts no "Duration (18)" crash + raw-Int64 on this layer).
- F8: pinned the no-`Failed` invariant in status() (failures reject at submit).
- F12: renamed SeaResultsProvider's SeaStatementHandle → SeaFetchHandle (was a
  name collision with the lifecycle interface of a different shape).
- F13: dropped the no-op await on the synchronous statement.schema().
- F14: fixed the Float-precision comment (Precision enum, not bit-width).
- F15: SeaResultsProvider.prime loops instead of self-recursing on empty batches.

Deferred (noted on the PR): F4 (per-batch triple-decode perf) and F11
(hasResultSet() hard-coded true for M0).

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

Author: msrathore-db

lib/sea/SeaArrowIpc.ts

@@ -205,7 +205,9 @@ function arrowTypeToTTypeId(field: Field<DataType>): TTypeId {
     }
   }
   if (DataType.isFloat(arrowType)) {
-    // arrow Float precision: 16=HALF, 32=SINGLE, 64=DOUBLE
+    // `precision` is the Arrow `Precision` enum (HALF=0, SINGLE=1, DOUBLE=2),
+    // NOT a bit-width — `=== 2` is DOUBLE. Everything else (HALF/SINGLE) maps
+    // to the thrift FLOAT type.
     return arrowType.precision === 2 ? TTypeId.DOUBLE_TYPE : TTypeId.FLOAT_TYPE;
   }
   if (DataType.isDecimal(arrowType)) return TTypeId.DECIMAL_TYPE;

lib/sea/SeaArrowIpcDurationFix.ts

@@ -32,17 +32,24 @@
  * bytes pass through unchanged. We embed the original `Duration` time
  * unit (`SECOND`/`MILLISECOND`/`MICROSECOND`/`NANOSECOND`) into the
  * rewritten field's `custom_metadata` under the key
- * `databricks.arrow.duration_unit` so the JS converter can format the
- * Int64 value back into a thrift-equivalent string (e.g.
- * `"1 02:03:04.000000000"`).
+ * `databricks.arrow.duration_unit`.
+ *
+ * **Scope on this layer (SEA execution + results, PR 2/3):** the rewrite's
+ * job here is purely to make the stream *decodable* by apache-arrow@13. The
+ * resulting Int64 surfaces to callers as a raw microsecond/nanosecond number
+ * on this layer. The consumer that reads the `duration_unit` marker and
+ * formats it back into the thrift-equivalent string (e.g.
+ * `"1 02:03:04.000000000"`) lands in PR 3/3 (#411) — verified against a live
+ * warehouse to produce byte-identical output to the Thrift path. Until that
+ * consumer merges, INTERVAL DAY-TIME columns are raw Int64 under SEA.
  *
  * Why this lives in its own file: the rewriter is the only place in the
  * codebase that needs to construct FlatBuffers by hand using the
  * `flatbuffers` library; isolating it keeps `SeaArrowIpc.ts` focused on
  * the high-level Arrow-decoded views.
  *
- * @see lib/result/ArrowResultConverter.ts — Phase-1 INTERVAL formatting
- *      reads the metadata key written here.
+ * @see lib/result/ArrowResultConverter.ts — the Phase-1 INTERVAL formatter
+ *      (PR 3/3) reads the metadata key written here.
  * @see findings/parity-mismatch/round5-implementation-2026-05-15.md —
  *      original failure mode (`Unrecognized type: "Duration" (18)`).
  */
@@ -234,10 +241,17 @@ function maybeRewriteSchemaMessage(schemaMessageBytes: Buffer): Buffer | null {
     return null;
   }
 
-  // Scan top-level fields and children for Duration. We rewrite only
-  // top-level Duration fields for M0 (Spark INTERVAL DAY-TIME surfaces
-  // as a top-level column — children of Struct/List/Map are out of
-  // scope until we see a real-world payload with nested Duration).
+  // We rewrite only TOP-LEVEL Duration fields for M0 (Spark INTERVAL
+  // DAY-TIME surfaces as a top-level column). A Duration nested inside
+  // STRUCT/ARRAY/MAP is left untouched and apache-arrow@13 then throws
+  // `Unrecognized type: "Duration" (18)` when decoding the batch.
+  //
+  // This is a SHARED apache-arrow@13 limitation, NOT a SEA-specific gap:
+  // verified against a live warehouse, the Thrift backend throws the
+  // identical error for `array(INTERVAL '1' SECOND)` — so SEA matches Thrift
+  // here rather than diverging. Lifting it (recurse the rewrite into
+  // children) is deferred until there's a real-world nested-Duration payload
+  // to validate against, and would ideally land on both backends together.
   let hasDuration = false;
   const fieldsLength = fbSchema.fieldsLength();
   for (let i = 0; i < fieldsLength; i += 1) {

lib/sea/SeaOperationBackend.ts

@@ -43,10 +43,12 @@ import { OperationStatus, OperationState } from '../contracts/OperationStatus';
 import { ResultMetadata, ResultFormat } from '../contracts/ResultMetadata';
 import IClientContext from '../contracts/IClientContext';
 import Status from '../dto/Status';
+import HiveDriverError from '../errors/HiveDriverError';
 import ArrowResultConverter from '../result/ArrowResultConverter';
 import ResultSlicer from '../result/ResultSlicer';
 import SeaResultsProvider from './SeaResultsProvider';
-import { arrowSchemaToThriftSchema, decodeIpcSchema } from './SeaArrowIpc';
+import { arrowSchemaToThriftSchema, decodeIpcSchema, patchIpcBytes } from './SeaArrowIpc';
+import { decodeNapiKernelError } from './SeaErrorMapping';
 import { SeaStatement } from './SeaNativeLoader';
 import {
   SeaStatementHandle,
@@ -126,15 +128,36 @@ export default class SeaOperationBackend implements IOperationBackend {
   public async fetchChunk({
     limit,
     disableBuffering,
+    isClosed,
   }: {
     limit: number;
     disableBuffering?: boolean;
+    isClosed?: () => boolean;
   }): Promise<Array<object>> {
     // Cancel-mid-fetch propagation: if cancel() has flipped the
     // lifecycle flag, fail locally without a wire round-trip.
     failIfNotActive(this.lifecycle);
-    const slicer = await this.getResultSlicer();
-    return slicer.fetchNext({ limit, disableBuffering });
+    // Cooperative cancel (parity with ThriftOperationBackend): the facade
+    // supplies `isClosed` and re-checks `failIfClosed()` after we return, so
+    // bailing with `[]` at a yield point is the contract-correct way for a
+    // concurrent cancel()/close() to interrupt the fetch.
+    if (isClosed?.()) {
+      return [];
+    }
+    try {
+      const slicer = await this.getResultSlicer();
+      if (isClosed?.()) {
+        return [];
+      }
+      return await slicer.fetchNext({ limit, disableBuffering });
+    } catch (err) {
+      // The napi fetch contract leaves the stream in an unspecified state on
+      // 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);
+      throw decodeNapiKernelError(err);
+    }
   }
 
   public async hasMore(): Promise<boolean> {
@@ -153,9 +176,11 @@ export default class SeaOperationBackend implements IOperationBackend {
     }
     this.metadataPromise = (async () => {
       if (!this.statement.schema) {
-        throw new Error('SeaOperationBackend: statement.schema() is not available on this handle');
+        throw new HiveDriverError('SeaOperationBackend: statement.schema() is not available on this handle');
       }
-      const arrowSchemaIpc = await this.statement.schema();
+      // `schema()` is a synchronous napi getter (returns `ArrowSchema`, not a
+      // Promise) — no `await` needed.
+      const arrowSchemaIpc = this.statement.schema();
       const arrowSchema = decodeIpcSchema(arrowSchemaIpc.ipcBytes);
       // `ResultMetadata.schema` keeps the Thrift `TTableSchema` shape for
       // back-compat with the public `IOperation.getSchema()` surface.
@@ -166,8 +191,11 @@ export default class SeaOperationBackend implements IOperationBackend {
         // both flow through the same Arrow result converter.
         resultFormat: ResultFormat.ArrowBased,
         lz4Compressed: false,
-        // Carry the raw Arrow IPC schema bytes for ARROW_BASED consumers.
-        arrowSchema: arrowSchemaIpc.ipcBytes,
+        // Carry the *patched* Arrow IPC schema bytes (Duration → Int64 with the
+        // `duration_unit` marker) so an ARROW_BASED consumer decoding
+        // `arrowSchema` doesn't hit apache-arrow@13's `Unrecognized type
+        // "Duration" (18)`. Matches what the per-batch fetch path already does.
+        arrowSchema: patchIpcBytes(arrowSchemaIpc.ipcBytes),
         isStagingOperation: false,
       };
       this.metadata = meta;
@@ -187,10 +215,15 @@ export default class SeaOperationBackend implements IOperationBackend {
   public async status(_progress: boolean): Promise<OperationStatus> {
     // Synthesised — the kernel resolves `Statement::execute().await` before
     // it hands back a Statement handle, so by the time a SeaOperationBackend
-    // exists the statement is terminal. Report Cancelled/Closed if the
-    // lifecycle flag is set, else Succeeded. Returns the backend-neutral
-    // OperationStatus the IOperationBackend contract expects, so the
-    // DBSQLOperation facade switches on `state` identically across backends.
+    // exists the statement is terminal. Note there is intentionally no
+    // `Failed` arm: a failed execution rejects inside `executeStatement`
+    // (the kernel surfaces the error at submit), so a `Failed` statement
+    // never becomes a SeaOperationBackend — `status()` only ever observes
+    // Succeeded, or Cancelled/Closed from a client-side lifecycle call.
+    // Report Cancelled/Closed if the lifecycle flag is set, else Succeeded.
+    // Returns the backend-neutral OperationStatus the IOperationBackend
+    // contract expects, so the DBSQLOperation facade switches on `state`
+    // identically across backends.
     if (this.lifecycle.isCancelled) {
       return { state: OperationState.Cancelled, hasResultSet: true };
     }
@@ -227,7 +260,7 @@ export default class SeaOperationBackend implements IOperationBackend {
       return this.resultSlicer;
     }
     if (!this.statement.fetchNextBatch) {
-      throw new Error('SeaOperationBackend: statement.fetchNextBatch() is not available on this handle');
+      throw new HiveDriverError('SeaOperationBackend: statement.fetchNextBatch() is not available on this handle');
     }
     const metadata = await this.getResultMetadata();
     // The lifecycle subset has cancel/close only; fetch methods exist on

lib/sea/SeaOperationLifecycle.ts

@@ -261,10 +261,11 @@ export async function seaFinished(
  */
 export function failIfNotActive(state: SeaOperationLifecycleState): void {
   if (state.isCancelled) {
-    throw mapKernelErrorToJsError({
-      code: 'Cancelled',
-      message: 'The operation was cancelled.',
-    });
+    // Use the canonical `OperationStateError(Canceled)` (message "The operation
+    // was canceled by a client") rather than a custom string, so the message
+    // matches the Thrift path verbatim and the code branch stays consistent
+    // with the Closed case below.
+    throw new OperationStateError(OperationStateErrorCode.Canceled);
   }
   if (state.isClosed) {
     throw new OperationStateError(OperationStateErrorCode.Closed);

lib/sea/SeaResultsProvider.ts

@@ -22,7 +22,7 @@ import { decodeIpcBatch, patchIpcBytes } from './SeaArrowIpc';
  * d.ts) so the loader layer's loose `unknown` typing doesn't force
  * unsafe casts at every call site, and so unit tests can pass a stub.
  */
-export interface SeaStatementHandle {
+export interface SeaFetchHandle {
   fetchNextBatch(): Promise<{ ipcBytes: Buffer } | null>;
 }
 
@@ -48,7 +48,7 @@ export interface SeaStatementHandle {
  * `SeaArrowIpc.ts:decodeIpcBatch` for the cost rationale.
  */
 export default class SeaResultsProvider implements IResultsProvider<ArrowBatch> {
-  private readonly statement: SeaStatementHandle;
+  private readonly statement: SeaFetchHandle;
 
   // Prefetched next batch so `hasMore()` can be answered without an
   // extra round-trip. Set by `prime()` (lazy) and by `fetchNext`.
@@ -57,7 +57,7 @@ export default class SeaResultsProvider implements IResultsProvider<ArrowBatch>
   // Set once the kernel returns `null` from `fetchNextBatch()`.
   private exhausted = false;
 
-  constructor(statement: SeaStatementHandle) {
+  constructor(statement: SeaFetchHandle) {
     this.statement = statement;
   }
 
@@ -89,29 +89,30 @@ export default class SeaResultsProvider implements IResultsProvider<ArrowBatch>
   // to keep one batch buffered ahead so `hasMore` is accurate without
   // re-asking the kernel.
   private async prime(): Promise<void> {
-    if (this.exhausted || this.prefetched !== undefined) {
-      return;
+    // Loop rather than self-recurse: a long run of empty batches drains
+    // iteratively toward the null sentinel without building a deep promise
+    // chain.
+    while (!this.exhausted && this.prefetched === undefined) {
+      // eslint-disable-next-line no-await-in-loop
+      const next = await this.statement.fetchNextBatch();
+      if (next === null) {
+        this.exhausted = true;
+        return;
+      }
+      // 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`.
+      const ipcBytes = patchIpcBytes(next.ipcBytes);
+      const { rowCount } = decodeIpcBatch(ipcBytes, { alreadyPatched: true });
+      // 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.
+      if (rowCount > 0) {
+        this.prefetched = { batches: [ipcBytes], rowCount };
+      }
     }
-    const next = await this.statement.fetchNextBatch();
-    if (next === null) {
-      this.exhausted = true;
-      return;
-    }
-    // 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`.
-    const ipcBytes = patchIpcBytes(next.ipcBytes);
-    const { rowCount } = decodeIpcBatch(ipcBytes, { alreadyPatched: true });
-    if (rowCount === 0) {
-      // Skip empty batches — the converter handles them but pre-filtering
-      // here avoids one round-trip through the converter's prefetch loop.
-      // Re-prime to either find a non-empty batch or hit exhaustion.
-      await this.prime();
-      return;
-    }
-    this.prefetched = { batches: [ipcBytes], rowCount };
   }
 }

tests/e2e/sea/interval-duration-e2e.test.ts

@@ -0,0 +1,104 @@
+// 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.
+
+/* eslint-disable no-console */
+
+import { expect } from 'chai';
+import { DBSQLClient } from '../../../lib';
+import { ConnectionOptions } from '../../../lib/contracts/IDBSQLClient';
+import { InternalConnectionOptions } from '../../../lib/contracts/InternalConnectionOptions';
+
+// Exercises the SeaArrowIpcDurationFix rewriter against a REAL kernel-produced
+// Arrow Duration buffer (Spark INTERVAL DAY-TIME surfaces as Arrow Duration,
+// type id 18, which apache-arrow@13 can't decode). Without the rewriter this
+// query throws `Unrecognized type: "Duration" (18)`, so a passing fetch proves
+// the hand-rolled FlatBuffer schema re-encode + Int64 splice-back ran.
+//
+// On THIS layer (SEA execution + results, PR 2/3) the converter does not yet
+// consume the `duration_unit` marker, so the value surfaces as a raw Int64 —
+// asserted here. The Phase-1 formatter that turns it into the thrift string
+// "1 02:03:04.000000000" lands in PR 3/3 (#411), where SeaIntervalParity covers
+// the formatted output. Requires the pecotesting secrets; skips otherwise.
+
+const DURATION_QUERY = "SELECT INTERVAL '1 02:03:04' DAY TO SECOND AS dt";
+
+interface PecoSecrets {
+  host: string;
+  path: string;
+  token: string;
+}
+
+function readSecrets(): PecoSecrets | null {
+  const host = process.env.DATABRICKS_PECOTESTING_SERVER_HOSTNAME;
+  const path = process.env.DATABRICKS_PECOTESTING_HTTP_PATH;
+  const token = process.env.DATABRICKS_PECOTESTING_TOKEN_PERSONAL;
+  if (!host || !path || !token) return null;
+  return { host, path, token };
+}
+
+async function fetchOneRow(useSEA: boolean, secrets: PecoSecrets): Promise<Record<string, unknown>> {
+  const client = new DBSQLClient();
+  await client.connect({
+    host: secrets.host,
+    path: secrets.path,
+    token: secrets.token,
+    useSEA,
+  } as ConnectionOptions & InternalConnectionOptions);
+  try {
+    const session = await client.openSession();
+    try {
+      const operation = await session.executeStatement(DURATION_QUERY);
+      try {
+        const rows = (await operation.fetchAll()) as Array<Record<string, unknown>>;
+        return rows[0];
+      } finally {
+        await operation.close();
+      }
+    } finally {
+      await session.close();
+    }
+  } finally {
+    await client.close();
+  }
+}
+
+describe('SEA INTERVAL DAY-TIME (Arrow Duration rewriter) end-to-end', function suite() {
+  this.timeout(120_000);
+
+  const secrets = readSecrets();
+
+  before(function gate() {
+    if (!secrets) {
+      // eslint-disable-next-line no-invalid-this
+      this.skip();
+    }
+  });
+
+  it('decodes a real Arrow Duration column without the "Duration (18)" crash (rewriter ran)', async () => {
+    const row = await fetchOneRow(true, secrets as PecoSecrets);
+    // A returned row at all means the schema rewrite + Int64 splice-back ran;
+    // an un-rewritten Duration would have thrown during RecordBatchReader.from.
+    expect(row).to.have.property('dt');
+    expect(row.dt, 'INTERVAL DAY-TIME value should be present').to.not.equal(undefined);
+  });
+
+  it('surfaces the value as a raw Int64 on this layer (formatter is PR 3/3)', async () => {
+    const row = await fetchOneRow(true, secrets as PecoSecrets);
+    // Documents the M0/2-of-3 contract: the rewriter makes the column
+    // *decodable* but the duration_unit formatter is not wired here yet, so the
+    // value is the raw integer microsecond/nanosecond count, not the thrift
+    // "1 02:03:04.000000000" string. (#411 flips this to the formatted string.)
+    expect(['number', 'bigint']).to.include(typeof row.dt);
+  });
+});