PoC: Faster HANA streaming

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,15 +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 = isSimple && 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 = 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)
     }
@@ -189,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 }) {
@@ -442,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
         )
@@ -465,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)
@@ -599,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 => {
@@ -672,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

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,7 +177,83 @@ class HDBDriver extends driver {
   }
 }
 
+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 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]()
 

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)