Add support for StatelessSession#upsert

README.md

@@ -37,7 +37,7 @@ Hibernate Reactive has been tested with:
 - CockroachDB 22.1
 - MS SQL Server 2019
 - Oracle 21.3
-- [Hibernate ORM][] 6.3.0.Final
+- [Hibernate ORM][] 6.3.1.Final
 - [Vert.x Reactive PostgreSQL Client](https://vertx.io/docs/vertx-pg-client/java/) 4.4.5
 - [Vert.x Reactive MySQL Client](https://vertx.io/docs/vertx-mysql-client/java/) 4.4.5
 - [Vert.x Reactive Db2 Client](https://vertx.io/docs/vertx-db2-client/java/) 4.4.5

build.gradle

@@ -53,7 +53,7 @@ version = projectVersion
 // ./gradlew clean build -PhibernateOrmVersion=5.6.15-SNAPSHOT
 ext {
     if ( !project.hasProperty('hibernateOrmVersion') ) {
-        hibernateOrmVersion = '6.3.0.Final'
+        hibernateOrmVersion = '6.3.1.Final'
     }
     if ( !project.hasProperty( 'hibernateOrmGradlePluginVersion' ) ) {
         // Same as ORM as default

hibernate-reactive-core/src/main/java/org/hibernate/reactive/engine/impl/ReactiveEntityDeleteAction.java

@@ -64,12 +64,14 @@ public CompletionStage<Void> reactiveExecute() throws HibernateException {
 		final boolean veto = isInstanceLoaded() && preDelete();
 
 		final Object ck = lockCacheItem();
-
-		final CompletionStage<Void> deleteStep = !isCascadeDeleteEnabled() && !veto
-				? ( (ReactiveEntityPersister) persister ).deleteReactive( id, version, instance, session )
-				: voidFuture();
-
-		return deleteStep.thenAccept( v -> {
+		return deleteStep(
+				veto,
+				(ReactiveEntityPersister) persister,
+				id,
+				version,
+				instance,
+				session
+		).thenAccept( v -> {
 			if ( isInstanceLoaded() ) {
 				postDeleteLoaded( id, persister, session, instance, ck );
 			}
@@ -84,4 +86,16 @@ public CompletionStage<Void> reactiveExecute() throws HibernateException {
 			}
 		} );
 	}
+
+	private CompletionStage<Void> deleteStep(
+			boolean veto,
+			ReactiveEntityPersister persister,
+			Object id,
+			Object version,
+			Object instance,
+			SharedSessionContractImplementor session) {
+		return !isCascadeDeleteEnabled() && !veto
+				? persister.deleteReactive( id, version, instance, session )
+				: voidFuture();
+	}
 }

hibernate-reactive-core/src/main/java/org/hibernate/reactive/mutiny/Mutiny.java

@@ -1736,6 +1736,28 @@ default <T> Uni<T> get(Class<T> entityClass, Object id, LockModeType lockModeTyp
 		 */
 		Uni<Void> refresh(Object entity);
 
+		/**
+		 * Use a SQL {@code merge into} statement to perform an upsert.
+		 *
+		 * @param entity a detached entity instance
+		 *
+		 * @see org.hibernate.StatelessSession#upsert(Object)
+		 */
+		@Incubating
+		Uni<Void> upsert(Object entity);
+
+		/**
+		 * Use a SQL {@code merge into} statement to perform an upsert.
+		 *
+		 * @param entityName The entityName for the entity to be merged
+		 * @param entity a detached entity instance
+		 * @throws org.hibernate.TransientObjectException is the entity is transient
+		 *
+		 * @see org.hibernate.StatelessSession#upsert(String, Object)
+		 */
+		@Incubating
+		Uni<Void> upsert(String entityName, Object entity);
+
 		/**
 		 * Refresh the entity instance state from the database.
 		 *

hibernate-reactive-core/src/main/java/org/hibernate/reactive/mutiny/impl/MutinyStatelessSessionImpl.java

@@ -163,6 +163,16 @@ public Uni<Void> refresh(Object entity) {
 		return uni( () -> delegate.reactiveRefresh( entity ) );
 	}
 
+	@Override
+	public Uni<Void> upsert(Object entity) {
+		return uni( () -> delegate.reactiveUpsert( entity ) );
+	}
+
+	@Override
+	public Uni<Void> upsert(String entityName, Object entity) {
+		return uni( () -> delegate.reactiveUpsert( entityName, entity ) );
+	}
+
 	@Override
 	public Uni<Void> refreshAll(Object... entities) {
 		return uni( () -> delegate.reactiveRefreshAll( entities ) );

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/impl/ReactiveCoordinatorFactory.java

@@ -44,4 +44,20 @@ public static ReactiveDeleteCoordinator buildDeleteCoordinator(
 			SessionFactoryImplementor factory) {
 		return new ReactiveDeleteCoordinator( entityPersister, factory );
 	}
+
+	public static ReactiveUpdateCoordinator buildMergeCoordinator(
+			AbstractEntityPersister entityPersister,
+			SessionFactoryImplementor factory) {
+		// we only have updates to issue for entities with one or more singular attributes
+		final AttributeMappingsList attributeMappings = entityPersister.getAttributeMappings();
+		for ( int i = 0; i < attributeMappings.size(); i++ ) {
+			AttributeMapping attributeMapping = attributeMappings.get( i );
+			if ( attributeMapping instanceof SingularAttributeMapping ) {
+				return new ReactiveMergeCoordinatorStandardScopeFactory( entityPersister, factory );
+			}
+		}
+
+		// otherwise, nothing to update
+		return new ReactiveUpdateCoordinatorNoOp( entityPersister );
+	}
 }

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/impl/ReactiveEntityPersister.java

@@ -63,6 +63,22 @@ CompletionStage<Void> updateReactive(
 			final Object rowId,
 			final SharedSessionContractImplementor session);
 
+	/**
+	 * Update the given instance state without blocking.
+	 *
+	 * @see EntityPersister#merge(Object, Object[], int[], boolean, Object[], Object, Object, Object, SharedSessionContractImplementor)
+	 */
+	CompletionStage<Void> mergeReactive(
+			final Object id,
+			final Object[] fields,
+			final int[] dirtyFields,
+			final boolean hasDirtyCollection,
+			final Object[] oldFields,
+			final Object oldVersion,
+			final Object object,
+			final Object rowId,
+			final SharedSessionContractImplementor session);
+
 	/**
 	 * Obtain a pessimistic lock without blocking
 	 */

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/impl/ReactiveJoinedSubclassEntityPersister.java

@@ -5,6 +5,10 @@
  */
 package org.hibernate.reactive.persister.entity.impl;
 
+import java.sql.PreparedStatement;
+import java.util.List;
+import java.util.concurrent.CompletionStage;
+
 import org.hibernate.FetchMode;
 import org.hibernate.HibernateException;
 import org.hibernate.LockMode;
@@ -36,8 +40,8 @@
 import org.hibernate.persister.entity.mutation.DeleteCoordinator;
 import org.hibernate.persister.entity.mutation.InsertCoordinator;
 import org.hibernate.persister.entity.mutation.UpdateCoordinator;
-import org.hibernate.reactive.loader.ast.internal.ReactiveSingleIdArrayLoadPlan;
 import org.hibernate.property.access.spi.PropertyAccess;
+import org.hibernate.reactive.loader.ast.internal.ReactiveSingleIdArrayLoadPlan;
 import org.hibernate.reactive.loader.ast.spi.ReactiveSingleUniqueKeyEntityLoader;
 import org.hibernate.reactive.persister.entity.mutation.ReactiveDeleteCoordinator;
 import org.hibernate.reactive.persister.entity.mutation.ReactiveInsertCoordinator;
@@ -51,11 +55,6 @@
 import org.hibernate.sql.results.graph.entity.internal.EntityResultJoinedSubclassImpl;
 import org.hibernate.type.EntityType;
 
-import java.sql.PreparedStatement;
-import java.util.List;
-import java.util.concurrent.CompletionStage;
-
-
 /**
  * An {@link ReactiveEntityPersister} backed by {@link JoinedSubclassEntityPersister}
  * and {@link ReactiveAbstractEntityPersister}.
@@ -223,6 +222,29 @@ public CompletionStage<Void> updateReactive(
 				.coordinateReactiveUpdate( object, id, rowId, values, oldVersion, oldValues, dirtyAttributeIndexes, hasDirtyCollection, session );
 	}
 
+	/**
+	 * Merge an Object
+	 *
+	 * @see #merge(Object, Object[], int[], boolean, Object[], Object, Object, Object, SharedSessionContractImplementor)
+	 */
+	@Override
+	public CompletionStage<Void> mergeReactive(
+			Object id,
+			Object[] values,
+			int[] dirtyAttributeIndexes,
+			boolean hasDirtyCollection,
+			Object[] oldValues,
+			Object oldVersion,
+			Object object,
+			Object rowId,
+			SharedSessionContractImplementor session) {
+		// This is different from Hibernate ORM because our reactive update coordinator cannot be share among
+		// multiple update operations
+		return ( (ReactiveUpdateCoordinator) getMergeCoordinator() )
+				.makeScopedCoordinator()
+				.coordinateReactiveUpdate( object, id, rowId, values, oldVersion, oldValues, dirtyAttributeIndexes, hasDirtyCollection, session );
+	}
+
 	@Override
 	public <K> CompletionStage<? extends List<?>> reactiveMultiLoad(K[] ids, EventSource session, MultiIdLoadOptions loadOptions) {
 		return reactiveDelegate.multiLoad( ids, session, loadOptions );
@@ -248,6 +270,23 @@ public void update(
 		throw LOG.nonReactiveMethodCall( "updateReactive" );
 	}
 
+	/**
+	 * @see #mergeReactive(Object, Object[], int[], boolean, Object[], Object, Object, Object, SharedSessionContractImplementor)
+	 */
+	@Override
+	public void merge(
+			Object id,
+			Object[] values,
+			int[] dirtyAttributeIndexes,
+			boolean hasDirtyCollection,
+			Object[] oldValues,
+			Object oldVersion,
+			Object object,
+			Object rowId,
+			SharedSessionContractImplementor session) throws HibernateException {
+		throw LOG.nonReactiveMethodCall( "mergeReactive" );
+	}
+
 	@Override
 	public boolean check(int rows, Object id, int tableNumber, Expectation expectation, PreparedStatement statement, String sql) throws HibernateException {
 		return super.check(rows, id, tableNumber, expectation, statement, sql);

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/impl/ReactiveMergeCoordinatorStandardScopeFactory.java

@@ -0,0 +1,33 @@
+/* Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright: Red Hat Inc. and Hibernate Authors
+ */
+package org.hibernate.reactive.persister.entity.impl;
+
+import org.hibernate.engine.spi.SessionFactoryImplementor;
+import org.hibernate.persister.entity.AbstractEntityPersister;
+import org.hibernate.persister.entity.mutation.MergeCoordinator;
+import org.hibernate.reactive.persister.entity.mutation.ReactiveMergeCoordinator;
+import org.hibernate.reactive.persister.entity.mutation.ReactiveScopedUpdateCoordinator;
+import org.hibernate.reactive.persister.entity.mutation.ReactiveUpdateCoordinator;
+
+public class ReactiveMergeCoordinatorStandardScopeFactory extends MergeCoordinator
+		implements ReactiveUpdateCoordinator {
+
+	public ReactiveMergeCoordinatorStandardScopeFactory(AbstractEntityPersister entityPersister, SessionFactoryImplementor factory) {
+		super( entityPersister, factory );
+	}
+
+	@Override
+	public ReactiveScopedUpdateCoordinator makeScopedCoordinator() {
+		return new ReactiveMergeCoordinator(
+				entityPersister(),
+				factory(),
+				this.getStaticUpdateGroup(),
+				this.getBatchKey(),
+				this.getVersionUpdateGroup(),
+				this.getVersionUpdateBatchkey()
+		);
+	}
+}

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/impl/ReactiveSingleTableEntityPersister.java

@@ -102,6 +102,11 @@ protected DeleteCoordinator buildDeleteCoordinator() {
 		return ReactiveCoordinatorFactory.buildDeleteCoordinator( this, getFactory() );
 	}
 
+	@Override
+	protected UpdateCoordinator buildMergeCoordinator() {
+		return ReactiveCoordinatorFactory.buildMergeCoordinator( this, getFactory() );
+	}
+
 	@Override
 	public Generator getGenerator() throws HibernateException {
 		return reactiveDelegate.reactive( super.getGenerator() );
@@ -218,6 +223,20 @@ public void update(
 		throw LOG.nonReactiveMethodCall( "updateReactive" );
 	}
 
+	@Override
+	public void merge(
+			Object id,
+			Object[] values,
+			int[] dirtyAttributeIndexes,
+			boolean hasDirtyCollection,
+			Object[] oldValues,
+			Object oldVersion,
+			Object object,
+			Object rowId,
+			SharedSessionContractImplementor session) throws HibernateException {
+		throw LOG.nonReactiveMethodCall( "mergeReactive" );
+	}
+
 	/**
 	 * Process properties generated with an insert
 	 *
@@ -315,6 +334,29 @@ public CompletionStage<Void> updateReactive(
 				.coordinateReactiveUpdate( object, id, rowId, values, oldVersion, oldValues, dirtyAttributeIndexes, hasDirtyCollection, session );
 	}
 
+	/**
+	 * Merge an object
+	 *
+	 * @see SingleTableEntityPersister#merge(Object, Object[], int[], boolean, Object[], Object, Object, Object, SharedSessionContractImplementor)
+	 */
+	@Override
+	public CompletionStage<Void> mergeReactive(
+			final Object id,
+			final Object[] values,
+			int[] dirtyAttributeIndexes,
+			final boolean hasDirtyCollection,
+			final Object[] oldValues,
+			final Object oldVersion,
+			final Object object,
+			final Object rowId,
+			SharedSessionContractImplementor session) {
+		return ( (ReactiveUpdateCoordinator) getMergeCoordinator() )
+				// This is different from Hibernate ORM because our reactive update coordinator cannot be share among
+				// multiple update operations
+				.makeScopedCoordinator()
+				.coordinateReactiveUpdate( object, id, rowId, values, oldVersion, oldValues, dirtyAttributeIndexes, hasDirtyCollection, session );
+	}
+
 	@Override
 	public <K> CompletionStage<? extends List<?>> reactiveMultiLoad(K[] ids, EventSource session, MultiIdLoadOptions loadOptions) {
 		return reactiveDelegate.multiLoad( ids, session, loadOptions );

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/impl/ReactiveUnionSubclassEntityPersister.java

@@ -245,6 +245,23 @@ public void update(
 		throw LOG.nonReactiveMethodCall( "updateReactive" );
 	}
 
+	/**
+	 * @see #mergeReactive(Object, Object[], int[], boolean, Object[], Object, Object, Object, SharedSessionContractImplementor)
+	 */
+	@Override
+	public void merge(
+			Object id,
+			Object[] values,
+			int[] dirtyAttributeIndexes,
+			boolean hasDirtyCollection,
+			Object[] oldValues,
+			Object oldVersion,
+			Object object,
+			Object rowId,
+			SharedSessionContractImplementor session) throws HibernateException {
+		throw LOG.nonReactiveMethodCall( "mergeReactive" );
+	}
+
 	/**
 	 * Process properties generated with an insert
 	 *
@@ -338,6 +355,27 @@ public CompletionStage<Void> updateReactive(
 				.coordinateReactiveUpdate( object, id, rowId, values, oldVersion, oldValues, dirtyAttributeIndexes, hasDirtyCollection, session );
 	}
 
+	/**
+	 * @see #merge(Object, Object[], int[], boolean, Object[], Object, Object, Object, SharedSessionContractImplementor)
+	 */
+	@Override
+	public CompletionStage<Void> mergeReactive(
+			Object id,
+			Object[] values,
+			int[] dirtyAttributeIndexes,
+			boolean hasDirtyCollection,
+			Object[] oldValues,
+			Object oldVersion,
+			Object object,
+			Object rowId,
+			SharedSessionContractImplementor session) {
+		// This is different from Hibernate ORM because our reactive update coordinator cannot be share among
+		// multiple update operations
+		return ( (ReactiveUpdateCoordinator) getMergeCoordinator() )
+				.makeScopedCoordinator()
+				.coordinateReactiveUpdate( object, id, rowId, values, oldVersion, oldValues, dirtyAttributeIndexes, hasDirtyCollection, session );
+	}
+
 	@Override
 	public <K> CompletionStage<? extends List<?>> reactiveMultiLoad(K[] ids, EventSource session, MultiIdLoadOptions loadOptions) {
 		return reactiveDelegate.multiLoad( ids, session, loadOptions );