From c5b0192eba5aaafda3f5fa11da73961c7c30a9ad Mon Sep 17 00:00:00 2001 From: Bob den Os Date: Fri, 21 Nov 2025 12:50:26 +0100 Subject: [PATCH 1/2] Simple implementation for hyper streaming --- hana/lib/HANAService.js | 5 +- hana/lib/drivers/hdb.js | 22 ++++++++ test/scenarios/bookshop/stream.perf.test.js | 59 +++++++-------------- 3 files changed, 46 insertions(+), 40 deletions(-) diff --git a/hana/lib/HANAService.js b/hana/lib/HANAService.js index aba127ff0..dcbc689ff 100644 --- a/hana/lib/HANAService.js +++ b/hana/lib/HANAService.js @@ -161,9 +161,12 @@ class HANAService extends SQLService { const isSimple = temporary.length + blobs.length + withclause.length === 0 const isOne = cqn.SELECT.one || query.SELECT.from.ref?.[0].cardinality?.max === 1 const isStream = iterator && !isLockQuery + const isHyper = isStream && withclause.length < 2 && cds.env.features.hyper_streaming // REVISIT: add prepare options when param:true is used - let sqlScript = isLockQuery || isSimple ? sql : this.wrapTemporary(temporary, withclause, blobs) + let sqlScript = isHyper + ? `${sql} FOR JSON ('format'='no', 'omitnull'='no', 'arraywrap'='${isOne ? 'no' : 'yes'}') RETURNS NVARCHAR(2147483647)` + : isLockQuery || isSimple ? sql : this.wrapTemporary(temporary, withclause, blobs) const { hints } = query.SELECT if (hints) sqlScript += ` WITH HINT (${hints.join(',')})` let rows diff --git a/hana/lib/drivers/hdb.js b/hana/lib/drivers/hdb.js index a2a200f66..49d05e1ac 100644 --- a/hana/lib/drivers/hdb.js +++ b/hana/lib/drivers/hdb.js @@ -180,6 +180,28 @@ async function rsIterator(rs, one, objectMode) { // Raw binary data stream unparsed const raw = rs.createBinaryStream()[Symbol.asyncIterator]() + // Hyper stream + if (rs.metadata.length === 1) { + const row = await raw.next() + let length = row.value[0] + let offset = 1 + switch (length) { + case 0xff: + return null + case 0xf6: + length = row.value.readInt16LE(offset) + offset += 2 + break + case 0xf7: + length = row.value.readInt32LE(offset) + offset += 4 + break + default: + } + const data = row.value.slice(offset, offset + length) + return Readable.from(data, { objectMode: false }) + } + const blobs = rs.metadata.slice(4).map(b => b.columnName) const levels = [ { diff --git a/test/scenarios/bookshop/stream.perf.test.js b/test/scenarios/bookshop/stream.perf.test.js index 51b0a9f6e..e7f7bc915 100644 --- a/test/scenarios/bookshop/stream.perf.test.js +++ b/test/scenarios/bookshop/stream.perf.test.js @@ -19,7 +19,7 @@ describe.skip('Bookshop: Stream Performance', () => { let i = 1000 const gen = function* () { yield `[{"ID":${i++},"title":"${i}","author_ID":101,"genre_ID":11}` - for (; i < 100000; i++) { + for (; i < 100_000; i++) { yield `,{"ID":${i},"title":"${i}","author_ID":101,"genre_ID":11}` } yield ']' @@ -30,14 +30,14 @@ describe.skip('Bookshop: Stream Performance', () => { }) const measure = [ - // { async: false }, + { async: false }, { async: true }, ] const modes = [ { objectMode: false }, - { objectMode: true }, - { objectMode: null }, + // { objectMode: true }, + // { objectMode: null }, ] const scenarios = [ @@ -74,45 +74,22 @@ describe.skip('Bookshop: Stream Performance', () => { let peakMemory = 0 const proms = [] + const txs = await Promise.all(new Array(20).fill().map(async () => (await cds.tx()).begin())) const s = performance.now() - for (let x = 0; x < 20; x++) { - const prom = cds.tx(async tx => { - const stream = await tx.dispatch(req) - - if (objectMode !== false) { - await pipeline( - stream, - async function* (source) { - for await (const row of source) { - if (withImage) { - /* const buffer = */ await streamConsumers.buffer(row.image) - // if (imageData.compare(buffer)) throw new Error('Blob stream does not contain the original data') - } - - if (withExpands) { - await Promise.all([ - pipeline(row.genre, devNull()), - pipeline(row.author, devNull()), - ]) - } - - yield JSON.stringify(row) - } - }, - devNull() - ) - } else { - await pipeline(stream, devNull()) - } + let fastest = Number.POSITIVE_INFINITY + for (let total = 0; total < 100; total++) { + const tx = txs[total % txs.length] + proms.push((async () => { + const s = performance.now() + const stream = await tx.dispatch(req) + await pipeline(stream, devNull()) + const dur = performance.now() - s + if (dur < fastest) fastest = dur const curMemory = process.memoryUsage().heapUsed if (curMemory > peakMemory) peakMemory = curMemory - }) - - proms.push(prom) - if (!async) { - await prom - } + })()) + if (async === false) await proms.at(-1) } const allResults = await Promise.allSettled(proms) @@ -123,6 +100,10 @@ describe.skip('Bookshop: Stream Performance', () => { process.stdout.write( `${scenarios.length > 1 ? ' ' : ''} - Duration: ${dur >>> 0} ms Rows: ${totalRows} (${(totalRows / dur) >>> 0} rows/ms) (${(peakMemory / 1024 / 1024) >>> 0} MiB mem)\n` ) + process.stdout.write( + `${scenarios.length > 1 ? ' ' : ''} - Fastest: ${fastest >>> 0} ms Rows: ${rows} (${(rows / fastest) >>> 0} rows/ms)\n` + ) + await Promise.all(txs.map(tx => tx.commit())) }, 120 * 1000) From 781bef1046953e71c42cf23b7eb5b9b577193648 Mon Sep 17 00:00:00 2001 From: Bob den Os Date: Mon, 4 May 2026 11:17:38 +0200 Subject: [PATCH 2/2] re introduce sql_simple_queries flag in parallel to temporary hyper flag --- hana/lib/HANAService.js | 47 ++++++++++++-------- hana/lib/drivers/hdb.js | 97 ++++++++++++++++++++++++++++++++--------- 2 files changed, 105 insertions(+), 39 deletions(-) diff --git a/hana/lib/HANAService.js b/hana/lib/HANAService.js index a172c4673..63a5db3e1 100644 --- a/hana/lib/HANAService.js +++ b/hana/lib/HANAService.js @@ -1,6 +1,7 @@ const fs = require('fs') const path = require('path') const { Readable } = require('stream') +const { json } = require('stream/consumers') const { SQLService } = require('@cap-js/db-service') const drivers = require('./drivers') @@ -73,8 +74,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 }) @@ -159,18 +160,20 @@ class HANAService extends SQLService { const isSimple = temporary.length + blobs.length + withclause.length === 0 const isOne = cqn.SELECT.one || query.SELECT.from.ref?.[0].cardinality?.max === 1 const isStream = iterator && !isLockQuery - const isHyper = isStream && withclause.length < 2 && cds.env.features.hyper_streaming + const isHyper = isSimple && cds.env.features.hyper_streaming - // REVISIT: add prepare options when param:true is used - let sqlScript = isHyper - ? `${sql} FOR JSON ('format'='no', 'omitnull'='no', 'arraywrap'='${isOne ? 'no' : 'yes'}') RETURNS NVARCHAR(2147483647)` - : isLockQuery || isSimple ? sql : this.wrapTemporary(temporary, withclause, blobs) + let sqlScript = isLockQuery ? sql + : isHyper ? objectMode + ? `SELECT '$[0' as "_path_",'{}' as "_blobs_",'{}' as "_expands_",(` + sql.replace('FROM', `FROM DUMMY FOR JSON ('format'='no', 'omitnull'='no', 'arraywrap'='no') RETURNS NVARCHAR(2147483647)) as "_json_" FROM`) + : `${sql} FOR JSON ('format'='no', 'omitnull'='no', 'arraywrap'='${isOne ? 'no' : 'yes'}') RETURNS NVARCHAR(2147483647)` + : isSimple ? sql : this.wrapTemporary(temporary, withclause, blobs) const { hints } = query.SELECT if (hints) sqlScript += ` WITH HINT (${hints.join(',')})` let rows - if (values?.length || blobs.length > 0 || isStream) { + if (values?.length || blobs.length > 0 || isStream || isHyper) { + // REVISIT: add prepare options when param:true is used const ps = await this.prepare(sqlScript, blobs.length) - rows = this.ensureDBC() && await ps[isStream ? 'stream' : 'all'](values || [], isOne, objectMode) + rows = this.ensureDBC() && await ps[isStream || isHyper ? 'stream' : 'all'](values || [], isOne, objectMode) } else { rows = await this.exec(sqlScript) } @@ -192,14 +195,18 @@ class HANAService extends SQLService { return this.onSELECT({ query: resultQuery, __proto__: req }) } - if (rows.length && !isSimple) { - rows = this.parseRows(rows) + if (isStream) return rows + if (isHyper) { + if (rows[Symbol.asyncIterator]) rows = await json(rows) + else JSON.parse(rows[0]?.JSONRESULT ?? 'null') } + else if (rows.length && !isSimple) rows = this.parseRows(rows) + if (cqn.SELECT.count) { // REVISIT: the runtime always expects that the count is preserved with .map, required for renaming in mocks return HANAService._arrayWithCount(rows, await this.count(query, rows)) } - return isOne && !isStream ? rows[0] : rows + return isOne && !isStream && !isHyper ? rows[0] : rows } async onINSERT({ query, data }) { @@ -445,8 +452,10 @@ class HANAService extends SQLService { }) const isSimpleQuery = ( - cds.env.features.sql_simple_queries && - (cds.env.features.sql_simple_queries > 1 || !hasBooleans) && + ( + (cds.env.features.sql_simple_queries && (cds.env.features.sql_simple_queries > 1 || !hasBooleans)) + || cds.env.features.hyper_streaming + ) && !hasStructures && !parent ) @@ -468,7 +477,7 @@ class HANAService extends SQLService { q.as = q.SELECT.from.as } - const outputAliasSimpleQueriesRequired = cds.env.features.sql_simple_queries + const outputAliasSimpleQueriesRequired = (cds.env.features.sql_simple_queries || cds.env.features.hyper_streaming) && (orderByHasOutputColumnRef || having) if (outputAliasSimpleQueriesRequired || rowNumberRequired || q.SELECT.columns.length !== aliasedOutputColumns.length) { q = cds.ql.SELECT(aliasedOutputColumns).from(q) @@ -602,7 +611,7 @@ class HANAService extends SQLService { // if (col.ref?.length === 1) { col.ref.unshift(parent.as) } if (col.ref?.length > 1) { const colName = this.column_name(col) - + const isSource = from => { if (from.as === col.ref[0]) return true return from.args?.some(a => { @@ -675,8 +684,10 @@ class HANAService extends SQLService { const blobColumns = Object.keys(blobs) this.blobs.push(...blobColumns.filter(b => !this.blobs.includes(b))) if ( - cds.env.features.sql_simple_queries && - (cds.env.features.sql_simple_queries > 1 || !hasBooleans) && + ( + (cds.env.features.sql_simple_queries && (cds.env.features.sql_simple_queries > 1 || !hasBooleans)) + || cds.env.features.hyper_streaming + ) && structures.length + ObjectKeys(expands).length + ObjectKeys(blobs).length === 0 && !q?.src?.SELECT?.parent && this.temporary.length === 0 diff --git a/hana/lib/drivers/hdb.js b/hana/lib/drivers/hdb.js index 82b9e4a85..f83c3ef69 100644 --- a/hana/lib/drivers/hdb.js +++ b/hana/lib/drivers/hdb.js @@ -126,6 +126,7 @@ class HDBDriver extends driver { ret.stream = async (values, one, objectMode) => { const stmt = await ret._prep + // if (stmt.resultSetMetadata.length === 1) return hyperStream(stmt, values, one, objectMode) const rs = await prom(stmt, 'execute')(values || []) return rsIterator(rs, one, objectMode) } @@ -176,32 +177,86 @@ class HDBDriver extends driver { } } -async function rsIterator(rs, one, objectMode) { - // Raw binary data stream unparsed - const raw = rs.createBinaryStream()[Symbol.asyncIterator]() +async function hyperStream(stmt, values, one, objectMode) { + const message = require('hdb/lib/protocol/reply/index.js') + const connection = stmt._connection + const socket = connection._socket + const stream = Readable.from(slice(socket), { objectMode: false }) + + const ondata = socket._events.data + socket._events.data = undefined + // socket.off('data', ondata) + + async function* slice(stream) { + let segment + let packetLength + let packetRead = 0 + let rssize + let streamsize + let streamRead = 0 + const it = stream.iterator({ destroyOnReturn: false }) + for await (const chunk of it) { + let offset = 0 + packetRead += chunk.length + if (packetLength == null) { + packetLength = chunk.readUInt32LE(12) + 32 + offset += 32 + } + if (!segment) { + segment = message.Segment.create(chunk.subarray(offset), 0) + const rs = segment.parts.at(-1) + rssize = rs.buffer?.length ?? 0 + // rs.buffer = null // release buffer memory allocation again + offset = segment.parts.reduce((l, c) => l + c.byteLength, 24 + offset) + } + if (offset < chunk.length && !streamsize) { + let length = chunk[offset++] + switch (length) { + case 0xff: + return null + case 0xf6: + length = chunk.readInt16LE(offset) + offset += 2 + break + case 0xf7: + length = chunk.readInt32LE(offset) + offset += 4 + break + default: + } + streamsize = length + } + if (offset < chunk.length && streamsize) { + const part = chunk.subarray(offset) + streamRead += part.length + yield part + if (streamRead >= streamsize) break + } + if (packetRead >= packetLength) break + } + if (!streamRead) yield 'null' + socket._events.data = ondata + connection._queue.busy = false + connection._queue.resume() + } + + stmt.execute(values || [], (err, res) => { debugger }) + + return stream +} + +async function rsIterator(rs, one, objectMode) { // Hyper stream if (rs.metadata.length === 1) { - const row = await raw.next() - let length = row.value[0] - let offset = 1 - switch (length) { - case 0xff: - return null - case 0xf6: - length = row.value.readInt16LE(offset) - offset += 2 - break - case 0xf7: - length = row.value.readInt32LE(offset) - offset += 4 - break - default: - } - const data = row.value.slice(offset, offset + length) - return Readable.from(data, { objectMode: false }) + const raw = rs.createReadStream()[Symbol.asyncIterator]() + const { value } = await raw.next() + return Readable.from(value?.JSONRESULT ?? 'null', { objectMode: false }) } + // Raw binary data stream unparsed + const raw = rs.createBinaryStream()[Symbol.asyncIterator]() + const blobs = rs.metadata.slice(4).map(b => b.columnName) const levels = [ {