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

Author: dreab8

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

@@ -77,4 +77,8 @@ public interface ReactiveStatelessSession extends ReactiveQueryProducer, Reactiv
 	void close(CompletableFuture<Void> closing);
 
 	Object getIdentifier(Object entity);
+
+	void setJdbcBatchSize(Integer jdbcBatchSize);
+
+	Integer getJdbcBatchSize();
 }

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

@@ -150,9 +150,12 @@ private ReactiveStatelessSessionImpl(
 			PersistenceContext persistenceContext) {
 		super( factory, options );
 		this.persistenceContext = persistenceContext;
+		// StatelessSessionImpl constructor sets the Jdbc Batch Size to 0,
+		// setting it to null allows getConfiguredJdbcBatchSize() to return correct configured size
+		setJdbcBatchSize( null );
 		Integer batchSize = getConfiguredJdbcBatchSize();
 		reactiveConnection = batchSize == null || batchSize < 2
-				? connection
+				? new BatchingConnection( connection, 1 )
 				: new BatchingConnection( connection, batchSize );
 		batchingHelperSession = this;
 		influencers = new LoadQueryInfluencers( factory );
@@ -551,9 +554,16 @@ public CompletionStage<Void> reactiveInsertAll(Object... entities) {
 
 	@Override
 	public CompletionStage<Void> reactiveInsertAll(int batchSize, Object... entities) {
-		final ReactiveConnection connection = batchingConnection( batchSize );
-		return loop( entities, batchingHelperSession::reactiveInsert )
-				.thenCompose( v -> connection.executeBatch() );
+		final Integer jdbcBatchSize = batchingHelperSession.getJdbcBatchSize();
+		batchingHelperSession.setJdbcBatchSize( batchSize );
+		try {
+			final ReactiveConnection connection = batchingConnection( batchSize );
+			return loop( entities, batchingHelperSession::reactiveInsert )
+					.thenCompose( v -> connection.executeBatch() );
+		}
+		finally {
+			batchingHelperSession.setJdbcBatchSize( jdbcBatchSize );
+		}
 	}
 
 	@Override
@@ -564,9 +574,16 @@ public CompletionStage<Void> reactiveUpdateAll(Object... entities) {
 
 	@Override
 	public CompletionStage<Void> reactiveUpdateAll(int batchSize, Object... entities) {
-		final ReactiveConnection connection = batchingConnection( batchSize );
-		return loop( entities, batchingHelperSession::reactiveUpdate )
-				.thenCompose( v -> connection.executeBatch() );
+		final Integer jdbcBatchSize = batchingHelperSession.getJdbcBatchSize();
+		batchingHelperSession.setJdbcBatchSize( batchSize );
+		try {
+			final ReactiveConnection connection = batchingConnection( batchSize );
+			return loop( entities, batchingHelperSession::reactiveUpdate )
+					.thenCompose( v -> connection.executeBatch() );
+		}
+		finally {
+			batchingHelperSession.setJdbcBatchSize( jdbcBatchSize );
+		}
 	}
 
 	@Override
@@ -577,9 +594,16 @@ public CompletionStage<Void> reactiveDeleteAll(Object... entities) {
 
 	@Override
 	public CompletionStage<Void> reactiveDeleteAll(int batchSize, Object... entities) {
-		final ReactiveConnection connection = batchingConnection( batchSize );
-		return loop( entities, batchingHelperSession::reactiveDelete )
-				.thenCompose( v -> connection.executeBatch() );
+		final Integer jdbcBatchSize = batchingHelperSession.getJdbcBatchSize();
+		batchingHelperSession.setJdbcBatchSize( batchSize );
+		try {
+			final ReactiveConnection connection = batchingConnection( batchSize );
+			return loop( entities, batchingHelperSession::reactiveDelete )
+					.thenCompose( v -> connection.executeBatch() );
+		}
+		finally {
+			batchingHelperSession.setJdbcBatchSize( jdbcBatchSize );
+		}
 	}
 
 
@@ -591,9 +615,16 @@ public CompletionStage<Void> reactiveRefreshAll(Object... entities) {
 
 	@Override
 	public CompletionStage<Void> reactiveRefreshAll(int batchSize, Object... entities) {
-		final ReactiveConnection connection = batchingConnection( batchSize );
-		return loop( entities, batchingHelperSession::reactiveRefresh )
-				.thenCompose( v -> connection.executeBatch() );
+		final Integer jdbcBatchSize = batchingHelperSession.getJdbcBatchSize();
+		batchingHelperSession.setJdbcBatchSize( batchSize );
+		try {
+			final ReactiveConnection connection = batchingConnection( batchSize );
+			return loop( entities, batchingHelperSession::reactiveRefresh )
+					.thenCompose( v -> connection.executeBatch() );
+		}
+		finally {
+			batchingHelperSession.setJdbcBatchSize( jdbcBatchSize );
+		}
 	}
 
 	private ReactiveConnection batchingConnection(int batchSize) {