Add support for @SoftDelete

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/impl/ReactiveCoordinatorFactory.java

@@ -9,8 +9,11 @@
 import org.hibernate.metamodel.mapping.AttributeMapping;
 import org.hibernate.metamodel.mapping.AttributeMappingsList;
 import org.hibernate.metamodel.mapping.SingularAttributeMapping;
+import org.hibernate.metamodel.mapping.SoftDeleteMapping;
 import org.hibernate.persister.entity.AbstractEntityPersister;
+import org.hibernate.persister.entity.mutation.DeleteCoordinator;
 import org.hibernate.reactive.persister.entity.mutation.ReactiveDeleteCoordinator;
+import org.hibernate.reactive.persister.entity.mutation.ReactiveDeleteCoordinatorSoft;
 import org.hibernate.reactive.persister.entity.mutation.ReactiveInsertCoordinatorStandard;
 import org.hibernate.reactive.persister.entity.mutation.ReactiveUpdateCoordinator;
 import org.hibernate.reactive.persister.entity.mutation.ReactiveUpdateCoordinatorNoOp;
@@ -39,10 +42,13 @@ public static ReactiveUpdateCoordinator buildUpdateCoordinator(
 		return new ReactiveUpdateCoordinatorNoOp( entityPersister );
 	}
 
-	public static ReactiveDeleteCoordinator buildDeleteCoordinator(
+	public static DeleteCoordinator buildDeleteCoordinator(
+			SoftDeleteMapping softDeleteMapping,
 			AbstractEntityPersister entityPersister,
 			SessionFactoryImplementor factory) {
-		return new ReactiveDeleteCoordinator( entityPersister, factory );
+		return softDeleteMapping != null
+				? new ReactiveDeleteCoordinatorSoft( entityPersister, factory )
+				: new ReactiveDeleteCoordinator( entityPersister, factory );
 	}
 
 	public static ReactiveUpdateCoordinator buildMergeCoordinator(

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/impl/ReactiveJoinedSubclassEntityPersister.java

@@ -152,7 +152,7 @@ protected UpdateCoordinator buildUpdateCoordinator() {
 
 	@Override
 	protected DeleteCoordinator buildDeleteCoordinator() {
-		return ReactiveCoordinatorFactory.buildDeleteCoordinator( this, getFactory() );
+		return ReactiveCoordinatorFactory.buildDeleteCoordinator( super.getSoftDeleteMapping(), this, getFactory() );
 	}
 
 	@Override

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/impl/ReactiveSingleTableEntityPersister.java

@@ -9,7 +9,6 @@
 import java.util.List;
 import java.util.concurrent.CompletionStage;
 
-
 import org.hibernate.FetchMode;
 import org.hibernate.HibernateException;
 import org.hibernate.LockMode;
@@ -49,7 +48,7 @@
 import org.hibernate.reactive.generator.values.ReactiveInsertGeneratedIdentifierDelegate;
 import org.hibernate.reactive.loader.ast.internal.ReactiveSingleIdArrayLoadPlan;
 import org.hibernate.reactive.loader.ast.spi.ReactiveSingleUniqueKeyEntityLoader;
-import org.hibernate.reactive.persister.entity.mutation.ReactiveDeleteCoordinator;
+import org.hibernate.reactive.persister.entity.mutation.ReactiveAbstractDeleteCoordinator;
 import org.hibernate.reactive.persister.entity.mutation.ReactiveInsertCoordinatorStandard;
 import org.hibernate.reactive.persister.entity.mutation.ReactiveUpdateCoordinator;
 import org.hibernate.reactive.util.impl.CompletionStages;
@@ -115,7 +114,7 @@ protected InsertCoordinator buildInsertCoordinator() {
 
 	@Override
 	protected DeleteCoordinator buildDeleteCoordinator() {
-		return ReactiveCoordinatorFactory.buildDeleteCoordinator( this, getFactory() );
+		return ReactiveCoordinatorFactory.buildDeleteCoordinator( super.getSoftDeleteMapping(), this, getFactory() );
 	}
 
 	@Override
@@ -357,7 +356,7 @@ public CompletionStage<GeneratedValues> insertReactive(Object id, Object[] field
 
 	@Override
 	public CompletionStage<Void> deleteReactive(Object id, Object version, Object entity, SharedSessionContractImplementor session) {
-		return ( (ReactiveDeleteCoordinator) getDeleteCoordinator() ).reactiveDelete( entity, id, version, session );
+		return ( (ReactiveAbstractDeleteCoordinator) getDeleteCoordinator() ).reactiveDelete( entity, id, version, session );
 	}
 
 	/**

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/impl/ReactiveUnionSubclassEntityPersister.java

@@ -170,7 +170,7 @@ protected UpdateCoordinator buildUpdateCoordinator() {
 
 	@Override
 	protected DeleteCoordinator buildDeleteCoordinator() {
-		return ReactiveCoordinatorFactory.buildDeleteCoordinator( this, getFactory() );
+		return ReactiveCoordinatorFactory.buildDeleteCoordinator( super.getSoftDeleteMapping(), this, getFactory() );
 	}
 
 	@Override

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/mutation/ReactiveAbstractDeleteCoordinator.java

@@ -0,0 +1,21 @@
+/* Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright: Red Hat Inc. and Hibernate Authors
+ */
+package org.hibernate.reactive.persister.entity.mutation;
+
+import java.util.concurrent.CompletionStage;
+
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
+
+/**
+ * With this interface we can have multiple delete coordinators that extend {@link org.hibernate.persister.entity.mutation.AbstractDeleteCoordinator}.
+ *
+ * @see ReactiveDeleteCoordinatorSoft
+ * @see ReactiveDeleteCoordinator
+ */
+public interface ReactiveAbstractDeleteCoordinator {
+
+	CompletionStage<Void> reactiveDelete(Object entity, Object id, Object version, SharedSessionContractImplementor session);
+}

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/mutation/ReactiveDeleteCoordinator.java

@@ -30,7 +30,7 @@
 import static org.hibernate.reactive.util.impl.CompletionStages.failedFuture;
 import static org.hibernate.reactive.util.impl.CompletionStages.voidFuture;
 
-public class ReactiveDeleteCoordinator extends DeleteCoordinatorStandard {
+public class ReactiveDeleteCoordinator extends DeleteCoordinatorStandard implements ReactiveAbstractDeleteCoordinator {
 
 	private static final Log LOG = LoggerFactory.make( Log.class, MethodHandles.lookup() );
 
@@ -45,6 +45,7 @@ public void delete(Object entity, Object id, Object version, SharedSessionContra
 		throw LOG.nonReactiveMethodCall( "coordinateReactiveDelete" );
 	}
 
+	@Override
 	public CompletionStage<Void> reactiveDelete(Object entity, Object id, Object version, SharedSessionContractImplementor session) {
 		try {
 			super.delete( entity, id, version, session );
@@ -142,7 +143,6 @@ protected void doStaticDelete(Object entity, Object id, Object rowId, Object[] l
 		final JdbcValueBindings jdbcValueBindings = mutationExecutor.getJdbcValueBindings();
 		bindPartitionColumnValueBindings( loadedState, session, jdbcValueBindings );
 		applyId( id, rowId, mutationExecutor, getStaticMutationOperationGroup(), session );
-		String[] identifierColumnNames = entityPersister().getIdentifierColumnNames();
 		mutationExecutor.executeReactive(
 						entity,
 						null,

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/mutation/ReactiveDeleteCoordinatorSoft.java

@@ -0,0 +1,186 @@
+/* Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright: Red Hat Inc. and Hibernate Authors
+ */
+package org.hibernate.reactive.persister.entity.mutation;
+
+import java.lang.invoke.MethodHandles;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+
+import org.hibernate.engine.jdbc.mutation.JdbcValueBindings;
+import org.hibernate.engine.jdbc.mutation.MutationExecutor;
+import org.hibernate.engine.jdbc.mutation.group.PreparedStatementDetails;
+import org.hibernate.engine.jdbc.mutation.spi.MutationExecutorService;
+import org.hibernate.engine.spi.SessionFactoryImplementor;
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.persister.entity.AbstractEntityPersister;
+import org.hibernate.persister.entity.mutation.DeleteCoordinatorSoft;
+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.env.internal.ReactiveMutationExecutor;
+import org.hibernate.reactive.logging.impl.Log;
+import org.hibernate.reactive.logging.impl.LoggerFactory;
+import org.hibernate.sql.model.MutationOperation;
+import org.hibernate.sql.model.MutationOperationGroup;
+
+import static org.hibernate.engine.jdbc.mutation.internal.ModelMutationHelper.identifiedResultsCheck;
+import static org.hibernate.reactive.util.impl.CompletionStages.failedFuture;
+import static org.hibernate.reactive.util.impl.CompletionStages.voidFuture;
+
+public class ReactiveDeleteCoordinatorSoft extends DeleteCoordinatorSoft implements ReactiveAbstractDeleteCoordinator {
+
+	private static final Log LOG = LoggerFactory.make( Log.class, MethodHandles.lookup() );
+
+	private CompletionStage<Void> stage;
+
+	public ReactiveDeleteCoordinatorSoft(
+			AbstractEntityPersister entityPersister,
+			SessionFactoryImplementor factory) {
+		super( entityPersister, factory );
+	}
+
+	@Override
+	public void delete(Object entity, Object id, Object version, SharedSessionContractImplementor session) {
+		throw LOG.nonReactiveMethodCall( "coordinateReactiveDelete" );
+	}
+
+	@Override
+	public CompletionStage<Void> reactiveDelete(Object entity, Object id, Object version, SharedSessionContractImplementor session) {
+		try {
+			super.delete( entity, id, version, session );
+			return stage != null ? stage : voidFuture();
+		}
+		catch (Throwable t) {
+			if ( stage == null ) {
+				return failedFuture( t );
+			}
+			stage.toCompletableFuture().completeExceptionally( t );
+			return stage;
+		}
+	}
+
+	@Override
+	protected void doDynamicDelete(Object entity, Object id, Object rowId, Object[] loadedState, SharedSessionContractImplementor session) {
+		stage = new CompletableFuture<>();
+		final MutationOperationGroup operationGroup = generateOperationGroup( null, loadedState, true, session );
+		final ReactiveMutationExecutor mutationExecutor = mutationExecutor( session, operationGroup );
+
+		for ( int i = 0; i < operationGroup.getNumberOfOperations(); i++ ) {
+			final MutationOperation mutation = operationGroup.getOperation( i );
+			if ( mutation != null ) {
+				final String tableName = mutation.getTableDetails().getTableName();
+				mutationExecutor.getPreparedStatementDetails( tableName );
+			}
+		}
+		applyDynamicDeleteTableDetails( id, rowId, loadedState, mutationExecutor, operationGroup, session );
+		mutationExecutor.executeReactive(
+						entity,
+						null,
+						null,
+						(statementDetails, affectedRowCount, batchPosition) -> identifiedResultsCheck(
+								statementDetails,
+								affectedRowCount,
+								batchPosition,
+								entityPersister(),
+								id,
+								factory()
+						),
+						session
+				)
+				.whenComplete( (o, t) -> mutationExecutor.release() )
+				.whenComplete( this::complete );
+	}
+
+	@Override
+	protected void applyId(
+			Object id,
+			Object rowId,
+			MutationExecutor mutationExecutor,
+			MutationOperationGroup operationGroup,
+			SharedSessionContractImplementor session) {
+		final JdbcValueBindings jdbcValueBindings = mutationExecutor.getJdbcValueBindings();
+
+		for ( int position = 0; position < operationGroup.getNumberOfOperations(); position++ ) {
+			final MutationOperation jdbcMutation = operationGroup.getOperation( position );
+			final EntityTableMapping tableDetails = (EntityTableMapping) jdbcMutation.getTableDetails();
+			breakDownKeyJdbcValues( id, rowId, session, jdbcValueBindings, tableDetails );
+			final PreparedStatementDetails statementDetails = mutationExecutor.getPreparedStatementDetails( tableDetails.getTableName() );
+			if ( statementDetails != null ) {
+				PreparedStatementAdaptor.bind( statement -> {
+					PrepareStatementDetailsAdaptor detailsAdaptor = new PrepareStatementDetailsAdaptor(
+							statementDetails,
+							statement,
+							session.getJdbcServices()
+					);
+					// force creation of the PreparedStatement
+					//noinspection resource
+					detailsAdaptor.resolveStatement();
+				} );
+			}
+		}
+	}
+
+	@Override
+	protected void doStaticDelete(Object entity, Object id, Object rowId, Object[] loadedState, Object version, SharedSessionContractImplementor session) {
+		stage = new CompletableFuture<>();
+		final boolean applyVersion = entity != null;
+		final MutationOperationGroup operationGroupToUse = entity == null
+				? resolveNoVersionDeleteGroup( session )
+				: getStaticMutationOperationGroup();
+
+		final ReactiveMutationExecutor mutationExecutor = mutationExecutor( session, operationGroupToUse );
+		for ( int position = 0; position < getStaticMutationOperationGroup().getNumberOfOperations(); position++ ) {
+			final MutationOperation mutation = getStaticMutationOperationGroup().getOperation( position );
+			if ( mutation != null ) {
+				mutationExecutor.getPreparedStatementDetails( mutation.getTableDetails().getTableName() );
+			}
+		}
+
+		if ( applyVersion ) {
+			applyLocking( version, null, mutationExecutor, session );
+		}
+		final JdbcValueBindings jdbcValueBindings = mutationExecutor.getJdbcValueBindings();
+		bindPartitionColumnValueBindings( loadedState, session, jdbcValueBindings );
+		applyId( id, rowId, mutationExecutor, getStaticMutationOperationGroup(), session );
+		mutationExecutor.executeReactive(
+						entity,
+						null,
+						null,
+						(statementDetails, affectedRowCount, batchPosition) -> identifiedResultsCheck(
+								statementDetails,
+								affectedRowCount,
+								batchPosition,
+								entityPersister(),
+								id,
+								factory()
+						),
+						session
+				)
+				.thenAccept( v -> mutationExecutor.release() )
+				.whenComplete( this::complete );
+	}
+
+	private void complete(Object o, Throwable throwable) {
+		if ( throwable != null ) {
+			stage.toCompletableFuture().completeExceptionally( throwable );
+		}
+		else {
+			stage.toCompletableFuture().complete( null );
+		}
+	}
+
+	private ReactiveMutationExecutor mutationExecutor(
+			SharedSessionContractImplementor session,
+			MutationOperationGroup operationGroup) {
+		final MutationExecutorService mutationExecutorService = session
+				.getFactory()
+				.getServiceRegistry()
+				.getService( MutationExecutorService.class );
+
+		return (ReactiveMutationExecutor) mutationExecutorService
+				.createExecutor( this::getBatchKey, operationGroup, session );
+	}
+}