From 325e9c485bc45fc170ade4ad9120d421a4be8c6a Mon Sep 17 00:00:00 2001 From: Bob den Os Date: Thu, 5 Feb 2026 12:21:47 +0100 Subject: [PATCH 1/2] cache prepare statement for re use --- hana/lib/HANAService.js | 92 ++++++++++++++++++++++----------- hana/lib/drivers/base.js | 38 +++++++++++--- hana/lib/drivers/hana-client.js | 3 ++ hana/lib/drivers/hdb.js | 2 + hana/test/run.test.js | 2 +- hana/test/spatial.test.js | 1 - test/compliance/DELETE.test.js | 1 - test/compliance/UPDATE.test.js | 3 +- 8 files changed, 101 insertions(+), 41 deletions(-) diff --git a/hana/lib/HANAService.js b/hana/lib/HANAService.js index 549d6e621..960fb413c 100644 --- a/hana/lib/HANAService.js +++ b/hana/lib/HANAService.js @@ -73,8 +73,8 @@ class HANAService extends SQLService { try { await require('@sap/cds-mtxs/lib').xt.serviceManager.get(tenant, { disableCache: true, invalidCredentials: credentials, retryUntil: deadline }) } catch (smErr) { - smErr.cause = err - throw new Error(`Failed connecting to pool - could not get valid credentials from Service Manager`, { cause: smErr }) + smErr.cause = err + throw new Error(`Failed connecting to pool - could not get valid credentials from Service Manager`, { cause: smErr }) } if (Date.now() < deadline) return create(tenant, start) else throw new Error(`Pool exceeded for '${tenant}' within ${acquireTimeoutMillis}ms`, { cause: err }) @@ -167,7 +167,11 @@ class HANAService extends SQLService { let rows if (values?.length || blobs.length > 0 || isStream) { const ps = await this.prepare(sqlScript, blobs.length) - rows = this.ensureDBC() && await ps[isStream ? 'stream' : 'all'](values || [], isOne, objectMode) + try { + rows = this.ensureDBC() && await ps[isStream ? 'stream' : 'all'](values || [], isOne, objectMode) + } finally { + ps.drop() + } } else { rows = await this.exec(sqlScript) } @@ -203,13 +207,34 @@ class HANAService extends SQLService { const { sql, entries, cqn } = this.cqn2sql(query, data) if (!sql) return // Do nothing when there is nothing to be done const ps = await this.prepare(sql) - // HANA driver supports batch execution - const results = await (entries - ? this.server.major <= 2 - ? entries.reduce((l, c) => l.then(() => this.ensureDBC() && ps.run(c)), Promise.resolve(0)) - : entries.length > 1 ? this.ensureDBC() && await ps.runBatch(entries) : this.ensureDBC() && await ps.run(entries[0]) - : this.ensureDBC() && ps.run()) - return new this.class.InsertResults(cqn, results) + try { + // HANA driver supports batch execution + const results = await (entries + ? this.server.major <= 2 + ? entries.reduce((l, c) => l.then(() => this.ensureDBC() && ps.run(c)), Promise.resolve(0)) + : entries.length > 1 ? this.ensureDBC() && await ps.runBatch(entries) : this.ensureDBC() && await ps.run(entries[0]) + : this.ensureDBC() && ps.run()) + return new this.class.InsertResults(cqn, results) + } finally { + ps.drop() + } + } + + async onUPSERT({ query, data }) { + const { sql, entries, cqn } = this.cqn2sql(query, data) + if (!sql) return // Do nothing when there is nothing to be done + const ps = await this.prepare(sql) + try { + // HANA driver supports batch execution + const results = await (entries + ? this.server.major <= 2 + ? entries.reduce((l, c) => l.then(() => this.ensureDBC() && ps.run(c)), Promise.resolve(0)) + : entries.length > 1 ? this.ensureDBC() && await ps.runBatch(entries) : this.ensureDBC() && await ps.run(entries[0]) + : this.ensureDBC() && ps.run()) + return results.changes ?? results + } finally { + ps.drop() + } } async onNOTFOUND(req, next) { @@ -307,11 +332,8 @@ class HANAService extends SQLService { } // prepare and exec are both implemented inside the drivers - prepare(sql, hasBlobs) { - const stmt = this.ensureDBC().prepare(sql, hasBlobs) - // we store the statements, to release them on commit/rollback all at once - this.dbc.statements.push(stmt) - return stmt + async prepare(sql, hasBlobs) { + return this.ensureDBC().prepare(sql, hasBlobs) } exec(sql) { @@ -1278,8 +1300,9 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON E async onSIMPLE({ query, data, event }) { const { sql, values } = this.cqn2sql(query, data) + let ps try { - let ps = await this.prepare(sql) + ps = await this.prepare(sql) return (this.ensureDBC() && await ps.run(values)).changes } catch (err) { // Allow drop to fail when the view or table does not exist @@ -1287,6 +1310,8 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON E return } throw err + } finally { + ps?.drop() } } @@ -1311,8 +1336,26 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON E const isAsync = /\sASYNC\s*$/.test(query) const outParameters = isAsync ? [{ PARAMETER_NAME: 'ASYNC_CALL_ID' }] : await this._getProcedureMetadata(name, schema) const ps = await this.prepare(query) - const ret = this.ensureDBC() && await ps.proc(data, outParameters) - return isAsync ? ret.ASYNC_CALL_ID[0] : ret + try { + const ret = this.ensureDBC() && await ps.proc(data, outParameters) + return isAsync ? ret.ASYNC_CALL_ID[0] : ret + } finally { + ps.drop() + } + } + + async _onPlainSQL({ query, data }, next) { + if (typeof query === 'string') { + DEBUG?.(query, data) + const ps = await this.prepare(query) + try { + const exec = this.hasResults(query) ? d => ps.all(d) : d => ps.run(d) + if (Array.isArray(data) && Array.isArray(data[0])) return await Promise.all(data.map(exec)) + else return exec(data) + } finally { + ps.drop() + } + } else return next() } async onPlainSQL(req, next) { @@ -1320,7 +1363,7 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON E if (/ IF EXISTS /i.test(req.query)) { req.query = req.query.replace(/ IF EXISTS/gi, '') try { - return await super.onPlainSQL(req, next) + return await this._onPlainSQL(req, next) } catch (err) { if (/(^|')DROP /i.test(req.query) && (err.code === 259 || err.code === 321)) { return @@ -1332,30 +1375,21 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON E const proc = this._getProcedureNameAndSchema(req.query) if (proc && proc.name) return this.onCall(req, proc.name, proc.schema) - return super.onPlainSQL(req, next) + return this._onPlainSQL(req, next) } onBEGIN() { DEBUG?.('BEGIN') - if (this.dbc) this.dbc.statements = [] return this.dbc?.begin() } onCOMMIT() { DEBUG?.('COMMIT') - this.dbc?.statements?.forEach(stmt => stmt - .then(stmt => stmt.drop()) - .catch(() => { }) - ) return this.dbc?.commit() } onROLLBACK() { DEBUG?.('ROLLBACK') - this.dbc?.statements?.forEach(stmt => stmt - .then(stmt => stmt.drop()) - .catch(() => { }) - ) return this.dbc?.rollback() } diff --git a/hana/lib/drivers/base.js b/hana/lib/drivers/base.js index 428986cf2..f945d1bd1 100644 --- a/hana/lib/drivers/base.js +++ b/hana/lib/drivers/base.js @@ -9,6 +9,17 @@ class HANADriver { constructor(creds) { this._creds = creds this.connected = false + this.statements = {} + } + + _prepare(sql) { + return module.exports.prom( + this._native, + 'prepare', + )(sql).then(stmt => { + stmt._parentConnection = this._native + return stmt + }) } /** @@ -17,13 +28,17 @@ class HANADriver { * @returns {import('@cap-js/db-service/lib/SQLService').PreparedStatement} */ async prepare(sql) { - const prep = module.exports.prom( - this._native, - 'prepare', - )(sql).then(stmt => { - stmt._parentConnection = this._native - return stmt + this.statements[sql] ??= this._prepare(sql) + + const prep = this.statements[sql].ready ?? this.statements[sql] + const release = {} // TODO: for node@22 use Promise.withResolvers() + release.promise = new Promise((resolve, reject) => { + release.resolve = resolve + release.reject = reject }) + release.promise.catch(() => { }) + this.statements[sql].ready = release.promise + return { _prep: prep, run: async params => { @@ -46,9 +61,16 @@ class HANADriver { const stmt = await prep return module.exports.prom(stmt, 'exec')(params) }, + detach: async () => { + // Check whether something is waiting for the prepared statement already + if (this.statements[sql].ready === release.promise) { + this.statements[sql] = undefined + return + } + this._prepare(sql).then(release.resolve, release.reject) + }, drop: async () => { - const stmt = await prep - return stmt.drop() + prep.then(release.resolve, release.reject) } } } diff --git a/hana/lib/drivers/hana-client.js b/hana/lib/drivers/hana-client.js index 16db4b6a9..bafc31315 100644 --- a/hana/lib/drivers/hana-client.js +++ b/hana/lib/drivers/hana-client.js @@ -66,6 +66,7 @@ class HANAClientDriver extends driver { // It would be required to allow using getDate() on previous rows if (hasBlobs) { ret.all = async (values) => { + ret.detach() const stmt = await ret._prep // Create result set const reset = async function () { @@ -140,6 +141,7 @@ class HANAClientDriver extends driver { ret.stream = async (values, one, objectMode) => { const stmt = await ret._prep + ret.detach() values = Array.isArray(values) ? values : [] // Uses the native exec method instead of executeQuery to initialize a full stream // As executeQuery does not request the whole result set at once @@ -169,6 +171,7 @@ class HANAClientDriver extends driver { } async validate() { + if (Object.keys(this.statements).length > 10_000) return false return this._native.state() === 'connected' } diff --git a/hana/lib/drivers/hdb.js b/hana/lib/drivers/hdb.js index 514efb9b5..184bb9664 100644 --- a/hana/lib/drivers/hdb.js +++ b/hana/lib/drivers/hdb.js @@ -60,6 +60,7 @@ class HDBDriver extends driver { } async validate() { + if (Object.keys(this.statements).length > 10_000) return false return this._native.readyState === 'connected' } @@ -126,6 +127,7 @@ class HDBDriver extends driver { ret.stream = async (values, one, objectMode) => { const stmt = await ret._prep + ret.detach() const rs = await prom(stmt, 'execute')(values || []) return rsIterator(rs, one, objectMode) } diff --git a/hana/test/run.test.js b/hana/test/run.test.js index c634145aa..7042703a9 100644 --- a/hana/test/run.test.js +++ b/hana/test/run.test.js @@ -125,7 +125,7 @@ BEGIN SELECT ERROR_CODE, ERROR_TEXT, :dur AS WAITED_SECONDS FROM M_PROCEDURE_ASYNC_EXECUTIONS WHERE ASYNC_CALL_ID = :ID; END;`, [ASYNC_CALL_ID]) // Ensure that the procedure succeeded - expect(status.changes[1][0].ERROR_CODE).to.eq(0) + expect(status.changes[0].ERROR_CODE ?? status.changes[1][0].ERROR_CODE).to.eq(0) throw new Error('ROLLBACK') }).catch((err) => { if (err.message !== 'ROLLBACK') throw err }) diff --git a/hana/test/spatial.test.js b/hana/test/spatial.test.js index c37e49785..f6bac1d1d 100644 --- a/hana/test/spatial.test.js +++ b/hana/test/spatial.test.js @@ -3,7 +3,6 @@ const cds = require('../../test/cds.js') describe('Spatial Types', () => { const { data, expect } = cds.test(__dirname + '/../../test/compliance/resources') data.autoIsolation(true) - data.autoReset() test('point', async () => { const { HANA_ST } = cds.entities('edge.hana.literals') diff --git a/test/compliance/DELETE.test.js b/test/compliance/DELETE.test.js index f5ce0cd46..cded2f155 100644 --- a/test/compliance/DELETE.test.js +++ b/test/compliance/DELETE.test.js @@ -8,7 +8,6 @@ const ChildPWithWhere = 'complex.ChildPWithWhere' describe('DELETE', () => { const { data, expect } = cds.test(__dirname + '/resources') data.autoIsolation(true) - data.autoReset() describe('from', () => { describe('deep', () => { diff --git a/test/compliance/UPDATE.test.js b/test/compliance/UPDATE.test.js index 8c90cb110..a674a051b 100644 --- a/test/compliance/UPDATE.test.js +++ b/test/compliance/UPDATE.test.js @@ -5,7 +5,6 @@ const BooksUnique = 'complex.uniques.Books' describe('UPDATE', () => { const { data, expect } = cds.test(__dirname + '/resources') data.autoIsolation(true) - data.autoReset() describe('entity', () => { test('string', async () => { @@ -70,6 +69,7 @@ describe('UPDATE', () => { test('xpr', async () => { const { number } = cds.entities('basic.literals') + await DELETE(number) await INSERT({ integer32: 1 }).into(number) await UPDATE(number).with({ integer32: { xpr: [{ ref: ['integer32'] }, '+', { val: 2 }] } }) const result = await SELECT.one.from(number) @@ -109,6 +109,7 @@ describe('UPDATE', () => { test('number', async () => { const { number } = cds.entities('basic.literals') + await DELETE(number) await INSERT({ integer32: 0 }).into(number) await UPDATE(number) .data({ integer32: 1 }) From 896fe42d82cdef8284c4822765f4e9e5e55700dd Mon Sep 17 00:00:00 2001 From: Bob den Os Date: Fri, 6 Feb 2026 11:07:47 +0100 Subject: [PATCH 2/2] Cleanup prepare cache impact --- hana/lib/HANAService.js | 73 +++++++++---------------------- hana/lib/drivers/base.js | 76 +++++++++++++++++---------------- hana/lib/drivers/hana-client.js | 42 +++++++++--------- hana/lib/drivers/hdb.js | 63 +++++++++++++-------------- hana/lib/drivers/stream.js | 4 +- 5 files changed, 117 insertions(+), 141 deletions(-) diff --git a/hana/lib/HANAService.js b/hana/lib/HANAService.js index 960fb413c..ab6ca8b12 100644 --- a/hana/lib/HANAService.js +++ b/hana/lib/HANAService.js @@ -167,11 +167,7 @@ class HANAService extends SQLService { let rows if (values?.length || blobs.length > 0 || isStream) { const ps = await this.prepare(sqlScript, blobs.length) - try { - rows = this.ensureDBC() && await ps[isStream ? 'stream' : 'all'](values || [], isOne, objectMode) - } finally { - ps.drop() - } + rows = this.ensureDBC() && await ps[isStream ? 'stream' : 'all'](values || [], isOne, objectMode) } else { rows = await this.exec(sqlScript) } @@ -207,34 +203,26 @@ class HANAService extends SQLService { const { sql, entries, cqn } = this.cqn2sql(query, data) if (!sql) return // Do nothing when there is nothing to be done const ps = await this.prepare(sql) - try { - // HANA driver supports batch execution - const results = await (entries - ? this.server.major <= 2 - ? entries.reduce((l, c) => l.then(() => this.ensureDBC() && ps.run(c)), Promise.resolve(0)) - : entries.length > 1 ? this.ensureDBC() && await ps.runBatch(entries) : this.ensureDBC() && await ps.run(entries[0]) - : this.ensureDBC() && ps.run()) - return new this.class.InsertResults(cqn, results) - } finally { - ps.drop() - } + // HANA driver supports batch execution + const results = await (entries + ? this.server.major <= 2 + ? entries.reduce((l, c) => l.then(() => this.ensureDBC() && ps.run(c)), Promise.resolve(0)) + : entries.length > 1 ? this.ensureDBC() && await ps.runBatch(entries) : this.ensureDBC() && await ps.run(entries[0]) + : this.ensureDBC() && ps.run()) + return new this.class.InsertResults(cqn, results) } async onUPSERT({ query, data }) { const { sql, entries, cqn } = this.cqn2sql(query, data) if (!sql) return // Do nothing when there is nothing to be done const ps = await this.prepare(sql) - try { - // HANA driver supports batch execution - const results = await (entries - ? this.server.major <= 2 - ? entries.reduce((l, c) => l.then(() => this.ensureDBC() && ps.run(c)), Promise.resolve(0)) - : entries.length > 1 ? this.ensureDBC() && await ps.runBatch(entries) : this.ensureDBC() && await ps.run(entries[0]) - : this.ensureDBC() && ps.run()) - return results.changes ?? results - } finally { - ps.drop() - } + // HANA driver supports batch execution + const results = await (entries + ? this.server.major <= 2 + ? entries.reduce((l, c) => l.then(() => this.ensureDBC() && ps.run(c)), Promise.resolve(0)) + : entries.length > 1 ? this.ensureDBC() && await ps.runBatch(entries) : this.ensureDBC() && await ps.run(entries[0]) + : this.ensureDBC() && ps.run()) + return results.changes ?? results } async onNOTFOUND(req, next) { @@ -1300,9 +1288,8 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON E async onSIMPLE({ query, data, event }) { const { sql, values } = this.cqn2sql(query, data) - let ps try { - ps = await this.prepare(sql) + const ps = await this.prepare(sql) return (this.ensureDBC() && await ps.run(values)).changes } catch (err) { // Allow drop to fail when the view or table does not exist @@ -1310,8 +1297,6 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON E return } throw err - } finally { - ps?.drop() } } @@ -1336,26 +1321,8 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON E const isAsync = /\sASYNC\s*$/.test(query) const outParameters = isAsync ? [{ PARAMETER_NAME: 'ASYNC_CALL_ID' }] : await this._getProcedureMetadata(name, schema) const ps = await this.prepare(query) - try { - const ret = this.ensureDBC() && await ps.proc(data, outParameters) - return isAsync ? ret.ASYNC_CALL_ID[0] : ret - } finally { - ps.drop() - } - } - - async _onPlainSQL({ query, data }, next) { - if (typeof query === 'string') { - DEBUG?.(query, data) - const ps = await this.prepare(query) - try { - const exec = this.hasResults(query) ? d => ps.all(d) : d => ps.run(d) - if (Array.isArray(data) && Array.isArray(data[0])) return await Promise.all(data.map(exec)) - else return exec(data) - } finally { - ps.drop() - } - } else return next() + const ret = this.ensureDBC() && await ps.proc(data, outParameters) + return isAsync ? ret.ASYNC_CALL_ID[0] : ret } async onPlainSQL(req, next) { @@ -1363,7 +1330,7 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON E if (/ IF EXISTS /i.test(req.query)) { req.query = req.query.replace(/ IF EXISTS/gi, '') try { - return await this._onPlainSQL(req, next) + return await super.onPlainSQL(req, next) } catch (err) { if (/(^|')DROP /i.test(req.query) && (err.code === 259 || err.code === 321)) { return @@ -1375,7 +1342,7 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON E const proc = this._getProcedureNameAndSchema(req.query) if (proc && proc.name) return this.onCall(req, proc.name, proc.schema) - return this._onPlainSQL(req, next) + return super.onPlainSQL(req, next) } onBEGIN() { diff --git a/hana/lib/drivers/base.js b/hana/lib/drivers/base.js index f945d1bd1..6fd0c4998 100644 --- a/hana/lib/drivers/base.js +++ b/hana/lib/drivers/base.js @@ -12,14 +12,30 @@ class HANADriver { this.statements = {} } - _prepare(sql) { - return module.exports.prom( + _prepare(sql, detached) { + let prep = (!detached && this.statements[sql]) || module.exports.prom( this._native, 'prepare', )(sql).then(stmt => { stmt._parentConnection = this._native return stmt }) + if (!detached) { + const release = {} // TODO: for node@22 use Promise.withResolvers() + release.promise = new Promise((resolve, reject) => { + release.resolve = resolve + release.reject = reject + }) + release.promise.catch(() => { }) + this.statements[sql] = release.promise + prep.catch(release.reject) + prep = prep.then(stmt => { + stmt.release = () => release.resolve(stmt) + return stmt + }) + } + + return prep } /** @@ -28,50 +44,38 @@ class HANADriver { * @returns {import('@cap-js/db-service/lib/SQLService').PreparedStatement} */ async prepare(sql) { - this.statements[sql] ??= this._prepare(sql) - - const prep = this.statements[sql].ready ?? this.statements[sql] - const release = {} // TODO: for node@22 use Promise.withResolvers() - release.promise = new Promise((resolve, reject) => { - release.resolve = resolve - release.reject = reject - }) - release.promise.catch(() => { }) - this.statements[sql].ready = release.promise - return { - _prep: prep, run: async params => { const { values, streams } = this._extractStreams(params) - const stmt = await prep - let changes = await module.exports.prom(stmt, 'exec')(values) - await this._sendStreams(stmt, streams) - return { changes } + const stmt = await this._prepare(sql) + try { + let changes = await module.exports.prom(stmt, 'exec')(values) + await this._sendStreams(stmt, streams) + return { changes } + } finally { stmt.release() } }, runBatch: async params => { - const stmt = await prep - const changes = await module.exports.prom(stmt, 'exec')(params) - return { changes: !Array.isArray(changes) ? changes : changes.reduce((l, c) => l + c, 0) } + const stmt = await this._prepare(sql) + try { + const changes = await module.exports.prom(stmt, 'exec')(params) + return { changes: !Array.isArray(changes) ? changes : changes.reduce((l, c) => l + c, 0) } + } finally { stmt.release() } }, get: async params => { - const stmt = await prep - return (await module.exports.prom(stmt, 'exec')(params))[0] + const stmt = await this._prepare(sql) + try { + const ret = (await module.exports.prom(stmt, 'exec')(params))[0] + return ret + } finally { stmt.release() } }, all: async params => { - const stmt = await prep - return module.exports.prom(stmt, 'exec')(params) + const stmt = await this._prepare(sql) + try { + const ret = await module.exports.prom(stmt, 'exec')(params) + stmt.release() + return ret + } finally { stmt.release() } }, - detach: async () => { - // Check whether something is waiting for the prepared statement already - if (this.statements[sql].ready === release.promise) { - this.statements[sql] = undefined - return - } - this._prepare(sql).then(release.resolve, release.reject) - }, - drop: async () => { - prep.then(release.resolve, release.reject) - } } } diff --git a/hana/lib/drivers/hana-client.js b/hana/lib/drivers/hana-client.js index bafc31315..46c75b845 100644 --- a/hana/lib/drivers/hana-client.js +++ b/hana/lib/drivers/hana-client.js @@ -66,8 +66,7 @@ class HANAClientDriver extends driver { // It would be required to allow using getDate() on previous rows if (hasBlobs) { ret.all = async (values) => { - ret.detach() - const stmt = await ret._prep + const stmt = await this._prepare(sql, true) // Create result set const reset = async function () { if (this) await prom(this, 'close')() @@ -121,27 +120,30 @@ class HANAClientDriver extends driver { ret.run = async params => { const { values, streams } = this._extractStreams(params) - const stmt = await ret._prep - let changes = await prom(stmt, 'exec')(values) - await this._sendStreams(stmt, streams) - // REVISIT: hana-client does not return any changes when doing an update with streams - // This causes the best assumption to be that the changes are one - // To get the correct information it is required to send a count with the update where clause - if (streams.length && changes === 0) { - changes = 1 - } - return { changes } + const stmt = await this._prepare(sql) + try { + let changes = await prom(stmt, 'exec')(values) + await this._sendStreams(stmt, streams) + // REVISIT: hana-client does not return any changes when doing an update with streams + // This causes the best assumption to be that the changes are one + // To get the correct information it is required to send a count with the update where clause + if (streams.length && changes === 0) { + changes = 1 + } + return { changes } + } finally { stmt.release() } } ret.proc = async (data, outParameters) => { - const stmt = await ret._prep - const rows = await prom(stmt, 'execQuery')(data) - return this._getResultForProcedure(rows, outParameters, stmt) + const stmt = await this._prepare(sql) + try { + const rows = await prom(stmt, 'execQuery')(data) + return this._getResultForProcedure(rows, outParameters, stmt) + } finally { stmt.release() } } ret.stream = async (values, one, objectMode) => { - const stmt = await ret._prep - ret.detach() + const stmt = await this._prepare(sql, true) values = Array.isArray(values) ? values : [] // Uses the native exec method instead of executeQuery to initialize a full stream // As executeQuery does not request the whole result set at once @@ -165,7 +167,7 @@ class HANAClientDriver extends driver { if (rs.isNull(0)) return null return Readable.from(streamBlob(rs, undefined, 0), { objectMode: false }) } - return rsIterator(rs, one, objectMode) + return rsIterator(rs, one, objectMode, () => { stmt.drop() }) } return ret } @@ -239,7 +241,7 @@ class HANAClientDriver extends driver { HANAClientDriver.pool = true -async function rsIterator(rs, one, objectMode) { +async function rsIterator(rs, one, objectMode, onDone) { rs._rowPosition = -1 rs.nextAsync = prom(rs, 'next') rs.getValueAsync = prom(rs, 'getValue') @@ -317,7 +319,7 @@ async function rsIterator(rs, one, objectMode) { } } - return resultSetStream(state, one, objectMode) + return resultSetStream(state, one, objectMode, onDone) } async function* streamBlob(rs, rowIndex = -1, columnIndex, binaryBuffer) { diff --git a/hana/lib/drivers/hdb.js b/hana/lib/drivers/hdb.js index 184bb9664..edd6f4585 100644 --- a/hana/lib/drivers/hdb.js +++ b/hana/lib/drivers/hdb.js @@ -89,34 +89,36 @@ class HDBDriver extends driver { if (hasBlobs) { ret.all = async (values) => { - const stmt = await ret._prep - // Create result set - const rs = await prom(stmt, 'execute')(values) - const cols = rs.metadata.map(b => b.columnName) - const stream = rs.createReadStream() - - const result = [] - for await (const row of stream) { - const obj = {} - for (let i = 0; i < cols.length; i++) { - const col = cols[i] - // hdb returns large strings as streams sometimes - if (col === '_json_' && typeof row[col] === 'object') { - obj[col] = await text(row[col].createReadStream()) - continue + const stmt = await this._prepare(sql) + try { + // Create result set + const rs = await prom(stmt, 'execute')(values) + const cols = rs.metadata.map(b => b.columnName) + const stream = rs.createReadStream() + + const result = [] + for await (const row of stream) { + const obj = {} + for (let i = 0; i < cols.length; i++) { + const col = cols[i] + // hdb returns large strings as streams sometimes + if (col === '_json_' && typeof row[col] === 'object') { + obj[col] = await text(row[col].createReadStream()) + continue + } + obj[col] = i > 3 + ? row[col] === null + ? null + : ( + row[col].createReadStream?.() + || row[col] + ) + : row[col] } - obj[col] = i > 3 - ? row[col] === null - ? null - : ( - row[col].createReadStream?.() - || row[col] - ) - : row[col] + result.push(obj) } - result.push(obj) - } - return result + return result + } finally { stmt.release() } } } @@ -126,10 +128,9 @@ class HDBDriver extends driver { } ret.stream = async (values, one, objectMode) => { - const stmt = await ret._prep - ret.detach() + const stmt = await this._prepare(sql, true) const rs = await prom(stmt, 'execute')(values || []) - return rsIterator(rs, one, objectMode) + return rsIterator(rs, one, objectMode, () => { stmt.drop() }) } return ret } @@ -178,7 +179,7 @@ class HDBDriver extends driver { } } -async function rsIterator(rs, one, objectMode) { +async function rsIterator(rs, one, objectMode, onDone) { // Raw binary data stream unparsed const raw = rs.createBinaryStream()[Symbol.asyncIterator]() @@ -399,7 +400,7 @@ async function rsIterator(rs, one, objectMode) { } } - return resultSetStream(state, one, objectMode) + return resultSetStream(state, one, objectMode, onDone) } const readString = function (state, isJson = false) { diff --git a/hana/lib/drivers/stream.js b/hana/lib/drivers/stream.js index 92c589079..56464a3f7 100644 --- a/hana/lib/drivers/stream.js +++ b/hana/lib/drivers/stream.js @@ -134,7 +134,7 @@ function rsNextRaw(state, next) { return writeBlobs ? writeBlobs.then(_return) : _return() } -async function rsIterator(state, one, objectMode) { +async function rsIterator(state, one, objectMode, onDone) { const stream = state.stream = new Readable({ objectMode, async read() { @@ -160,6 +160,8 @@ async function rsIterator(state, one, objectMode) { stream.push('[') } + if (onDone) { stream.on('close', onDone) } + return stream }