From 0970af698244e5eebe22b26e3155ecaff18f0d1b Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Wed, 11 Nov 2026 23:15:17 +0000 Subject: [PATCH] fix: treat a failed close of a finished operation as best effort Once the backend has reported the statement finished, a failed close request (e.g. a 404 on CloseOperation / CloseStatement) is logged as a warning, emitted as error telemetry and reported in the returned status, and the operation is closed locally. Before the statement finishes, close failures still throw, since closing is what stops a running statement. Fixes #480 Signed-off-by: Vu Anh Phung Co-authored-by: Isaac --- CHANGELOG.md | 1 + lib/DBSQLOperation.ts | 28 +++- tests/unit/DBSQLOperation.test.ts | 121 +++++++++++++++++- tests/unit/DBSQLSession.test.ts | 17 +++ tests/unit/kernel/operation-lifecycle.test.ts | 59 +++++++++ 5 files changed, 223 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 77d44b38..9bc027cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ - CloudFetch retry-exhaustion errors now identify the download and the last failure: the `RetryError` message reads `CloudFetch result download failed: (last error: )`. `` is `HTTP : ` (or `HTTP ` when the status text is empty) for HTTP failures, and `network error `, `network error ()` (e.g. `aborted`, when there is no `code`) or a bare `network error` for network failures (the raw network-error message, which can contain the presigned URL, is not copied). If no cause can be determined, the ` (last error: …)` suffix is omitted. The `RetryError` instance, `errorCode` and `payload` (including the node-fetch `Request` / `Response`) are unchanged. (#471) - Report untyped NULL columns as STRING in kernel result metadata, matching Thrift. (#470) +- `IOperation.close()` no longer throws when the close request fails after the operation has finished (for example, once its results were read): the failure is logged as a warning and reported in the returned status's `info`, and the operation is still closed locally. Closing an operation that hasn't finished still throws on failure. (#480) - Reject non-scalar query parameter values at bind time with a `ParameterError` on both the Thrift and kernel backends, instead of stringifying them (which could leave a Thrift operation stuck before FINISHED). Newly rejected on the Thrift backend: arrays, plain objects, boxed primitives (e.g. `new String('x')`, `new Number(1)`), functions, symbols, and a `DBSQLParameter` that wraps another `DBSQLParameter`. Primitives, `Date`, `Int64`, and `DBSQLParameter` are still accepted. - `DBSQLLogger`: a `filepath` naming an existing directory now writes to `databricks-sql-nodejs.log` inside it, instead of silently producing no logs. (#565) - `DBSQLLogger`: if the log file can't be opened (including when its directory can't be created, which used to throw from the constructor), the logger now logs to the console only, leaves `transports.file` undefined, and emits a process warning with code `DBSQL_LOG_FILE_UNAVAILABLE`. Previously an unwritable file could silently stall all logging, including the console. (#565) diff --git a/lib/DBSQLOperation.ts b/lib/DBSQLOperation.ts index 1ec1922c..42a6ecd4 100644 --- a/lib/DBSQLOperation.ts +++ b/lib/DBSQLOperation.ts @@ -55,6 +55,10 @@ export default class DBSQLOperation implements IOperation { private cancelled: boolean = false; + // Set once the backend reports the statement finished successfully, after + // which a failed close cannot leave it running server-side. + private executionFinished: boolean = false; + public readonly _data?: unknown; private metadata?: ResultMetadata; @@ -212,7 +216,8 @@ export default class DBSQLOperation implements IOperation { } /** - * Closes operation + * Closes operation. Once the operation has finished, a failed close request + * is logged and reported in the returned status's `info` instead of thrown. * @throws {StatusError} */ public async close(): Promise { @@ -223,7 +228,25 @@ export default class DBSQLOperation implements IOperation { if (this.closed || this.cancelled) { return Status.success(); } - const result = await this.backend.close(); + let result: Status; + try { + result = await this.backend.close(); + } catch (error: any) { + // Until the statement finished, closing it is what stops it server-side, + // so the failure is the caller's to handle (and the close can be retried). + if (!this.executionFinished) { + throw error; + } + // Afterwards the close only releases server-side results, which closing + // the session also does: report the failure and finish the teardown. + // `StatusError` doesn't extend `Error`, so read `message` structurally. + const message = typeof error?.message === 'string' ? error.message : String(error); + this.context + .getLogger() + .log(LogLevel.warn, `Failed to close finished operation ${this.id}; it is closed locally. Cause: ${message}`); + this.emitErrorEvent(error); + result = Status.success([`Failed to close operation: ${message}`]); + } this.closed = true; // Emit statement.complete telemetry event @@ -367,6 +390,7 @@ export default class DBSQLOperation implements IOperation { }; try { await this.backend.waitUntilReady(backendOptions); + this.executionFinished = true; } catch (err) { // Reflect terminal states back into facade flags so subsequent calls // short-circuit via failIfClosed(). diff --git a/tests/unit/DBSQLOperation.test.ts b/tests/unit/DBSQLOperation.test.ts index f9195105..9cfa57a1 100644 --- a/tests/unit/DBSQLOperation.test.ts +++ b/tests/unit/DBSQLOperation.test.ts @@ -11,7 +11,8 @@ import { TTypeId, } from '../../thrift/TCLIService_types'; import StatusError from '../../lib/errors/StatusError'; -import OperationStateError from '../../lib/errors/OperationStateError'; +import OperationStateError, { OperationStateErrorCode } from '../../lib/errors/OperationStateError'; +import { LogLevel } from '../../lib/contracts/IDBSQLLogger'; import HiveDriverError from '../../lib/errors/HiveDriverError'; import JsonResultHandler from '../../lib/result/JsonResultHandler'; import ArrowResultConverter from '../../lib/result/ArrowResultConverter'; @@ -416,6 +417,108 @@ describe('DBSQLOperation', () => { await expectFailure(() => operation.finished()); await expectFailure(() => operation.getSchema()); }); + + describe('after the operation finished', () => { + // How the HTTP layer reports the 404 a teardown CloseOperation can get. + const closeFailure = () => new Error('Received a response with a bad HTTP status: 404'); + + function finishedOperationWithFailingClose(failure: () => Promise) { + const context = new ClientContextStub(); + const log = sinon.spy(context.logger, 'log'); + const closeOperation = sinon.stub(context.driver, 'closeOperation').callsFake(failure as any); + // One page of rows, then an empty last page. + sinon + .stub(context.driver, 'fetchResults') + .onFirstCall() + .resolves(context.driver.fetchResultsResp) + .resolves({ + status: { statusCode: TStatusCode.SUCCESS_STATUS }, + hasMoreRows: false, + results: { startRowOffset: new Int64(3), rows: [], columns: [] }, + }); + const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); + const onClose = sinon.stub(); + operation.onClose = onClose; + return { context, log, closeOperation, operation, onClose }; + } + + it('should complete teardown when CloseOperation fails after results were consumed', async () => { + const { log, closeOperation, operation, onClose } = finishedOperationWithFailingClose(() => + Promise.reject(closeFailure()), + ); + + expect(await operation.fetchAll()).to.have.lengthOf(3); + + const status = await operation.close(); + + expect(closeOperation.calledOnce).to.be.true; + expect(status.isSuccess).to.be.true; + expect(status.info.join()).to.contain('404'); + expect(operation['closed']).to.be.true; + expect(onClose.calledOnce).to.be.true; + expect(log.calledWith(LogLevel.warn, sinon.match(operation.id).and(sinon.match('404')))).to.be.true; + + try { + await operation.fetchAll(); + expect.fail('It should throw an OperationStateError'); + } catch (e) { + if (e instanceof AssertionError) { + throw e; + } + expect(e).to.be.instanceOf(OperationStateError); + expect((e as OperationStateError).errorCode).to.equal(OperationStateErrorCode.Closed); + } + + await operation.close(); + expect(closeOperation.calledOnce).to.be.true; + }); + + it('should complete teardown when CloseOperation returns an error status after finished()', async () => { + const { operation, onClose } = finishedOperationWithFailingClose(async () => ({ + status: { statusCode: TStatusCode.ERROR_STATUS, errorMessage: 'Invalid OperationHandle' }, + })); + + await operation.finished(); + const status = await operation.close(); + + expect(status.isSuccess).to.be.true; + expect(status.info.join()).to.contain('Invalid OperationHandle'); + expect(operation['closed']).to.be.true; + expect(onClose.calledOnce).to.be.true; + }); + + it('should not throw from an autoClose iterator when CloseOperation fails', async () => { + const { closeOperation, operation, onClose } = finishedOperationWithFailingClose(() => + Promise.reject(closeFailure()), + ); + + const rows: Array = []; + for await (const row of operation.iterateRows({ autoClose: true })) { + rows.push(row); + } + + expect(rows).to.have.lengthOf(3); + expect(closeOperation.calledOnce).to.be.true; + expect(operation['closed']).to.be.true; + expect(onClose.calledOnce).to.be.true; + }); + + it('should still rethrow when the operation was not observed to finish', async () => { + const { closeOperation, operation, onClose } = finishedOperationWithFailingClose(() => + Promise.reject(closeFailure()), + ); + + await expectFailure(() => operation.close()); + expect(operation['closed']).to.be.false; + expect(onClose.called).to.be.false; + + // Still open, so the close can be retried. + closeOperation.resolves({ status: { statusCode: TStatusCode.SUCCESS_STATUS } }); + await operation.close(); + expect(operation['closed']).to.be.true; + expect(onClose.calledOnce).to.be.true; + }); + }); }); describe('finished', () => { @@ -1211,6 +1314,22 @@ describe('DBSQLOperation', () => { expect(emitError.calledOnce).to.be.true; }); + it('emits ERROR and statement-complete telemetry when a finished operation fails to close', async () => { + const { context, emitError } = makeContextWithEmitter(); + const completeStatement = sinon.stub(); + context.telemetryAggregator = { completeStatement } as any; + context.driver.closeOperationResp.status.statusCode = TStatusCode.ERROR_STATUS; + const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); + + await operation.finished(); + await operation.close(); + + expect(emitError.calledOnce).to.be.true; + expect(emitError.firstCall.args[0].errorName).to.match(/Status\s?Error/); + expect((context.telemetryEmitter!.emitStatementComplete as sinon.SinonStub).calledOnce).to.be.true; + expect(completeStatement.calledOnceWith(operation.id)).to.be.true; + }); + it('emits ERROR telemetry when getMetadata rejects', async () => { const { context, emitError } = makeContextWithEmitter(); // Force the driver to fail metadata fetch diff --git a/tests/unit/DBSQLSession.test.ts b/tests/unit/DBSQLSession.test.ts index ce782207..22e76e4f 100644 --- a/tests/unit/DBSQLSession.test.ts +++ b/tests/unit/DBSQLSession.test.ts @@ -724,6 +724,23 @@ describe('DBSQLSession', () => { expect(session['operations']['items'].size).to.eq(0); }); + it('should close the session when a finished operation fails to close', async () => { + const context = new ClientContextStub(); + const closeSession = sinon.spy(context.driver, 'closeSession'); + sinon + .stub(context.driver, 'closeOperation') + .rejects(new Error('Received a response with a bad HTTP status: 404')); + const session = createSessionForTest({ handle: sessionHandleStub, context }); + const operation = await session.executeStatement('SELECT * FROM table'); + await operation.finished(); + + await session.close(); + + expect(closeSession.calledOnce).to.be.true; + expect(session['isOpen']).to.be.false; + expect(session['operations']['items'].size).to.eq(0); + }); + it('should reject all methods once closed', async () => { const session = createSessionForTest({ handle: sessionHandleStub, context: new ClientContextStub() }); await session.close(); diff --git a/tests/unit/kernel/operation-lifecycle.test.ts b/tests/unit/kernel/operation-lifecycle.test.ts index 839a75d3..ac309816 100644 --- a/tests/unit/kernel/operation-lifecycle.test.ts +++ b/tests/unit/kernel/operation-lifecycle.test.ts @@ -25,6 +25,9 @@ import { expect } from 'chai'; import sinon from 'sinon'; +import { tableFromArrays, tableToIPC } from 'apache-arrow'; +import DBSQLOperation from '../../../lib/DBSQLOperation'; +import ClientContextStub from '../.stubs/ClientContextStub'; import { OperationStatus, OperationState } from '../../../lib/contracts/OperationStatus'; import IClientContext from '../../../lib/contracts/IClientContext'; import IDBSQLLogger, { LogLevel } from '../../../lib/contracts/IDBSQLLogger'; @@ -436,3 +439,59 @@ describe('KernelOperationBackend (lifecycle integration)', () => { expect(op.hasResultSet()).to.equal(true); }); }); + +describe('DBSQLOperation over KernelOperationBackend — close after the result was drained', () => { + function drainedOperation(close: () => Promise) { + const ipcBytes = Buffer.from(tableToIPC(tableFromArrays({ x: Int32Array.from([1, 2]) }), 'stream')); + const batches = [ipcBytes]; + const statement = { + cancel: sinon.stub().resolves(), + close: sinon.stub().callsFake(close), + schema: () => ({ ipcBytes }), + fetchNextBatch: async () => (batches.length > 0 ? { ipcBytes: batches.shift()! } : null), + }; + const context = new ClientContextStub(); + const log = sinon.spy(context.logger, 'log'); + const backend = new KernelOperationBackend({ statement, context, id: 'kernel-op' }); + const operation = new DBSQLOperation({ backend, context }); + const onClose = sinon.stub(); + operation.onClose = onClose; + return { statement, log, operation, onClose }; + } + + it('completes teardown when the CloseStatement RPC fails', async () => { + const { statement, log, operation, onClose } = drainedOperation(async () => { + throw new Error(`__databricks_error__:${JSON.stringify({ code: 'NotFound', message: 'HTTP 404 Not Found' })}`); + }); + + expect(await operation.fetchAll()).to.deep.equal([{ x: 1 }, { x: 2 }]); + const status = await operation.close(); + + expect(statement.close.calledOnce).to.equal(true); + expect(status.isSuccess).to.equal(true); + expect(onClose.calledOnce).to.equal(true); + expect(log.calledWith(LogLevel.warn, sinon.match('kernel-op').and(sinon.match('404')))).to.equal(true); + + let thrown: unknown; + try { + await operation.fetchAll(); + } catch (err) { + thrown = err; + } + expect(thrown).to.be.instanceOf(OperationStateError); + expect((thrown as OperationStateError).errorCode).to.equal(OperationStateErrorCode.Closed); + }); + + it('closes normally when the RPC succeeds', async () => { + const { statement, log, operation, onClose } = drainedOperation(async () => undefined); + + await operation.fetchAll(); + const status = await operation.close(); + + expect(statement.close.calledOnce).to.equal(true); + expect(status.isSuccess).to.equal(true); + expect(status.info).to.deep.equal([]); + expect(onClose.calledOnce).to.equal(true); + expect(log.calledWith(LogLevel.warn)).to.equal(false); + }); +});