[SEA-NodeJS] Extend eager-execute parity to the async (runAsync) path

The prior commit fixed fire-and-forget DDL/DML on the default sync path. The
opt-in async path (`runAsync: true`) had the same defect from a different cause:
`submitStatement` starts the statement server-side, but `asyncStatement.close()`
CANCELS a still-running statement — so `executeStatement(..., {runAsync:true})`
followed by `close()` with no fetch aborted the statement before it committed.

`SeaOperationBackend` now wraps the async handle in the same composite-close
shape as the sync path: unless the caller explicitly `cancel()`ed, `close()`
drives the result to terminal (`getFetchHandle()` → the kernel `awaitResult()`)
so the side effect lands, then closes. The metadata path (already-terminal
statement) still closes directly.

Verified: 249 SEA unit tests pass (added async drive-on-close and
close-after-cancel coverage); e2e confirms fire-and-forget CREATE/INSERT via
runAsync:true now persists on SEA, with no change to the cancel or normal-read
paths.

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

Author: msrathore-db

lib/sea/SeaOperationBackend.ts

@@ -215,33 +215,50 @@ export default class SeaOperationBackend implements IOperationBackend {
     this.asyncStatement = asyncStatement;
     this.cancellableExecution = cancellableExecution;
     this.blockingStatement = statement;
-    // Lifecycle surface. The async/metadata handles expose both cancel/close.
-    // The sync-execute path uses a composite: `cancel()` always routes to the
-    // cancellable execution (lock-free, interrupts a running `result()`
-    // mid-compute and is a no-op once terminal); `close()` closes the resolved
-    // terminal statement once `result()` produced it, OR — if `result()` is
-    // still in flight — proactively cancels the running execution so the server
-    // stops computing immediately rather than running on until the kernel's
-    // drop-guard fires whenever this handle is eventually GC'd.
-    this.lifecycleHandle = cancellableExecution
-      ? {
-          cancel: () => cancellableExecution.cancel(),
-          close: async () => {
-            // Eager-execute parity with Thrift and Python `use_kernel`: a
-            // statement run via `executeStatement(...).close()` with NO
-            // intervening fetch must still take effect server-side (DDL/DML side
-            // effects). If the execution was never driven to a result and the
-            // caller did NOT explicitly cancel, drive it to completion here
-            // rather than cancelling an un-run statement. Errors are swallowed —
-            // `close()` is best-effort cleanup and must not throw; the read path
-            // (fetch / finished) still surfaces execution errors.
-            if (!this.lifecycle.isCancelled && this.blockingStatement === undefined) {
-              await this.getFetchHandle().catch(() => {});
-            }
-            return this.blockingStatement ? this.blockingStatement.close() : cancellableExecution.cancel();
-          },
-        }
-      : ((asyncStatement ?? statement) as SeaStatementHandle);
+    // Lifecycle surface. Both query paths use a composite handle so that
+    // `close()` honours eager-execute parity with Thrift / Python `use_kernel`:
+    // a statement run via `executeStatement(...).close()` with NO intervening
+    // fetch must still take effect server-side (DDL/DML side effects). In both
+    // cases, unless the caller explicitly `cancel()`ed, `close()` drives the
+    // statement to a terminal result before closing rather than cancelling an
+    // un-run statement. Errors are swallowed — `close()` is best-effort cleanup
+    // and must not throw; the read path (fetch / finished) still surfaces them.
+    // The metadata path's statement is already terminal, so it closes directly.
+    if (cancellableExecution) {
+      // Sync (`runAsync: false`): `cancel()` routes to the cancellable execution
+      // (lock-free, interrupts a running `result()` mid-compute, no-op once
+      // terminal). `close()` drives `result()` to the terminal statement (set on
+      // `blockingStatement`) and closes that; if `result()` already resolved it
+      // closes directly; if cancelled it cancels.
+      this.lifecycleHandle = {
+        cancel: () => cancellableExecution.cancel(),
+        close: async () => {
+          if (!this.lifecycle.isCancelled && this.blockingStatement === undefined) {
+            await this.getFetchHandle().catch(() => {});
+          }
+          return this.blockingStatement ? this.blockingStatement.close() : cancellableExecution.cancel();
+        },
+      };
+    } else if (asyncStatement) {
+      // Async (`runAsync: true`): `submitStatement` already started the statement
+      // server-side, but `asyncStatement.close()` CANCELS a still-running
+      // statement — so `executeStatement(...).close()` with no fetch would abort
+      // a DDL/DML before it commits. Mirror the sync path: unless explicitly
+      // cancelled, drive the result to terminal (`getFetchHandle()` → the kernel
+      // `awaitResult()`) so the side effect lands, then close.
+      this.lifecycleHandle = {
+        cancel: () => asyncStatement.cancel(),
+        close: async () => {
+          if (!this.lifecycle.isCancelled) {
+            await this.getFetchHandle().catch(() => {});
+          }
+          return asyncStatement.close();
+        },
+      };
+    } else {
+      // Metadata path: the kernel statement is already terminal — close directly.
+      this.lifecycleHandle = statement as SeaStatementHandle;
+    }
     this.context = context;
     this._id =
       id ?? asyncStatement?.statementId ?? statement?.statementId ?? cancellableExecution?.statementId ?? uuidv4();

tests/unit/sea/execution.test.ts

@@ -88,6 +88,8 @@ class FakeAsyncStatement {
 
   public statusCalls = 0;
 
+  public awaitCalls = 0;
+
   public awaitResultError: Error | null = null;
 
   // Successive status() returns drain this queue; the last value sticks.
@@ -108,6 +110,7 @@ class FakeAsyncStatement {
   }
 
   public async awaitResult(): Promise<FakeNativeStatement> {
+    this.awaitCalls += 1;
     if (this.awaitResultError) {
       throw this.awaitResultError;
     }
@@ -1011,10 +1014,27 @@ describe('SeaOperationBackend — async (submitStatement) path', () => {
     expect(thrown).to.be.an('error');
   });
 
-  it('close() forwards to the async statement', async () => {
+  it('close() drives an un-fetched async statement to completion, then closes it (eager-execute parity)', async () => {
+    // submitStatement starts the statement server-side, but asyncStatement.close()
+    // CANCELS a still-running statement — so executeStatement(runAsync:true).close()
+    // with no fetch would abort a DDL/DML before it commits. close() must first
+    // drive the result to terminal (awaitResult), then close.
+    const stmt = new FakeAsyncStatement();
+    const op = makeAsyncOp(stmt);
+    await op.close();
+    expect(stmt.awaitCalls, 'close() drives awaitResult() to terminal').to.equal(1);
+    expect(stmt.closed).to.equal(true);
+    expect(stmt.cancelled).to.equal(false);
+  });
+
+  it('close() after an explicit cancel() does NOT drive the async statement', async () => {
     const stmt = new FakeAsyncStatement();
     const op = makeAsyncOp(stmt);
+    await op.cancel();
     await op.close();
+    expect(stmt.cancelled).to.equal(true);
+    // Already being torn down — close() must not await a fresh result drive.
+    expect(stmt.awaitCalls).to.equal(0);
     expect(stmt.closed).to.equal(true);
   });
 });