Add DeleteBatchingAggregateChange to batch deletes when deleting multiple aggregate roots.

+ Rename DefaultAggregateChange to DeleteAggregateChange.

Original pull request #1230

Author: ctailor2

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateOperations.java

@@ -80,6 +80,15 @@ public interface JdbcAggregateOperations {
 	 */
 	<T> void deleteById(Object id, Class<T> domainType);
 
+	/**
+	 * Deletes all aggregates identified by their aggregate root ids.
+	 *
+	 * @param ids the ids of the aggregate roots of the aggregates to be deleted. Must not be {@code null}.
+	 * @param domainType the type of the aggregate root.
+	 * @param <T> the type of the aggregate root.
+	 */
+	<T> void deleteAllById(Iterable<?> ids, Class<T> domainType);
+
 	/**
 	 * Delete an aggregate identified by it's aggregate root.
 	 *
@@ -96,6 +105,15 @@ public interface JdbcAggregateOperations {
 	 */
 	void deleteAll(Class<?> domainType);
 
+	/**
+	 * Delete all aggregates identified by their aggregate roots.
+	 *
+	 * @param aggregateRoots to delete. Must not be {@code null}.
+	 * @param domainType type of the aggregate roots to be deleted. Must not be {@code null}.
+	 * @param <T> the type of the aggregate roots.
+	 */
+	<T> void deleteAll(Iterable<? extends T> aggregateRoots, Class<T> domainType);
+
 	/**
 	 * Counts the number of aggregates of a given type.
 	 *

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateTemplate.java

@@ -17,7 +17,9 @@
 
 import java.util.ArrayList;
 import java.util.Iterator;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
@@ -33,6 +35,7 @@
 import org.springframework.data.mapping.callback.EntityCallbacks;
 import org.springframework.data.relational.core.conversion.AggregateChange;
 import org.springframework.data.relational.core.conversion.BatchingAggregateChange;
+import org.springframework.data.relational.core.conversion.DeleteAggregateChange;
 import org.springframework.data.relational.core.conversion.MutableAggregateChange;
 import org.springframework.data.relational.core.conversion.RelationalEntityDeleteWriter;
 import org.springframework.data.relational.core.conversion.RelationalEntityInsertWriter;
@@ -275,6 +278,25 @@ public <S> void deleteById(Object id, Class<S> domainType) {
 		deleteTree(id, null, domainType);
 	}
 
+	@Override
+	public <T> void deleteAllById(Iterable<?> ids, Class<T> domainType) {
+
+		Assert.isTrue(ids.iterator().hasNext(), "Ids must not be empty!");
+
+		BatchingAggregateChange<T, DeleteAggregateChange<T>> batchingAggregateChange = BatchingAggregateChange
+				.forDelete(domainType);
+
+		ids.forEach(id -> {
+			DeleteAggregateChange<T> change = createDeletingChange(id, null, domainType);
+			triggerBeforeDelete(null, id, change);
+			batchingAggregateChange.add(change);
+		});
+
+		executor.executeDelete(batchingAggregateChange);
+
+		ids.forEach(id -> triggerAfterDelete(null, id, batchingAggregateChange));
+	}
+
 	@Override
 	public void deleteAll(Class<?> domainType) {
 
@@ -284,6 +306,28 @@ public void deleteAll(Class<?> domainType) {
 		executor.executeDelete(change);
 	}
 
+	@Override
+	public <T> void deleteAll(Iterable<? extends T> instances, Class<T> domainType) {
+
+		Assert.isTrue(instances.iterator().hasNext(), "Aggregate instances must not be empty!");
+
+		BatchingAggregateChange<T, DeleteAggregateChange<T>> batchingAggregateChange = BatchingAggregateChange
+				.forDelete(domainType);
+		Map<Object, T> instancesBeforeExecute = new LinkedHashMap<>();
+
+		instances.forEach(instance -> {
+			Object id = context.getRequiredPersistentEntity(domainType).getIdentifierAccessor(instance)
+					.getRequiredIdentifier();
+			DeleteAggregateChange<T> change = createDeletingChange(id, instance, domainType);
+			instancesBeforeExecute.put(id, triggerBeforeDelete(instance, id, change));
+			batchingAggregateChange.add(change);
+		});
+
+		executor.executeDelete(batchingAggregateChange);
+
+		instancesBeforeExecute.forEach((id, instance) -> triggerAfterDelete(instance, id, batchingAggregateChange));
+	}
+
 	private <T> T afterExecute(AggregateChange<T> change, T entityAfterExecution) {
 
 		Object identifier = context.getRequiredPersistentEntity(change.getEntityType())
@@ -416,7 +460,7 @@ private <T> RelationalPersistentEntity<T> getRequiredPersistentEntity(T instance
 		return (RelationalPersistentEntity<T>) context.getRequiredPersistentEntity(instance.getClass());
 	}
 
-	private <T> MutableAggregateChange<T> createDeletingChange(Object id, @Nullable T entity, Class<T> domainType) {
+	private <T> DeleteAggregateChange<T> createDeletingChange(Object id, @Nullable T entity, Class<T> domainType) {
 
 		Number previousVersion = null;
 		if (entity != null) {
@@ -425,7 +469,7 @@ private <T> MutableAggregateChange<T> createDeletingChange(Object id, @Nullable
 				previousVersion = RelationalEntityVersionUtils.getVersionNumberFromEntity(entity, persistentEntity, converter);
 			}
 		}
-		MutableAggregateChange<T> aggregateChange = MutableAggregateChange.forDelete(domainType, previousVersion);
+		DeleteAggregateChange<T> aggregateChange = MutableAggregateChange.forDelete(domainType, previousVersion);
 		jdbcEntityDeleteWriter.write(id, aggregateChange);
 		return aggregateChange;
 	}
@@ -475,7 +519,7 @@ private <T> T triggerAfterSave(T aggregateRoot, AggregateChange<T> change) {
 		return entityCallbacks.callback(AfterSaveCallback.class, aggregateRoot);
 	}
 
-	private <T> void triggerAfterDelete(@Nullable T aggregateRoot, Object id, MutableAggregateChange<T> change) {
+	private <T> void triggerAfterDelete(@Nullable T aggregateRoot, Object id, AggregateChange<T> change) {
 
 		publisher.publishEvent(new AfterDeleteEvent<>(Identifier.of(id), aggregateRoot, change));
 

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/support/SimpleJdbcRepository.java

@@ -101,14 +101,13 @@ public void delete(T instance) {
 
 	@Override
 	public void deleteAllById(Iterable<? extends ID> ids) {
-		ids.forEach(it -> entityOperations.deleteById(it, entity.getType()));
+		entityOperations.deleteAllById(ids, entity.getType());
 	}
 
 	@Transactional
 	@Override
-	@SuppressWarnings("unchecked")
 	public void deleteAll(Iterable<? extends T> entities) {
-		entities.forEach(it -> entityOperations.delete(it, (Class<T>) it.getClass()));
+		entityOperations.deleteAll(entities, entity.getType());
 	}
 
 	@Transactional

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/JdbcAggregateTemplateIntegrationTests.java

@@ -331,6 +331,38 @@ void saveAndDeleteAllWithReferencedEntity() {
 		softly.assertAll();
 	}
 
+	@Test // GH-537
+	@EnabledOnFeature(SUPPORTS_QUOTED_IDS)
+	void saveAndDeleteAllByAggregateRootsWithReferencedEntity() {
+		LegoSet legoSet1 = template.save(legoSet);
+		LegoSet legoSet2 = template.save(createLegoSet("Some Name"));
+
+		template.deleteAll(List.of(legoSet1, legoSet2), LegoSet.class);
+
+		SoftAssertions softly = new SoftAssertions();
+
+		assertThat(template.findAll(LegoSet.class)).isEmpty();
+		assertThat(template.findAll(Manual.class)).isEmpty();
+
+		softly.assertAll();
+	}
+
+	@Test // GH-537
+	@EnabledOnFeature(SUPPORTS_QUOTED_IDS)
+	void saveAndDeleteAllByIdsWithReferencedEntity() {
+		LegoSet legoSet1 = template.save(legoSet);
+		LegoSet legoSet2 = template.save(createLegoSet("Some Name"));
+
+		template.deleteAllById(List.of(legoSet1.id, legoSet2.id), LegoSet.class);
+
+		SoftAssertions softly = new SoftAssertions();
+
+		assertThat(template.findAll(LegoSet.class)).isEmpty();
+		assertThat(template.findAll(Manual.class)).isEmpty();
+
+		softly.assertAll();
+	}
+
 	@Test // DATAJDBC-112
 	@EnabledOnFeature({ SUPPORTS_QUOTED_IDS, SUPPORTS_GENERATED_IDS_IN_REFERENCED_ENTITIES })
 	void updateReferencedEntityFromNull() {

spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/BatchingAggregateChange.java

@@ -47,4 +47,19 @@ static <T> BatchingAggregateChange<T, RootAggregateChange<T>> forSave(Class<T> e
 
 		return new SaveBatchingAggregateChange<>(entityClass);
 	}
+
+	/**
+	 * Factory method to create a {@link BatchingAggregateChange} for deleting entities.
+	 *
+	 * @param entityClass aggregate root type.
+	 * @param <T> entity type.
+	 * @return the {@link BatchingAggregateChange} for deleting root entities.
+	 * @since 3.0
+	 */
+	static <T> BatchingAggregateChange<T, DeleteAggregateChange<T>> forDelete(Class<T> entityClass) {
+
+		Assert.notNull(entityClass, "Entity class must not be null");
+
+		return new DeleteBatchingAggregateChange<>(entityClass);
+	}
 }

spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/DefaultAggregateChange.java renamed to spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/DeleteAggregateChange.java

@@ -30,9 +30,7 @@
  * @author Chirag Tailor
  * @since 2.0
  */
-class DefaultAggregateChange<T> implements MutableAggregateChange<T> {
-
-	private final Kind kind;
+public class DeleteAggregateChange<T> implements MutableAggregateChange<T> {
 
 	/** Type of the aggregate root to be changed */
 	private final Class<T> entityType;
@@ -42,9 +40,7 @@ class DefaultAggregateChange<T> implements MutableAggregateChange<T> {
 	/** The previous version assigned to the instance being changed, if available */
 	@Nullable private final Number previousVersion;
 
-	public DefaultAggregateChange(Kind kind, Class<T> entityType, @Nullable Number previousVersion) {
-
-		this.kind = kind;
+	public DeleteAggregateChange(Class<T> entityType, @Nullable Number previousVersion) {
 		this.entityType = entityType;
 		this.previousVersion = previousVersion;
 	}
@@ -64,7 +60,7 @@ public void addAction(DbAction<?> action) {
 
 	@Override
 	public Kind getKind() {
-		return this.kind;
+		return Kind.DELETE;
 	}
 
 	@Override

spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/DeleteBatchingAggregateChange.java

@@ -0,0 +1,89 @@
+package org.springframework.data.relational.core.conversion;
+
+import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import static java.util.Collections.*;
+
+/**
+ * A {@link BatchingAggregateChange} implementation for delete changes that can contain actions for one or more delete
+ * operations. When consumed, actions are yielded in the appropriate entity tree order with deletes carried out from
+ * leaves to root. All operations that can be batched are grouped and combined to offer the ability for an optimized
+ * batch operation to be used.
+ *
+ * @author Chirag Tailor
+ * @since 3.0
+ */
+public class DeleteBatchingAggregateChange<T> implements BatchingAggregateChange<T, DeleteAggregateChange<T>> {
+
+	private static final Comparator<PersistentPropertyPath<RelationalPersistentProperty>> pathLengthComparator = //
+			Comparator.comparing(PersistentPropertyPath::getLength);
+
+	private final Class<T> entityType;
+	private final List<DbAction.DeleteRoot<T>> rootActions = new ArrayList<>();
+	private final List<DbAction.AcquireLockRoot<?>> lockActions = new ArrayList<>();
+	private final Map<PersistentPropertyPath<RelationalPersistentProperty>, List<DbAction.Delete<Object>>> deleteActions = //
+			new HashMap<>();
+
+	public DeleteBatchingAggregateChange(Class<T> entityType) {
+		this.entityType = entityType;
+	}
+
+	@Override
+	public Kind getKind() {
+		return Kind.DELETE;
+	}
+
+	@Override
+	public Class<T> getEntityType() {
+		return entityType;
+	}
+
+	@Override
+	public void forEachAction(Consumer<? super DbAction<?>> consumer) {
+
+		lockActions.forEach(consumer);
+		deleteActions.entrySet().stream().sorted(Map.Entry.comparingByKey(pathLengthComparator.reversed()))
+				.forEach((entry) -> {
+					List<DbAction.Delete<Object>> deletes = entry.getValue();
+					if (deletes.size() > 1) {
+						consumer.accept(new DbAction.BatchDelete<>(deletes));
+					} else {
+						deletes.forEach(consumer);
+					}
+				});
+		rootActions.forEach(consumer);
+	}
+
+	@Override
+	public void add(DeleteAggregateChange<T> aggregateChange) {
+
+		aggregateChange.forEachAction(action -> {
+			if (action instanceof DbAction.DeleteRoot<?> deleteRootAction) {
+				//noinspection unchecked
+				rootActions.add((DbAction.DeleteRoot<T>) deleteRootAction);
+			} else if (action instanceof DbAction.Delete<?> deleteAction) {
+				// noinspection unchecked
+				addDelete((DbAction.Delete<Object>) deleteAction);
+			} else if (action instanceof DbAction.AcquireLockRoot<?> lockRootAction) {
+				lockActions.add(lockRootAction);
+			}
+		});
+	}
+
+	private void addDelete(DbAction.Delete<Object> action) {
+
+		PersistentPropertyPath<RelationalPersistentProperty> propertyPath = action.getPropertyPath();
+		deleteActions.merge(propertyPath, new ArrayList<>(singletonList(action)), (actions, defaultValue) -> {
+			actions.add(action);
+			return actions;
+		});
+	}
+}

spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/MutableAggregateChange.java

@@ -59,47 +59,47 @@ static <T> RootAggregateChange<T> forSave(T entity, @Nullable Number previousVer
 	}
 
 	/**
-	 * Factory method to create an {@link MutableAggregateChange} for deleting entities.
+	 * Factory method to create a {@link DeleteAggregateChange} for deleting entities.
 	 *
 	 * @param entity aggregate root to delete.
 	 * @param <T> entity type.
-	 * @return the {@link MutableAggregateChange} for deleting the root {@code entity}.
+	 * @return the {@link DeleteAggregateChange} for deleting the root {@code entity}.
 	 * @since 1.2
 	 */
 	@SuppressWarnings("unchecked")
-	static <T> MutableAggregateChange<T> forDelete(T entity) {
+	static <T> DeleteAggregateChange<T> forDelete(T entity) {
 
 		Assert.notNull(entity, "Entity must not be null");
 
 		return forDelete((Class<T>) ClassUtils.getUserClass(entity));
 	}
 
 	/**
-	 * Factory method to create an {@link MutableAggregateChange} for deleting entities.
+	 * Factory method to create a {@link DeleteAggregateChange} for deleting entities.
 	 *
 	 * @param entityClass aggregate root type.
 	 * @param <T> entity type.
-	 * @return the {@link MutableAggregateChange} for deleting the root {@code entity}.
+	 * @return the {@link DeleteAggregateChange} for deleting the root {@code entity}.
 	 * @since 1.2
 	 */
-	static <T> MutableAggregateChange<T> forDelete(Class<T> entityClass) {
+	static <T> DeleteAggregateChange<T> forDelete(Class<T> entityClass) {
 		return forDelete(entityClass, null);
 	}
 
 	/**
-	 * Factory method to create an {@link MutableAggregateChange} for deleting entities.
+	 * Factory method to create a {@link DeleteAggregateChange} for deleting entities.
 	 *
 	 * @param entityClass aggregate root type.
 	 * @param previousVersion the previous version assigned to the instance being saved. May be {@literal null}.
 	 * @param <T> entity type.
-	 * @return the {@link MutableAggregateChange} for deleting the root {@code entity}.
+	 * @return the {@link DeleteAggregateChange} for deleting the root {@code entity}.
 	 * @since 2.4
 	 */
-	static <T> MutableAggregateChange<T> forDelete(Class<T> entityClass, @Nullable Number previousVersion) {
+	static <T> DeleteAggregateChange<T> forDelete(Class<T> entityClass, @Nullable Number previousVersion) {
 
 		Assert.notNull(entityClass, "Entity class must not be null");
 
-		return new DefaultAggregateChange<>(Kind.DELETE, entityClass, previousVersion);
+		return new DeleteAggregateChange<>(entityClass, previousVersion);
 	}
 
 	/**