Skip to content

PoC: support returning #1613

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
BobdenOs wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

PoC: support returning #1613

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6301970
initial version for returning
BobdenOs May 20, 2026
e6bb941
HANA returning simple INSERT
BobdenOs May 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 21 additions & 21 deletions db-service/lib/SQLService.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ class SQLService extends DatabaseService {
}

_buffer(val) {
if (val === null) return null
if (val == null) return null
return Buffer.from(val, 'base64')
}

Expand Down Expand Up @@ -184,25 +184,16 @@ class SQLService extends DatabaseService {
* Handler for INSERT
* @type {Handler}
*/
async onINSERT({ query, data }) {
const { sql, entries, cqn } = this.cqn2sql(query, data)
if (!sql) return // Do nothing when there is nothing to be done // REVISIT: fix within mtxs
const ps = await this.prepare(sql)
const results = entries ? await Promise.all(entries.map(e => ps.run(e))) : await ps.run()
return new this.class.InsertResults(cqn, results)
async onINSERT(req) {
return this.onSIMPLE(req)
}

/**
* Handler for UPSERT
* @type {Handler}
*/
async onUPSERT({ query, data }) {
const { sql, entries } = this.cqn2sql(query, data)
if (!sql) return // Do nothing when there is nothing to be done // REVISIT: When does this happen?
const ps = await this.prepare(sql)
const results = entries ? await Promise.all(entries.map(e => ps.run(e))) : await ps.run()
// REVISIT: results isn't an array, when no entries -> how could that work? when do we have no entries?
return results.reduce((total, affectedRows) => total + affectedRows.changes, 0)
async onUPSERT(req) {
return this.onSIMPLE(req)
}

/**
Expand All @@ -225,9 +216,18 @@ class SQLService extends DatabaseService {
* @type {Handler}
*/
async onSIMPLE({ query, data }) {
const { sql, values } = this.cqn2sql(query, data)
const { sql, values, entries, cqn } = this.cqn2sql(query, data)
let ps = await this.prepare(sql)
return (await ps.run(values)).changes
const run = cqn.kind === 'SELECT' || cqn[cqn.kind]?.returning
? e => ps.all(e)
: e => ps.run(e)
let results = entries
? (await Promise.all(entries.map(run)))
.reduce((l, c) => 'changes' in c ? (l.affected += c.changes) && l : [...l, ...c], Object.assign([], { affected: 0 }))
: await run(values)
if (!Array.isArray(results)) results = Object.assign([], { affected: results.changes })
results.affected = results.length || results.affected // use || to ignore length:0
return results
}

get onDELETE() {
Expand Down Expand Up @@ -300,7 +300,7 @@ class SQLService extends DatabaseService {
* @type {Handler}
*/
async onEVENT({ event }) {
if(DEBUG._debug) DEBUG.debug(event) // in the other cases above DEBUG happens in cqn2sql
if (DEBUG._debug) DEBUG.debug(event) // in the other cases above DEBUG happens in cqn2sql
return await this.exec(event)
}

Expand All @@ -310,7 +310,7 @@ class SQLService extends DatabaseService {
*/
async onPlainSQL({ query, data }, next) {
if (typeof query === 'string') {
if(DEBUG._debug) DEBUG.debug(query, data)
if (DEBUG._debug) DEBUG.debug(query, data)
const ps = await this.prepare(query)
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))
Expand Down Expand Up @@ -415,7 +415,7 @@ class SQLService extends DatabaseService {
* @param {import('@sap/cds/apis/cqn').Query} q
* @returns {import('./infer/cqn').Query}
*/
cqn4sql(q, useTechnicalAlias=true) {
cqn4sql(q, useTechnicalAlias = true) {
if (
!cds.env.features.db_strict &&
!q.SELECT?.from?.join &&
Expand Down Expand Up @@ -517,7 +517,7 @@ const DEBUG_PQL = cds.log('pql')
if (DEBUG_PQL._debug || cds.repl) {

// Add helper method to convert CQN to PQL, used below...
SQLService.prototype.cqn2pql = function cqn2pql (query, values) {
SQLService.prototype.cqn2pql = function cqn2pql(query, values) {
const CQN2PQL = cqn2pql.renderer ??= require('./cqn2pql')
return new CQN2PQL(this).render(query, values)
}
Expand Down Expand Up @@ -568,7 +568,7 @@ if (DEBUG_PQL._debug || cds.repl) {
* if no real SQL service is available yet through cds.db.
*/
class db extends SQLService {
/** @returns {SQLService} */
/** @returns {SQLService} */
static get srv() { return cds.db || (this.singleton ??= new this) }
get factory() { return null }
get model() { return cds.model }
Expand Down
27 changes: 18 additions & 9 deletions db-service/lib/cqn2sql.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -920,8 +920,9 @@ class CQN2SQLRenderer {
}

const extractions = this._managed = this.managed(columns.map(c => ({ name: c })), elements)
return (this.sql = `INSERT INTO ${this.quote(entity)}${alias ? ' as ' + this.quote(alias) : ''} (${this.columns.map(c => this.quote(transitions.mapping.get(c)?.ref?.[0] || c))
}) SELECT ${extractions.slice(0, columns.length).map(c => c.insert)} FROM json_each(?)`)

return (this.sql = this.returning(`INSERT INTO ${this.quote(entity)}${alias ? ' as ' + this.quote(alias) : ''} (${this.columns.map(c => this.quote(transitions.mapping.get(c)?.ref?.[0] || c))
}) SELECT ${extractions.slice(0, columns.length).map(c => c.insert)} FROM json_each(?)`, INSERT.returning, q))
}

async *INSERT_entries_stream(entries, binaryEncoding = 'base64') {
Expand Down Expand Up @@ -1053,8 +1054,8 @@ class CQN2SQLRenderer {
.map(c => c.converter(c.extract))

const transitions = this.srv.resolve.transitions(q)
return (this.sql = `INSERT INTO ${this.quote(entity)}${alias ? ' as ' + this.quote(alias) : ''} (${this.columns.map(c => this.quote(transitions.mapping.get(c)?.ref?.[0] || c))
}) SELECT ${extraction} FROM json_each(?)`)
return (this.sql = this.returning(`INSERT INTO ${this.quote(entity)}${alias ? ' as ' + this.quote(alias) : ''} (${this.columns.map(c => this.quote(transitions.mapping.get(c)?.ref?.[0] || c))
}) SELECT ${extraction} FROM json_each(?)`, INSERT.returning, q))
}

/**
Expand Down Expand Up @@ -1091,7 +1092,7 @@ class CQN2SQLRenderer {
? `SELECT ${extractions.map(c => `${c.insert} AS ${this.quote(c.name)}`)} FROM (${this.SELECT(src)}) AS NEW`
: this.SELECT(src)
if (extractions.length > columns.length) columns = this.columns = extractions.map(c => c.name)
this.sql = `INSERT INTO ${this.quote(entity)}${alias ? ' as ' + this.quote(alias) : ''} (${columns.map(c => this.quote(transitions.mapping.get(c)?.ref?.[0] || c))}) ${sql}`
this.sql = this.returning(`INSERT INTO ${this.quote(entity)}${alias ? ' as ' + this.quote(alias) : ''} (${columns.map(c => this.quote(transitions.mapping.get(c)?.ref?.[0] || c))}) ${sql}`, INSERT.returning, q)
this.entries = [this.values]
return this.sql
}
Expand Down Expand Up @@ -1182,8 +1183,9 @@ class CQN2SQLRenderer {
}).map(c => `${this.quote(c)} = excluded.${this.quote(c)}`)

const transitions = this.srv.resolve.transitions(q)
return (this.sql = `INSERT INTO ${this.quote(entity)} (${columns.map(c => this.quote(transitions.mapping.get(c)?.ref?.[0] || c))}) ${sql
} WHERE TRUE ON CONFLICT(${keys.map(c => this.quote(c))}) DO ${updateColumns.length ? `UPDATE SET ${updateColumns}` : 'NOTHING'}`)
return (this.sql = this.returning(`INSERT INTO ${this.quote(entity)} (${columns.map(c => this.quote(transitions.mapping.get(c)?.ref?.[0] || c))}) ${sql
} WHERE TRUE ON CONFLICT(${keys.map(c => this.quote(c))}) DO ${updateColumns.length ? `UPDATE SET ${updateColumns}` : 'NOTHING'
}`, UPSERT.returning, q))
}

// UPDATE Statements ------------------------------------------------
Expand All @@ -1194,7 +1196,7 @@ class CQN2SQLRenderer {
* @returns {string} SQL
*/
UPDATE(q) {
const { entity, with: _with, data, where } = q.UPDATE
const { entity, with: _with, data, where, returning } = q.UPDATE
const transitions = this.srv.resolve.transitions(q)
const elements = q._target?.elements
let sql = `UPDATE ${this.quote(this.table_name(q))}`
Expand Down Expand Up @@ -1225,6 +1227,7 @@ class CQN2SQLRenderer {

sql += ` SET ${extraction}`
if (where) sql += ` WHERE ${this.where_resolved(entity.as, where, q)}`
if (returning) sql = this.returning(sql, returning, q)
return (this.sql = sql)
}

Expand All @@ -1236,10 +1239,11 @@ class CQN2SQLRenderer {
* @returns {string} SQL
*/
DELETE(q) {
const { DELETE: { where, from } } = q
const { DELETE: { where, from, returning } } = q
let sql = `DELETE FROM ${this.quote(this.table_name(q))}`
if (from.as) sql += ` AS ${this.quote(from.as)}`
if (where) sql += ` WHERE ${this.where(where)}`
if (returning) sql = this.returning(sql, returning, q)
return (this.sql = sql)
}

Expand Down Expand Up @@ -1485,6 +1489,11 @@ class CQN2SQLRenderer {
return s
}

returning(sql, cols, q) {
if (cols && cols.length) return `${sql} RETURNING ${cols.map(c => this.column_expr(c, q))}`
return sql
}

/**
* Converts the columns array into an array of SQL expressions that extract the correct value from inserted JSON data
* @param {object[]} columns
Expand Down
61 changes: 47 additions & 14 deletions hana/lib/HANAService.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 })
Expand Down Expand Up @@ -162,7 +162,7 @@ class HANAService extends SQLService {
let sqlScript = isLockQuery || 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) {
const ps = await this.prepare(sqlScript, blobs.length)
Expand All @@ -176,20 +176,20 @@ class HANAService extends SQLService {
const resultQuery = query.clone()
resultQuery.SELECT.forUpdate = undefined
resultQuery.SELECT.forShareLock = undefined

const keys = Object.keys(req.target?.keys || {})

if (keys.length && query.SELECT.forUpdate?.ignoreLocked) {
// Exit early when no row was found in the inital query
if (rows.length === 0) return isOne ? undefined : []

// Filter for those rows that were locked by the initial query
const left = { list: keys.map(k => ({ ref: [k] })) }
const right = { list: rows.map(r => ({ list: keys.map(k => ({ val: r[k.toUpperCase()] })) })) }
resultQuery.SELECT.limit = undefined
resultQuery.SELECT.where = [left, 'in', right]
}

return this.onSELECT({ query: resultQuery, __proto__: req })
}

Expand All @@ -208,12 +208,14 @@ class HANAService extends SQLService {
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
let 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)
if (Array.isArray(results.changes) && Array.isArray(results.changes[1])) results = Object.assign(results.changes[1], { affected: results.changes[1].length })
else results = Object.assign([], { affected: results.changes })
return results
}

async onNOTFOUND(req, next) {
Expand Down Expand Up @@ -603,7 +605,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 => {
Expand Down Expand Up @@ -817,9 +819,10 @@ class HANAService extends SQLService {
// With the buffer table approach the data is processed in chunks of a configurable size
// Which allows even smaller HANA systems to process large datasets
// But the chunk size determines the maximum size of a single row
return (this.sql = `INSERT INTO ${this.quote(entity)} (${this.columns.map(c => this.quote(transitions.mapping.get(c)?.ref?.[0] || c))
this._new = `JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON ERROR)`
return (this.sql = this.returning(`INSERT INTO ${this.quote(entity)} (${this.columns.map(c => this.quote(transitions.mapping.get(c)?.ref?.[0] || c))
}) WITH SRC AS (SELECT ? AS JSON FROM DUMMY UNION ALL SELECT TO_NCLOB(NULL) AS JSON FROM DUMMY)
SELECT ${converter} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON ERROR) AS NEW`)
SELECT ${converter} FROM ${this._new} AS NEW`, INSERT.returning, q))
}

INSERT_rows(q) {
Expand Down Expand Up @@ -888,8 +891,9 @@ class HANAService extends SQLService {

let sql
if (UPSERT.entries || UPSERT.rows || UPSERT.values) {
this._old = `JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON ERROR) AS NEW LEFT JOIN ${this.quote(entity)} AS OLD ON ${keyCompare}`
sql = `WITH SRC AS (SELECT ? AS JSON FROM DUMMY UNION ALL SELECT TO_NCLOB(NULL) AS JSON FROM DUMMY)
SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON ERROR) AS NEW LEFT JOIN ${this.quote(entity)} AS OLD ON ${keyCompare}`
SELECT ${mixing} FROM ${this._old}`
} else {
const src = this.cqn4sql(UPSERT.from || UPSERT.as)
if (this.values) this.values = []
Expand All @@ -898,7 +902,8 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON E
.map((c, i) => ({ ref: [this.column_name(c)], as: this.columns[i] }))
)
.from(src)
sql = `SELECT ${mixing} FROM (${this.SELECT(aliasedQuery)}) AS NEW LEFT JOIN ${this.quote(entity)} AS OLD ON ${keyCompare}`
this._old = `(${this.SELECT(aliasedQuery)}) AS NEW LEFT JOIN ${this.quote(entity)} AS OLD ON ${keyCompare}`
sql = `SELECT ${mixing} FROM ${this._old}`
this.entries = [this.values]
}

Expand Down Expand Up @@ -1222,6 +1227,34 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction}) ERROR ON E
this.values = [...prefix.map(p => p.values).flat(), ...values]
}

returning(sql, cols, q) {
if (cols && cols.length) {
// INSERT.from NEW has to be snapshotted before doing the INSERT
// Has to be joined back with actual entity (impossible for entities without keys)
// INSERT.entries NEW should be snapshotted before doing the INSERT
// So that the actual INSERT can use the snapshot
// As the snapshot can then also be joined back with the actual entity (also impossible for entities without keys)


// INSERT only NEW
// DELETE only OLD
// UPSERT, UPDATE both NEW and OLD
// TODO: ensure that OLD and NEW are generated based upon the data targeted by the query
const _q = q[q.kind]
// this._old ??= this.SELECT(cds.ql({ SELECT: { __proto__: _q, from: _q.entity } }))
this._new ??= `(${this.SELECT(q.INSERT.from)})`
return `DO (${q.INSERT?.entries ? 'IN JSON NCLOB => ?' : ''}) BEGIN
-- OLD = ${this._old};
NEW = SELECT * FROM ${this._new};
${sql};
SELECT ${cols.map(c => this.column_expr(c, q))} FROM :NEW;
END`
.replace(/WITH SRC AS.*/, '')
.replaceAll('JSON_TABLE(SRC.JSON', 'JSON_TABLE(:JSON')
}
return sql
}

// Loads a static result from the query `SELECT * FROM RESERVED_KEYWORDS`
static ReservedWords = { ...super.ReservedWords, ...hanaKeywords }

Expand Down
Loading
Loading