[#2108] StatelessSession insertAll in batch does not do batching

Author: dreab8

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

@@ -1779,7 +1779,8 @@ default <T> Uni<T> get(Class<T> entityClass, Object id, LockModeType lockModeTyp
 		Uni<Void> insert(Object entity);
 
 		/**
-		 * Insert multiple rows.
+		 * Insert multiple rows, using the number of the
+		 * given entities as the batch size.
 		 *
 		 * @param entities new transient instances
 		 *

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

@@ -138,7 +138,7 @@ public Uni<Void> insert(Object entity) {
 
 	@Override
 	public Uni<Void> insertAll(Object... entities) {
-		return uni( () -> delegate.reactiveInsertAll( entities ) );
+		return uni( () -> delegate.reactiveInsertAll( entities.length, entities ) );
 	}
 
 	@Override
@@ -163,7 +163,7 @@ public Uni<Void> deleteAll(Object... entities) {
 
 	@Override
 	public Uni<Void> deleteAll(int batchSize, Object... entities) {
-		return uni( () -> delegate.reactiveDeleteAll( entities ) );
+		return uni( () -> delegate.reactiveDeleteAll( batchSize, entities ) );
 	}
 
 	@Override

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

@@ -128,7 +128,7 @@ public class ReactiveStatelessSessionImpl extends StatelessSessionImpl implement
 
 	private final ReactiveConnection reactiveConnection;
 
-	private final ReactiveStatelessSession batchingHelperSession;
+	private final ReactiveStatelessSessionImpl batchingHelperSession;
 
 	private final PersistenceContext persistenceContext;
 
@@ -150,10 +150,9 @@ private ReactiveStatelessSessionImpl(
 			PersistenceContext persistenceContext) {
 		super( factory, options );
 		this.persistenceContext = persistenceContext;
-		Integer batchSize = getConfiguredJdbcBatchSize();
-		reactiveConnection = batchSize == null || batchSize < 2
-				? connection
-				: new BatchingConnection( connection, batchSize );
+		// Setting batch size to 0 because `StatelessSession` does not consider
+		// the value of `hibernate.jdbc.batch_size`
+		reactiveConnection = new BatchingConnection( connection, 0 );
 		batchingHelperSession = this;
 		influencers = new LoadQueryInfluencers( factory );
 	}
@@ -551,9 +550,12 @@ public CompletionStage<Void> reactiveInsertAll(Object... entities) {
 
 	@Override
 	public CompletionStage<Void> reactiveInsertAll(int batchSize, Object... entities) {
+		final Integer jdbcBatchSize = batchingHelperSession.getJdbcBatchSize();
+		batchingHelperSession.setJdbcBatchSize( batchSize );
 		final ReactiveConnection connection = batchingConnection( batchSize );
 		return loop( entities, batchingHelperSession::reactiveInsert )
-				.thenCompose( v -> connection.executeBatch() );
+				.thenCompose( v -> connection.executeBatch() )
+				.whenComplete( (v, throwable) -> batchingHelperSession.setJdbcBatchSize( jdbcBatchSize ) );
 	}
 
 	@Override
@@ -564,9 +566,12 @@ public CompletionStage<Void> reactiveUpdateAll(Object... entities) {
 
 	@Override
 	public CompletionStage<Void> reactiveUpdateAll(int batchSize, Object... entities) {
+		final Integer jdbcBatchSize = batchingHelperSession.getJdbcBatchSize();
+		batchingHelperSession.setJdbcBatchSize( batchSize );
 		final ReactiveConnection connection = batchingConnection( batchSize );
 		return loop( entities, batchingHelperSession::reactiveUpdate )
-				.thenCompose( v -> connection.executeBatch() );
+				.thenCompose( v -> connection.executeBatch() )
+				.whenComplete( (v, throwable) -> batchingHelperSession.setJdbcBatchSize( jdbcBatchSize ) );
 	}
 
 	@Override
@@ -577,9 +582,11 @@ public CompletionStage<Void> reactiveDeleteAll(Object... entities) {
 
 	@Override
 	public CompletionStage<Void> reactiveDeleteAll(int batchSize, Object... entities) {
+		final Integer jdbcBatchSize = batchingHelperSession.getJdbcBatchSize();
+		batchingHelperSession.setJdbcBatchSize( batchSize );
 		final ReactiveConnection connection = batchingConnection( batchSize );
-		return loop( entities, batchingHelperSession::reactiveDelete )
-				.thenCompose( v -> connection.executeBatch() );
+		return loop( entities, batchingHelperSession::reactiveDelete ).thenCompose( v -> connection.executeBatch() )
+				.whenComplete( (v, throwable) -> batchingHelperSession.setJdbcBatchSize( jdbcBatchSize ) );
 	}
 
 
@@ -591,9 +598,12 @@ public CompletionStage<Void> reactiveRefreshAll(Object... entities) {
 
 	@Override
 	public CompletionStage<Void> reactiveRefreshAll(int batchSize, Object... entities) {
+		final Integer jdbcBatchSize = batchingHelperSession.getJdbcBatchSize();
+		batchingHelperSession.setJdbcBatchSize( batchSize );
 		final ReactiveConnection connection = batchingConnection( batchSize );
 		return loop( entities, batchingHelperSession::reactiveRefresh )
-				.thenCompose( v -> connection.executeBatch() );
+				.thenCompose( v -> connection.executeBatch() )
+				.whenComplete( (v, throwable) -> batchingHelperSession.setJdbcBatchSize( jdbcBatchSize ) );
 	}
 
 	private ReactiveConnection batchingConnection(int batchSize) {

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

@@ -1836,7 +1836,8 @@ default <T> CompletionStage<T> get(Class<T> entityClass, Object id, LockModeType
 		CompletionStage<Void> insert(Object entity);
 
 		/**
-		 * Insert multiple rows.
+		 * Insert multiple rows, using the number of the
+		 * given entities as the batch size.
 		 *
 		 * @param entities new transient instances
 		 *

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

@@ -67,7 +67,7 @@ public CompletionStage<Void> insert(Object entity) {
 
 	@Override
 	public CompletionStage<Void> insert(Object... entities) {
-		return delegate.reactiveInsertAll( entities );
+		return delegate.reactiveInsertAll( entities.length, entities );
 	}
 
 	@Override