Error when inserting in batch with joined table inheritance

hibernate-reactive-core/src/main/java/org/hibernate/reactive/engine/jdbc/mutation/internal/ReactiveMutationExecutorStandard.java

@@ -5,10 +5,6 @@
  */
 package org.hibernate.reactive.engine.jdbc.mutation.internal;
 
-import java.lang.invoke.MethodHandles;
-import java.sql.SQLException;
-import java.util.concurrent.CompletionStage;
-
 import org.hibernate.engine.jdbc.batch.spi.Batch;
 import org.hibernate.engine.jdbc.mutation.JdbcValueBindings;
 import org.hibernate.engine.jdbc.mutation.OperationResultChecker;
@@ -25,6 +21,7 @@
 import org.hibernate.persister.entity.mutation.EntityTableMapping;
 import org.hibernate.reactive.adaptor.impl.PrepareStatementDetailsAdaptor;
 import org.hibernate.reactive.adaptor.impl.PreparedStatementAdaptor;
+import org.hibernate.reactive.engine.jdbc.ResultsCheckerUtil;
 import org.hibernate.reactive.engine.jdbc.env.internal.ReactiveMutationExecutor;
 import org.hibernate.reactive.generator.values.ReactiveGeneratedValuesMutationDelegate;
 import org.hibernate.reactive.logging.impl.Log;
@@ -37,9 +34,16 @@
 import org.hibernate.sql.model.TableMapping;
 import org.hibernate.sql.model.ValuesAnalysis;
 
+import java.lang.invoke.MethodHandles;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletionStage;
+
 import static org.hibernate.engine.jdbc.mutation.internal.ModelMutationHelper.checkResults;
 import static org.hibernate.reactive.logging.impl.LoggerFactory.make;
 import static org.hibernate.reactive.util.impl.CompletionStages.failedFuture;
+import static org.hibernate.reactive.util.impl.CompletionStages.loop;
 import static org.hibernate.reactive.util.impl.CompletionStages.nullFuture;
 import static org.hibernate.reactive.util.impl.CompletionStages.voidFuture;
 import static org.hibernate.sql.model.ModelMutationLogging.MODEL_MUTATION_LOGGER;
@@ -73,10 +77,64 @@ private ReactiveConnection connection(SharedSessionContractImplementor session)
 	@Override
 	public CompletionStage<Void> performReactiveBatchedOperations(
 			ValuesAnalysis valuesAnalysis,
-			TableInclusionChecker inclusionChecker, OperationResultChecker resultChecker,
+			TableInclusionChecker inclusionChecker,
+			OperationResultChecker resultChecker,
 			SharedSessionContractImplementor session) {
-		return ReactiveMutationExecutor.super
-				.performReactiveBatchedOperations( valuesAnalysis, inclusionChecker, resultChecker, session);
+		final PreparedStatementGroup batchedMutationOperationGroup = getBatchedPreparedStatementGroup();
+		if ( batchedMutationOperationGroup != null ) {
+			final List<PreparedStatementDetails> preparedStatementDetailsList = new ArrayList<>(
+					batchedMutationOperationGroup.getNumberOfStatements() );
+			batchedMutationOperationGroup.forEachStatement( (tableName, statementDetails) -> preparedStatementDetailsList
+					.add( statementDetails ) );
+			return loop( preparedStatementDetailsList, statementDetails -> {
+					  if ( statementDetails == null ) {
+						  return voidFuture();
+					  }
+					  final JdbcValueBindings valueBindings = getJdbcValueBindings();
+					  final TableMapping tableDetails = statementDetails.getMutatingTableDetails();
+					  if ( inclusionChecker != null && !inclusionChecker.include( tableDetails ) ) {
+						  if ( MODEL_MUTATION_LOGGER.isTraceEnabled() ) {
+							  MODEL_MUTATION_LOGGER.tracef(
+									  "Skipping execution of secondary insert : %s",
+									  tableDetails.getTableName()
+							  );
+						  }
+						  return voidFuture();
+					  }
+
+					  // If we get here the statement is needed - make sure it is resolved
+					  final Object[] paramValues = PreparedStatementAdaptor.bind( statement -> {
+						  PreparedStatementDetails details = new PrepareStatementDetailsAdaptor(
+								  statementDetails,
+								  statement,
+								  session.getJdbcServices()
+						  );
+						  valueBindings.beforeStatement( details );
+					  } );
+
+					  final ReactiveConnection reactiveConnection = ( (ReactiveConnectionSupplier) session ).getReactiveConnection();
+					  final String sql = statementDetails.getSqlString();
+					  return reactiveConnection.update(
+							  sql,
+							  paramValues,
+							  true,
+							  (rowCount, batchPosition, query) -> ResultsCheckerUtil.checkResults(
+									  session,
+									  statementDetails,
+									  resultChecker,
+									  rowCount,
+									  batchPosition
+							  )
+					  ).whenComplete( (o, throwable) -> { //TODO: is this part really needed?
+						  if ( statementDetails.getStatement() != null ) {
+							  statementDetails.releaseStatement( session );
+						  }
+						  valueBindings.afterStatement( tableDetails );
+					  } );
+				  }
+			);
+		}
+		return voidFuture();
 	}
 
 	@Override
@@ -159,6 +217,23 @@ public CompletionStage<GeneratedValues> performReactiveNonBatchedOperations(
 		}
 	}
 
+	@Override
+	public CompletionStage<Void> performReactiveSelfExecutingOperations(
+			ValuesAnalysis valuesAnalysis,
+			TableInclusionChecker inclusionChecker,
+			SharedSessionContractImplementor session) {
+		if ( getSelfExecutingMutations() == null || getSelfExecutingMutations().isEmpty() ) {
+			return voidFuture();
+		}
+
+		return loop( getSelfExecutingMutations(), operation -> {
+			if ( inclusionChecker.include( operation.getTableDetails() ) ) {
+				operation.performMutation( getJdbcValueBindings(), valuesAnalysis, session );
+			}
+			return voidFuture();
+		});
+	}
+
 	private class OperationsForEach {
 
 		private final Object id;
@@ -210,6 +285,7 @@ public CompletionStage<Void> buildLoop() {
 			return loop;
 		}
 	}
+
 	@Override
 	public CompletionStage<Void> performReactiveNonBatchedMutation(
 			PreparedStatementDetails statementDetails,

hibernate-reactive-core/src/test/java/org/hibernate/reactive/JoinedInheritanceBatchTest.java

@@ -0,0 +1,130 @@
+/* Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright: Red Hat Inc. and Hibernate Authors
+ */
+package org.hibernate.reactive;
+
+
+import org.junit.jupiter.api.Test;
+
+import io.vertx.junit5.Timeout;
+import io.vertx.junit5.VertxTestContext;
+import jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.persistence.Inheritance;
+import jakarta.persistence.InheritanceType;
+import jakarta.persistence.SequenceGenerator;
+import jakarta.persistence.Table;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CompletionStage;
+
+import static java.util.concurrent.TimeUnit.MINUTES;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.hibernate.reactive.util.impl.CompletionStages.voidFuture;
+
+@Timeout(value = 10, timeUnit = MINUTES)
+public class JoinedInheritanceBatchTest extends BaseReactiveTest {
+
+	@Override
+	protected Collection<Class<?>> annotatedEntities() {
+		return List.of( ClientA.class, Client.class );
+	}
+
+	@Override
+	protected CompletionStage<Void> cleanDb() {
+		return voidFuture();
+	}
+
+	@Test
+	public void test(VertxTestContext context) {
+		final ClientA client1 = new ClientA("Client 1", "email@c1", "123456");
+
+		test( context, getMutinySessionFactory().withTransaction( session -> {
+						  session.setBatchSize( 5 );
+						  return session.persist( client1 );
+					  } )
+					  .chain( () -> getMutinySessionFactory().withTransaction( session -> session
+							  .createQuery( "select c from Client c", Client.class )
+							  .getResultList()
+							  .invoke( persistedClients -> assertThat( persistedClients )
+									  .as( "Clients has not bee persisted" )
+									  .isNotEmpty() ) ) )
+		);
+	}
+
+	@Entity(name = "Client")
+	@Table(name = "`Client`")
+	@Inheritance(strategy = InheritanceType.JOINED)
+	public static class Client {
+
+		@Id
+		@SequenceGenerator(name = "seq", sequenceName = "id_seq", allocationSize = 1)
+		@GeneratedValue(generator = "seq", strategy = GenerationType.SEQUENCE)
+		private Long id;
+
+		private String name;
+
+		private String email;
+
+		private String phone;
+
+		public Client() {
+		}
+
+		public Client(String name, String email, String phone) {
+			this.name = name;
+			this.email = email;
+			this.phone = phone;
+		}
+
+		public Long getId() {
+			return id;
+		}
+
+		public void setId(Long id) {
+			this.id = id;
+		}
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+		public String getEmail() {
+			return email;
+		}
+
+		public void setEmail(String email) {
+			this.email = email;
+		}
+
+		public String getPhone() {
+			return phone;
+		}
+
+		public void setPhone(String phone) {
+			this.phone = phone;
+		}
+
+	}
+
+	@Entity
+	@Table(name = "`ClientA`")
+	public static class ClientA extends Client {
+
+		public ClientA() {
+		}
+
+		public ClientA(String name, String email, String phone) {
+			super( name, email, phone );
+		}
+	}
+
+}