Add upsertAll

Author: DavideD

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

@@ -1908,6 +1908,9 @@ default <T> Uni<T> get(Class<T> entityClass, Object id, LockModeType lockModeTyp
 		@Incubating
 		Uni<Void> upsert(String entityName, Object entity);
 
+		@Incubating
+		Uni<Void> upsertAll(Object... 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,11 @@ public Uni<Void> upsert(String entityName, Object entity) {
 		return uni( () -> delegate.reactiveUpsert( entityName, entity ) );
 	}
 
+	@Override
+	public Uni<Void> upsertAll(Object... entities) {
+		return uni( () -> delegate.reactiveUpsert( entities ) );
+	}
+
 	@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> reactiveUpsert(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,20 @@ public CompletionStage<Void> reactiveUpsert(String entityName, Object entity) {
 				.mergeReactive( id, state, null, false, null, oldVersion, entity, null, this );
 	}
 
+	@Override
+	public CompletionStage<Void> reactiveUpsert(Object... entities) {
+		return reactiveUpsertAll( entities.length, entities );
+	}
+
+	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/test/java/org/hibernate/ReactiveStatelessDefaultBatchSizeTest.java

@@ -89,7 +89,7 @@ protected void addServices(StandardServiceRegistryBuilder builder) {
 	}
 
 	private static boolean filter(String s) {
-		String[] accepted = { "insert ", "update ", "delete " };
+		String[] accepted = { "merge ", "insert ", "update ", "delete " };
 		for ( String valid : accepted ) {
 			if ( s.toLowerCase().startsWith( valid ) ) {
 				return true;
@@ -98,6 +98,15 @@ private static boolean filter(String s) {
 		return false;
 	}
 
+	@Test
+	public void testMutinyBatchingUpsert(VertxTestContext context) {
+		test( context, getMutinySessionFactory()
+				.withStatelessTransaction( s -> s.upsertAll( PIGS ) )
+				.invoke( () -> assertSqlLogTracker( "merge into pig as t using (.*)" ) )
+				.chain( () -> Uni.createFrom().completionStage( assertExpectedResult( PIGS ) ) )
+		);
+	}
+
 	@Test
 	public void testMutinyBatchingInsert(VertxTestContext context) {
 		test( context, getMutinySessionFactory()