[#1702] support StatelessSession.upsert()

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

@@ -1689,6 +1689,23 @@ default <T> Uni<T> get(Class<T> entityClass, Object id, LockModeType lockModeTyp
 		 */
 		Uni<Void> refresh(Object entity);
 
+		/**
+		 *
+		 * @param entity a detached entity instance
+		 *
+		 * @see org.hibernate.StatelessSession#upsert(String, Object)
+		 */
+		Uni<Void> upsert(Object entity);
+
+		/**
+		 *
+		 * @param entityName The entityName for the entity to be merged
+		 * @param entity a detached entity instance
+		 *
+		 * @see org.hibernate.StatelessSession#upsert(String, Object)
+		 */
+		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 upsert( null, entity );
+	}
+
+	@Override
+	public Uni<Void> upsert(String entityName, Object entity) {
+		return uni( () -> delegate.reactiveUpsert( 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

@@ -223,6 +223,20 @@ public CompletionStage<Void> updateReactive(
 				.coordinateReactiveUpdate( object, id, rowId, values, oldVersion, oldValues, dirtyAttributeIndexes, hasDirtyCollection, session );
 	}
 
+	@Override
+	public CompletionStage<Void> mergeReactive(
+			Object id,
+			Object[] fields,
+			int[] dirtyFields,
+			boolean hasDirtyCollection,
+			Object[] oldFields,
+			Object oldVersion,
+			Object object,
+			Object rowId,
+			SharedSessionContractImplementor session) {
+		return CompletionStages.nullFuture();
+	}
+
 	@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/ReactiveMergeCoordinatorStandardScopeFactory.java

@@ -0,0 +1,45 @@
+/* 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.EntityTableMapping;
+import org.hibernate.persister.entity.mutation.UpdateCoordinatorStandard;
+import org.hibernate.reactive.persister.entity.mutation.ReactiveScopedUpdateCoordinator;
+import org.hibernate.reactive.persister.entity.mutation.ReactiveUpdateCoordinator;
+import org.hibernate.reactive.persister.entity.mutation.ReactiveUpdateCoordinatorStandard;
+import org.hibernate.sql.model.MutationOperation;
+import org.hibernate.sql.model.ast.builder.AbstractTableUpdateBuilder;
+import org.hibernate.sql.model.ast.builder.TableMergeBuilder;
+
+public class ReactiveMergeCoordinatorStandardScopeFactory extends UpdateCoordinatorStandard
+		implements ReactiveUpdateCoordinator {
+
+	public ReactiveMergeCoordinatorStandardScopeFactory(
+			AbstractEntityPersister entityPersister,
+			SessionFactoryImplementor factory) {
+		super( entityPersister, factory );
+	}
+
+	@Override
+	public ReactiveScopedUpdateCoordinator makeScopedCoordinator() {
+		return new ReactiveUpdateCoordinatorStandard(
+				entityPersister(),
+				factory(),
+				this.getStaticUpdateGroup(),
+				this.getBatchKey(),
+				this.getVersionUpdateGroup(),
+				this.getVersionUpdateBatchkey()
+		);
+	}
+
+	@Override
+	protected <O extends MutationOperation> AbstractTableUpdateBuilder<O> newTableUpdateBuilder(EntityTableMapping tableMapping) {
+		return new TableMergeBuilder<>( entityPersister(), tableMapping, factory() );
+	}
+
+}

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() );
@@ -315,6 +320,22 @@ public CompletionStage<Void> updateReactive(
 				.coordinateReactiveUpdate( object, id, rowId, values, oldVersion, oldValues, dirtyAttributeIndexes, hasDirtyCollection, session );
 	}
 
+	@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())
+				.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

@@ -338,6 +338,22 @@ public CompletionStage<Void> updateReactive(
 				.coordinateReactiveUpdate( object, id, rowId, values, oldVersion, oldValues, dirtyAttributeIndexes, hasDirtyCollection, session );
 	}
 
+	@Override
+	public CompletionStage<Void> mergeReactive(
+			Object id,
+			Object[] values,
+			int[] dirtyAttributeIndexes,
+			boolean hasDirtyCollection,
+			Object[] oldValues,
+			Object oldVersion,
+			Object object,
+			Object rowId,
+			SharedSessionContractImplementor session) {
+		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 );

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/mutation/ReactiveTableMergeBuilder.java

@@ -0,0 +1,65 @@
+/* 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.mutation;
+
+import java.util.List;
+
+import org.hibernate.engine.spi.SessionFactoryImplementor;
+import org.hibernate.sql.model.MutationOperation;
+import org.hibernate.sql.model.MutationTarget;
+import org.hibernate.sql.model.TableMapping;
+import org.hibernate.sql.model.ast.ColumnValueBinding;
+import org.hibernate.sql.model.ast.MutatingTableReference;
+import org.hibernate.sql.model.ast.RestrictedTableMutation;
+import org.hibernate.sql.model.ast.builder.AbstractTableUpdateBuilder;
+import org.hibernate.sql.model.internal.OptionalTableUpdate;
+import org.hibernate.sql.model.internal.TableUpdateNoSet;
+
+public class ReactiveTableMergeBuilder <O extends MutationOperation> extends AbstractTableUpdateBuilder<O> {
+
+	public ReactiveTableMergeBuilder(
+			MutationTarget<?> mutationTarget,
+			TableMapping tableMapping,
+			SessionFactoryImplementor sessionFactory) {
+		super( mutationTarget, tableMapping, sessionFactory );
+	}
+
+	public ReactiveTableMergeBuilder(
+			MutationTarget<?> mutationTarget,
+			MutatingTableReference tableReference,
+			SessionFactoryImplementor sessionFactory) {
+		super( mutationTarget, tableReference, sessionFactory );
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public RestrictedTableMutation<O> buildMutation() {
+		final List<ColumnValueBinding> valueBindings = combine( getValueBindings(), getKeyBindings(), getLobValueBindings() );
+		if ( valueBindings.isEmpty() ) {
+			return (RestrictedTableMutation<O>) new TableUpdateNoSet( getMutatingTable(), getMutationTarget() );
+		}
+
+		// TODO: add getMergeDetails() (from ORM)
+//		if ( getMutatingTable().getTableMapping().getUpdateDetails().getCustomSql() != null ) {
+//			return (RestrictedTableMutation<O>) new TableUpdateCustomSql(
+//					getMutatingTable(),
+//					getMutationTarget(),
+//					getSqlComment(),
+//					valueBindings,
+//					getKeyRestrictionBindings(),
+//					getOptimisticLockBindings()
+//			);
+//		}
+
+		return (RestrictedTableMutation<O>) new OptionalTableUpdate(
+				getMutatingTable(),
+				getMutationTarget(),
+				valueBindings,
+				getKeyRestrictionBindings(),
+				getOptimisticLockBindings()
+		);
+	}
+}

hibernate-reactive-core/src/main/java/org/hibernate/reactive/session/ReactiveStatelessSession.java

@@ -40,6 +40,8 @@ public interface ReactiveStatelessSession extends ReactiveQueryProducer, Reactiv
 
 	CompletionStage<Void> reactiveUpdate(Object entity);
 
+	CompletionStage<Void> reactiveUpsert(Object entity);
+
 	CompletionStage<Void> reactiveRefresh(Object entity);
 
 	CompletionStage<Void> reactiveRefresh(Object entity, LockMode lockMode);

hibernate-reactive-core/src/main/java/org/hibernate/reactive/session/impl/ReactiveStatelessSessionImpl.java

@@ -12,6 +12,7 @@
 import org.hibernate.HibernateException;
 import org.hibernate.LockMode;
 import org.hibernate.LockOptions;
+import org.hibernate.TransientObjectException;
 import org.hibernate.UnknownEntityTypeException;
 import org.hibernate.UnknownProfileException;
 import org.hibernate.UnresolvableObjectException;
@@ -367,6 +368,43 @@ public CompletionStage<Void> reactiveRefresh(Object entity, LockMode lockMode) {
 				.whenComplete( (v, e) -> getLoadQueryInfluencers().setInternalFetchProfile( previousFetchProfile ) );
 	}
 
+	//TODO: Add javadoc to reference ORM StatelessSessionImpl.upsert()
+	@Override
+	public CompletionStage<Void> reactiveUpsert(Object entity) {
+		final ReactiveEntityPersister persister = getEntityPersister( null, entity );
+		Object id = persister.getIdentifier( entity, this );
+		Boolean knownTransient = persister.isTransient( entity, this );
+		if ( knownTransient!=null && knownTransient ) {
+			throw new TransientObjectException(
+					"Object passed to upsert() has a null identifier: "
+							+ persister.getEntityName() );
+//			final Generator generator = persister.getGenerator();
+//			if ( !generator.generatedOnExecution() ) {
+//				id = ( (BeforeExecutionGenerator) generator).generate( this, entity, null, INSERT );
+//			}
+		}
+		final Object[] state = persister.getValues( entity );
+		final Object oldVersion;
+		if ( persister.isVersioned() ) {
+			oldVersion = persister.getVersion( entity );
+			if ( oldVersion == null ) {
+				if ( seedVersion( entity, state, persister, this ) ) {
+					persister.setValues( entity, state );
+				}
+			}
+			else {
+				final Object newVersion = incrementVersion( entity, oldVersion, persister, this );
+				setVersion( state, newVersion, persister );
+				persister.setValues( entity, state );
+			}
+		}
+		else {
+			oldVersion = null;
+		}
+
+		return persister.mergeReactive( id, state, null, false, null, oldVersion, entity, null, this );
+	}
+
 	@Override
 	public CompletionStage<Void> reactiveInsertAll(Object... entities) {
 		return loop( entities, batchingHelperSession::reactiveInsert )

hibernate-reactive-core/src/main/java/org/hibernate/reactive/stage/Stage.java

@@ -1762,6 +1762,23 @@ default CompletionStage<Void> refresh(Object entity, LockModeType lockModeType)
 			return refresh( entity, convertToLockMode(lockModeType) );
 		}
 
+		/**
+		 *
+		 * @param entity a detached entity instance
+		 *
+		 * @see org.hibernate.StatelessSession#upsert(Object)
+		 */
+		CompletionStage<Void> upsert(Object entity);
+
+		/**
+		 *
+		 * @param entityName The entityName for the entity to be merged
+		 * @param entity a detached entity instance
+		 *
+		 * @see org.hibernate.StatelessSession#upsert(String, Object)
+		 */
+		CompletionStage<Void> upsert(String entityName, Object entity);
+
 		/**
 		 * Asynchronously fetch an association that's configured for lazy loading.
 		 *

hibernate-reactive-core/src/main/java/org/hibernate/reactive/stage/impl/StageStatelessSessionImpl.java

@@ -117,6 +117,16 @@ public CompletionStage<Void> refresh(Object entity, LockMode lockMode) {
 		return delegate.reactiveRefresh( entity, lockMode );
 	}
 
+	@Override
+	public CompletionStage<Void> upsert(Object entity) {
+		return upsert( null, entity );
+	}
+
+	@Override
+	public CompletionStage<Void> upsert(String entityName, Object entity) {
+		return delegate.reactiveUpsert( entity );
+	}
+
 	@Override
 	public <T> CompletionStage<T> fetch(T association) {
 		return delegate.reactiveFetch( association, false );