feat(core): support atomic updates via raw() helper (#4094)

When you want to issue an atomic update query via flush, you can use the
static `raw()` helper:

```ts
const ref = em.getReference(Author, 123);
ref.age = raw(`age * 2`);

await em.flush();
console.log(ref.age); // real value is available after flush
```

The `raw()` helper returns special raw query fragment object. It
disallows serialization (via `toJSON`) as well as working with the value
(via
[`valueOf()`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Object/valueOf)).
Only single use of this value is allowed, if you try to reassign it to
another entity, an error will be thrown to protect you from mistakes
like this:

```ts
order.number = raw(`(select max(num) + 1 from orders)`);
user.lastOrderNumber = order.number; // throws, it could resolve to a different value
JSON.stringify(order); // throws, raw value cannot be serialized
```

Closes #3657

docs/docs/entity-manager.md

@@ -401,6 +401,26 @@ await em.flush();
 
 This is a rough equivalent to calling `em.nativeUpdate()`, with one significant difference - we use the flush operation which handles event execution, so all life cycle hooks as well as flush events will be fired.
 
+## Atomic updates via `raw()` helper
+
+When you want to issue an atomic update query via flush, you can use the static `raw()` helper:
+
+```ts
+const ref = em.getReference(Author, 123);
+ref.age = raw(`age * 2`);
+
+await em.flush();
+console.log(ref.age); // real value is available after flush
+```
+
+The `raw()` helper returns special raw query fragment object. It disallows serialization (via `toJSON`) as well as working with the value (via [`valueOf()`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Object/valueOf)). Only single use of this value is allowed, if you try to reassign it to another entity, an error will be thrown to protect you from mistakes like this:
+
+```ts
+order.number = raw(`(select max(num) + 1 from orders)`);
+user.lastOrderNumber = order.number; // throws, it could resolve to a different value
+JSON.stringify(order); // throws, raw value cannot be serialized
+```
+
 ## Upsert
 
 We can use `em.upsert()` create or update the entity, based on whether it is already present in the database. This method performs an `insert on conflict merge` query ensuring the database is in sync, returning a managed entity instance. The method accepts either `entityName` together with the entity `data`, or just entity instance.

docs/docs/filters.md

@@ -55,7 +55,7 @@ import type { EntityManager } from '@mikro-orm/mysql';
 
   return {
     author: { name: args.name },
-    publishedAt: { $lte: em.raw('now()') },
+    publishedAt: { $lte: raw('now()') },
   };
 } })
 export class Book {

docs/docs/query-builder.md

@@ -403,11 +403,11 @@ console.log(qb4.getQuery());
 
 ## Referring to column in update queries
 
-You can use `qb.raw()` to insert raw SQL snippets like this:
+You can use static `raw()` helper to insert raw SQL snippets like this:
 
 ```ts
 const qb = em.createQueryBuilder(Book);
-qb.update({ price: qb.raw('price + 1') }).where({ uuid: '123' });
+qb.update({ price: raw('price + 1') }).where({ uuid: '123' });
 
 console.log(qb.getQuery());
 // update `book` set `price` = price + 1 where `uuid_pk` = ?

docs/docs/upgrading-v5-to-v6.md

@@ -220,3 +220,13 @@ Use `RequestContext.create` instead, it can be awaited now.
 
 The decorator was renamed to `@CreateRequestContext()` to make it clear it always creates new context, and a new `@EnsureRequestContext()` decorator was added that will reuse existing contexts if available. 
 
+## Removed `em.raw()` and `qb.raw()`
+
+Both removed in favour of new static `raw()` helper, which can be also used to do atomic updates via `flush`:
+
+```ts
+const ref = em.getReference(User, 1);
+ref.age = raw(`age * 2`);
+await em.flush();
+console.log(ref.age); // real value is available after flush
+```

package.json

@@ -90,7 +90,7 @@
     "ignoreDeps": []
   },
   "engines": {
-    "node": ">= 14.0.0",
+    "node": ">= 16.0.0",
     "yarn": ">=3.2.0"
   },
   "devDependencies": {

packages/core/src/typings.ts

@@ -473,6 +473,10 @@ export class EntityMetadata<T = any> {
           return this.__helper.__data[prop.name];
         },
         set(val: unknown) {
+          if (typeof val === 'object' && !!val && '__raw' in val) {
+            (val as Dictionary).use();
+          }
+
           this.__helper.__data[prop.name] = val;
           this.__helper.__touched = true;
         },

packages/core/src/unit-of-work/ChangeSetPersister.ts

@@ -1,5 +1,5 @@
 import type { MetadataStorage } from '../metadata';
-import type { AnyEntity, Dictionary, EntityData, EntityDictionary, EntityMetadata, EntityProperty, FilterQuery, IHydrator, IPrimaryKey } from '../typings';
+import type { AnyEntity, Dictionary, EntityData, EntityDictionary, EntityMetadata, EntityProperty, EntityKey, FilterQuery, IHydrator, IPrimaryKey } from '../typings';
 import { EntityIdentifier, helper, type EntityFactory, type EntityValidator, type Collection } from '../entity';
 import { ChangeSetType, type ChangeSet } from './ChangeSet';
 import type { QueryResult } from '../connections';
@@ -162,7 +162,9 @@ export class ChangeSetPersister {
         this.mapPrimaryKey(meta, res.rows![i][field], changeSet);
       }
 
-      this.mapReturnedValues(changeSet.entity, changeSet.payload, res.rows![i], meta);
+      if (res.rows) {
+        this.mapReturnedValues(changeSet.entity, changeSet.payload, res.rows[i], meta);
+      }
       this.markAsPopulated(changeSet, meta);
       wrapped.__initialized = true;
       wrapped.__managed = true;
@@ -174,6 +176,7 @@ export class ChangeSetPersister {
     const meta = this.metadata.find(changeSet.name)!;
     const res = await this.updateEntity(meta, changeSet, options);
     this.checkOptimisticLock(meta, changeSet, res);
+    this.mapReturnedValues(changeSet.entity, changeSet.payload, res.row, meta);
     await this.reloadVersionValues(meta, [changeSet], options);
     changeSet.persisted = true;
   }
@@ -217,8 +220,15 @@ export class ChangeSetPersister {
       this.checkConcurrencyKeys(meta, changeSet, cond[idx]);
     });
 
-    await this.driver.nativeUpdateMany(meta.className, cond, changeSets.map(cs => cs.payload), options);
-    changeSets.forEach(cs => cs.persisted = true);
+    const res = await this.driver.nativeUpdateMany(meta.className, cond, changeSets.map(cs => cs.payload), options);
+
+    changeSets.forEach((changeSet, idx) => {
+      if (res.rows) {
+        this.mapReturnedValues(changeSet.entity, changeSet.payload, res.rows[idx], meta);
+      }
+
+      changeSet.persisted = true;
+    });
   }
 
   private mapPrimaryKey<T extends object>(meta: EntityMetadata<T>, value: IPrimaryKey, changeSet: ChangeSet<T>): void {
@@ -320,15 +330,30 @@ export class ChangeSetPersister {
   }
 
   /**
-   * This method also handles reloading of database default values for inserts, so we use
-   * a single query in case of both versioning and default values is used.
+   * This method also handles reloading of database default values for inserts and raw property updates,
+   * so we use a single query in case of both versioning and default values is used.
    */
   private async reloadVersionValues<T extends object>(meta: EntityMetadata<T>, changeSets: ChangeSet<T>[], options?: DriverMethodOptions) {
     const reloadProps = meta.versionProperty ? [meta.properties[meta.versionProperty]] : [];
 
     if (changeSets[0].type === ChangeSetType.CREATE) {
       // do not reload things that already had a runtime value
-      reloadProps.push(...meta.props.filter(prop => prop.defaultRaw && prop.defaultRaw.toLowerCase() !== 'null' && changeSets[0].entity[prop.name] == null));
+      meta.hydrateProps
+        .filter(prop => prop.persist !== false && ((prop.primary && prop.autoincrement) || prop.defaultRaw))
+        .filter(prop => (changeSets[0].entity[prop.name] == null && prop.defaultRaw !== 'null') || Utils.isRawSql(changeSets[0].entity[prop.name]))
+        .forEach(prop => reloadProps.push(prop));
+    }
+
+    if (changeSets[0].type === ChangeSetType.UPDATE) {
+      const returning = new Set<EntityProperty<T>>();
+      changeSets.forEach(cs => {
+        Utils.keys(cs.payload).forEach(k => {
+          if (Utils.isRawSql(cs.payload[k]) && Utils.isRawSql(cs.entity[k as EntityKey<T>])) {
+            returning.add(meta.properties[k as EntityKey<T>]);
+          }
+        });
+      });
+      reloadProps.push(...returning);
     }
 
     if (reloadProps.length === 0) {
@@ -395,7 +420,7 @@ export class ChangeSetPersister {
   mapReturnedValues<T extends object>(entity: T, payload: EntityDictionary<T>, row: Dictionary | undefined, meta: EntityMetadata<T>, override = false): void {
     if (this.platform.usesReturningStatement() && row && Utils.hasObjectKeys(row)) {
       const data = meta.props.reduce((ret, prop) => {
-        if (prop.fieldNames && row[prop.fieldNames[0]] != null && (override || entity[prop.name] == null)) {
+        if (prop.fieldNames && row[prop.fieldNames[0]] != null && (override || entity[prop.name] == null || Utils.isRawSql(entity[prop.name]))) {
           ret[prop.name] = row[prop.fieldNames[0]];
         }
 

packages/core/src/utils/QueryHelper.ts

@@ -2,13 +2,13 @@ import { Reference } from '../entity/Reference';
 import { Utils } from './Utils';
 import type {
   Dictionary,
+  EntityKey,
   EntityMetadata,
   EntityProperty,
-  FilterDef,
-  EntityKey,
   EntityValue,
-  FilterQuery,
+  FilterDef,
   FilterKey,
+  FilterQuery,
 } from '../typings';
 import { ARRAY_OPERATORS, GroupOperator, ReferenceKind } from '../enums';
 import type { Platform } from '../platforms';
@@ -316,3 +316,59 @@ export function expr<T = unknown>(sql: (keyof T & string) | (keyof T & string)[]
 
   return sql;
 }
+
+export class RawQueryFragment {
+
+  readonly sql: string;
+  readonly params?: unknown[];
+
+  #used = 0;
+
+  constructor(sql: string, params?: unknown[]) {
+    this.sql = sql;
+
+    if (params) {
+      this.params = params;
+    }
+  }
+
+  valueOf() {
+    throw new Error(`Trying to modify raw SQL fragment: '${this.sql}'`);
+  }
+
+  toJSON() {
+    throw new Error(`Trying to serialize raw SQL fragment: '${this.sql}'`);
+  }
+
+  /** @internal */
+  use() {
+    if (this.#used > 0) {
+      throw new Error(`Cannot reassign already used RawQueryFragment: '${this.sql}'`);
+    }
+
+    this.#used++;
+  }
+
+}
+
+Object.defineProperties(RawQueryFragment.prototype, {
+  __raw: { value: true, enumerable: false },
+  // toString: { value() { throw new Error(`Trying to serialize raw SQL fragment: '${this.sql}'`); }, enumerable: false },
+  // toJSON: { value() { throw new Error(`Trying to serialize raw SQL fragment: '${this.sql}'`); }, enumerable: false },
+});
+
+/**
+ * Creates raw SQL query fragment that can be assigned to a property or part of a filter.
+ */
+export function raw<R = any>(sql: string, params?: unknown[] | Dictionary<unknown>): R {
+  if (typeof params === 'object' && !Array.isArray(params)) {
+    const pairs = Object.entries(params);
+    params = [];
+    for (const [key, value] of pairs) {
+      sql = sql.replace(':' + key, '?');
+      params.push(value);
+    }
+  }
+
+  return new RawQueryFragment(sql, params) as R;
+}

packages/core/src/utils/Utils.ts

@@ -1190,4 +1190,8 @@ export class Utils {
     return Object.entries(obj) as [keyof T, T[keyof T]][];
   }
 
+  static isRawSql(value: unknown): value is { sql: string; params?: unknown[]; use: () => void } {
+    return typeof value === 'object' && !!value && '__raw' in value;
+  }
+
 }

packages/knex/src/AbstractSqlDriver.ts

@@ -212,7 +212,7 @@ export abstract class AbstractSqlDriver<Connection extends AbstractSqlConnection
     const kqb = qb.getKnexQuery().clear('select');
 
     if (type === QueryType.COUNT) {
-      kqb.select(qb.raw('count(*) as count'));
+      kqb.select(this.connection.getKnex().raw('count(*) as count'));
     } else { // select
       kqb.select('*');
     }
@@ -419,9 +419,10 @@ export abstract class AbstractSqlDriver<Connection extends AbstractSqlConnection
       }).join(', ');
     }
 
-    if (this.platform.usesReturningStatement()) {
-      /* istanbul ignore next */
-      const returningProps = meta!.hydrateProps.filter(prop => prop.persist !== false && (prop.primary || prop.defaultRaw || prop.autoincrement));
+    if (meta && this.platform.usesReturningStatement()) {
+      const returningProps = meta.hydrateProps
+        .filter(prop => prop.persist !== false && ((prop.primary && prop.autoincrement) || prop.defaultRaw || prop.autoincrement))
+        .filter(prop => !(prop.name in data[0]) || Utils.isRawSql(data[0][prop.name]));
       const returningFields = Utils.flatten(returningProps.map(prop => prop.fieldNames));
       /* istanbul ignore next */
       sql += returningFields.length > 0 ? ` returning ${returningFields.map(field => this.platform.quoteIdentifier(field)).join(', ')}` : '';
@@ -520,7 +521,16 @@ export abstract class AbstractSqlDriver<Connection extends AbstractSqlConnection
 
     const collections = options.processCollections ? data.map(d => this.extractManyToMany(entityName, d)) : [];
     const keys = new Set<EntityKey<T>>();
-    data.forEach(row => Utils.keys(row).forEach(k => keys.add(k as EntityKey<T>)));
+    const returning = new Set<EntityKey<T>>();
+    data.forEach(row => {
+      Utils.keys(row).forEach(k => {
+        keys.add(k as EntityKey<T>);
+
+        if (Utils.isRawSql(row[k])) {
+          returning.add(k);
+        }
+      });
+    });
     const pkCond = Utils.flatten(meta.primaryKeys.map(pk => meta.properties[pk].fieldNames)).map(pk => `${this.platform.quoteIdentifier(pk)} = ?`).join(' and ');
     const params: any[] = [];
     let sql = `update ${this.getTableName(meta, options)} set `;
@@ -589,6 +599,13 @@ export abstract class AbstractSqlDriver<Connection extends AbstractSqlConnection
       return '?';
     });
     sql += ` in (${conds.join(', ')})`;
+
+    if (this.platform.usesReturningStatement() && returning.size > 0) {
+      const returningFields = Utils.flatten([...returning].map(prop => meta.properties[prop].fieldNames));
+      /* istanbul ignore next */
+      sql += returningFields.length > 0 ? ` returning ${returningFields.map(field => this.platform.quoteIdentifier(field)).join(', ')}` : '';
+    }
+
     const res = await this.rethrow(this.execute<QueryResult<T>>(sql, params, 'run', options.ctx));
 
     for (let i = 0; i < collections.length; i++) {
@@ -861,7 +878,7 @@ export abstract class AbstractSqlDriver<Connection extends AbstractSqlConnection
 
     if (prop.customType?.convertToJSValueSQL && tableAlias) {
       const prefixed = qb.ref(prop.fieldNames[0]).withSchema(tableAlias).toString();
-      return [qb.raw(prop.customType.convertToJSValueSQL(prefixed, this.platform) + ' as ' + aliased).toString()];
+      return [prop.customType.convertToJSValueSQL(prefixed, this.platform) + ' as ' + aliased];
     }
 
     if (prop.formula) {

packages/knex/src/AbstractSqlPlatform.ts

@@ -34,6 +34,10 @@ export abstract class AbstractSqlPlatform extends Platform {
   }
 
   override quoteValue(value: any): string {
+    if (Utils.isRawSql(value)) {
+      return this.formatQuery(value.sql, value.params ?? []);
+    }
+
     if (this.isRaw(value)) {
       return value;
     }

packages/knex/src/SqlEntityManager.ts

@@ -25,16 +25,6 @@ export class SqlEntityManager<D extends AbstractSqlDriver = AbstractSqlDriver> e
     return this.createQueryBuilder(entityName, alias, type);
   }
 
-  /**
-   * Creates raw SQL query that won't be escaped when used as a parameter.
-   */
-  raw<R = Knex.Raw>(sql: string, bindings: Knex.RawBinding[] | Knex.ValueDict = []): R {
-    const raw = this.getKnex().raw(sql, bindings);
-    (raw as Dictionary).__raw = true; // tag it as there is now way to check via `instanceof`
-
-    return raw as unknown as R;
-  }
-
   /**
    * Returns configured knex instance.
    */

packages/knex/src/query/CriteriaNodeFactory.ts

@@ -12,7 +12,7 @@ export class CriteriaNodeFactory {
 
   static createNode<T extends object>(metadata: MetadataStorage, entityName: string, payload: any, parent?: ICriteriaNode<T>, key?: EntityKey<T>): ICriteriaNode<T> {
     const customExpression = CriteriaNode.isCustomExpression(key || '');
-    const scalar = Utils.isPrimaryKey(payload) || payload as unknown instanceof RegExp || payload as unknown instanceof Date || customExpression;
+    const scalar = Utils.isPrimaryKey(payload) || Utils.isRawSql(payload) || payload as unknown instanceof RegExp || payload as unknown instanceof Date || customExpression;
 
     if (Array.isArray(payload) && !scalar) {
       return this.createArrayNode(metadata, entityName, payload, parent, key);

packages/knex/src/query/QueryBuilder.ts

@@ -1,6 +1,7 @@
 import { inspect } from 'util';
 import type { Knex } from 'knex';
 import {
+  raw,
   helper,
   LoadStrategy,
   LockMode,
@@ -194,7 +195,7 @@ export class QueryBuilder<T extends object = AnyEntity> {
     } else if (this.hasToManyJoins()) {
       this._fields = this.mainAlias.metadata!.primaryKeys;
     } else {
-      this._fields = [this.raw('*')];
+      this._fields = [raw('*')];
     }
 
     if (distinct) {
@@ -412,13 +413,6 @@ export class QueryBuilder<T extends object = AnyEntity> {
     return this.knex.ref(field);
   }
 
-  raw<R = Knex.Raw>(sql: string, bindings: Knex.RawBinding[] | Knex.ValueDict = []): R {
-    const raw = this.knex.raw(sql, bindings);
-    (raw as Dictionary).__raw = true; // tag it as there is now way to check via `instanceof`
-
-    return raw as unknown as R;
-  }
-
   limit(limit?: number, offset = 0): this {
     this.ensureNotFinalized();
     this._limit = limit;