[#2139] Add batching upsert to the StatelessSession

Author: DavideD

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

@@ -1908,6 +1908,42 @@ default <T> Uni<T> get(Class<T> entityClass, Object id, LockModeType lockModeTyp
 		@Incubating
 		Uni<Void> upsert(String entityName, Object entity);
 
+		/**
+		 * Use a SQL {@code merge into} statement to perform
+		 * an upsert on multiple rows using the size of the given array
+		 * as batch size.
+		 *
+		 * @param entities the entities to upsert
+		 *
+		 * @see org.hibernate.StatelessSession#upsert(Object)
+		 */
+		@Incubating
+		Uni<Void> upsertAll(Object... entities);
+
+		/**
+		 * Use a SQL {@code merge into} statement to perform
+		 * an upsert on multiple rows using the specified batch size.
+		 *
+		 * @param batchSize the batch size
+		 * @param entities the list of entities to upsert
+		 *
+		 * @see org.hibernate.StatelessSession#upsert(Object)
+		 */
+		@Incubating
+		Uni<Void> upsertAll(int batchSize, Object... entities);
+
+		/**
+		 * Use a SQL {@code merge into} statement to perform
+		 * an upsert on multiple rows using the size of the given list
+		 * as batch size.
+		 *
+		 * @param entities the entities to upsert
+		 *
+		 * @see org.hibernate.StatelessSession#upsert(Object)
+		 */
+		@Incubating
+		Uni<Void> upsertMultiple(List<?> entities);
+
 		/**
 		 * Refresh the entity instance state from the database.
 		 *

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

@@ -206,6 +206,21 @@ public Uni<Void> upsert(String entityName, Object entity) {
 		return uni( () -> delegate.reactiveUpsert( entityName, entity ) );
 	}
 
+	@Override
+	public Uni<Void> upsertAll(Object... entities) {
+		return uni( () -> delegate.reactiveUpsertAll( entities.length, entities ) );
+	}
+
+	@Override
+	public Uni<Void> upsertAll(int batchSize, Object... entities) {
+		return uni( () -> delegate.reactiveUpsertAll( batchSize, entities ) );
+	}
+
+	@Override
+	public Uni<Void> upsertMultiple(List<?> entities) {
+		return uni( () -> delegate.reactiveUpsertAll( entities.size(), entities.toArray() ) );
+	}
+
 	@Override
 	public Uni<Void> refreshAll(Object... entities) {
 		return uni( () -> delegate.reactiveRefreshAll( entities.length, entities ) );

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

@@ -48,6 +48,8 @@ public interface ReactiveStatelessSession extends ReactiveQueryProducer, Reactiv
 
 	CompletionStage<Void> reactiveUpsert(String entityName, Object entity);
 
+	CompletionStage<Void> reactiveUpsertAll(int batchSize, Object... entities);
+
 	CompletionStage<Void> reactiveRefresh(Object entity);
 
 	CompletionStage<Void> reactiveRefresh(String entityName, Object entity);

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

@@ -511,9 +511,7 @@ public CompletionStage<Void> reactiveUpsert(String entityName, Object 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() );
+			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 );
@@ -542,6 +540,16 @@ public CompletionStage<Void> reactiveUpsert(String entityName, Object entity) {
 				.mergeReactive( id, state, null, false, null, oldVersion, entity, null, this );
 	}
 
+	@Override
+	public CompletionStage<Void> reactiveUpsertAll(int batchSize, Object... entities) {
+		final Integer jdbcBatchSize = batchingHelperSession.getJdbcBatchSize();
+		batchingHelperSession.setJdbcBatchSize( batchSize );
+		final ReactiveConnection connection = batchingConnection( batchSize );
+		return loop( entities, batchingHelperSession::reactiveUpsert )
+				.thenCompose( v -> connection.executeBatch() )
+				.whenComplete( (v, throwable) -> batchingHelperSession.setJdbcBatchSize( jdbcBatchSize ) );
+	}
+
 	@Override
 	public CompletionStage<Void> reactiveInsertAll(Object... entities) {
 		return loop( entities, batchingHelperSession::reactiveInsert )

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

@@ -2021,6 +2021,42 @@ default CompletionStage<Void> refresh(Object entity, LockModeType lockModeType)
 		 */
 		CompletionStage<Void> upsert(String entityName, Object entity);
 
+		/**
+		 * Use a SQL {@code merge into} statement to perform
+		 * an upsert on multiple rows using the size of the given array
+		 * as batch size.
+		 *
+		 * @param entities the entities to upsert
+		 *
+		 * @see org.hibernate.StatelessSession#upsert(Object)
+		 */
+		@Incubating
+		CompletionStage<Void> upsertAll(Object... entities);
+
+		/**
+		 * Use a SQL {@code merge into} statement to perform
+		 * an upsert on multiple rows using the specified batch size.
+		 *
+		 * @param batchSize the batch size
+		 * @param entities the list of entities to upsert
+		 *
+		 * @see org.hibernate.StatelessSession#upsert(Object)
+		 */
+		@Incubating
+		CompletionStage<Void> upsertAll(int batchSize, Object... entities);
+
+		/**
+		 * Use a SQL {@code merge into} statement to perform
+		 * an upsert on multiple rows using the size of the given list
+		 * as batch size.
+		 *
+		 * @param entities the entities to upsert
+		 *
+		 * @see org.hibernate.StatelessSession#upsert(Object)
+		 */
+		@Incubating
+		CompletionStage<Void> upsertMultiple(List<?> entities);
+
 		/**
 		 * Asynchronously fetch an association that's configured for lazy loading.
 		 *

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

@@ -5,10 +5,6 @@
  */
 package org.hibernate.reactive.stage.impl;
 
-import jakarta.persistence.EntityGraph;
-import jakarta.persistence.criteria.CriteriaDelete;
-import jakarta.persistence.criteria.CriteriaQuery;
-import jakarta.persistence.criteria.CriteriaUpdate;
 import org.hibernate.LockMode;
 import org.hibernate.graph.spi.RootGraphImplementor;
 import org.hibernate.query.criteria.JpaCriteriaInsert;
@@ -20,6 +16,10 @@
 import org.hibernate.reactive.stage.Stage.Query;
 import org.hibernate.reactive.stage.Stage.SelectionQuery;
 
+import jakarta.persistence.EntityGraph;
+import jakarta.persistence.criteria.CriteriaDelete;
+import jakarta.persistence.criteria.CriteriaQuery;
+import jakarta.persistence.criteria.CriteriaUpdate;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
@@ -155,6 +155,21 @@ public CompletionStage<Void> upsert(String entityName, Object entity) {
 		return delegate.reactiveUpsert( entityName, entity );
 	}
 
+	@Override
+	public CompletionStage<Void> upsertAll(Object... entities) {
+		return delegate.reactiveUpsertAll( entities.length, entities );
+	}
+
+	@Override
+	public CompletionStage<Void> upsertAll(int batchSize, Object... entities) {
+		return delegate.reactiveUpsertAll( batchSize, entities );
+	}
+
+	@Override
+	public CompletionStage<Void> upsertMultiple(List<?> entities) {
+		return delegate.reactiveUpsertAll( entities.size(), entities.toArray() );
+	}
+
 	@Override
 	public <T> CompletionStage<T> fetch(T association) {
 		return delegate.reactiveFetch( association, false );