JdbcAggregateOperations delete by query

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

@@ -30,6 +30,7 @@
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
  * @author Mikhail Polivakha
+ * @author Jaeyeon Kim
  * @since 2.0
  */
 class AggregateChangeExecutor {
@@ -101,10 +102,16 @@ private void execute(DbAction<?> action, JdbcAggregateChangeExecutionContext exe
 			executionContext.executeBatchDeleteRoot(batchDeleteRoot);
 		} else if (action instanceof DbAction.DeleteAllRoot<?> deleteAllRoot) {
 			executionContext.executeDeleteAllRoot(deleteAllRoot);
+		} else if (action instanceof DbAction.DeleteRootByQuery<?> deleteRootByQuery) {
+			executionContext.excuteDeleteRootByQuery(deleteRootByQuery);
+		} else if (action instanceof DbAction.DeleteByQuery<?> deleteByQuery) {
+			executionContext.excuteDeleteByQuery(deleteByQuery);
 		} else if (action instanceof DbAction.AcquireLockRoot<?> acquireLockRoot) {
 			executionContext.executeAcquireLock(acquireLockRoot);
 		} else if (action instanceof DbAction.AcquireLockAllRoot<?> acquireLockAllRoot) {
 			executionContext.executeAcquireLockAllRoot(acquireLockAllRoot);
+		} else if (action instanceof DbAction.AcquireLockAllRootByQuery<?> acquireLockAllRootByQuery) {
+			executionContext.executeAcquireLockRootByQuery(acquireLockAllRootByQuery);
 		} else {
 			throw new RuntimeException("unexpected action");
 		}

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

@@ -60,6 +60,7 @@
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
  * @author Mark Paluch
+ * @author Jaeyeon Kim
  */
 @SuppressWarnings("rawtypes")
 class JdbcAggregateChangeExecutionContext {
@@ -169,6 +170,16 @@ <T> void executeDeleteAll(DbAction.DeleteAll<T> delete) {
 		accessStrategy.deleteAll(delete.getPropertyPath());
 	}
 
+	<T> void excuteDeleteRootByQuery(DbAction.DeleteRootByQuery<T> deleteRootByQuery) {
+
+		accessStrategy.deleteRootByQuery(deleteRootByQuery.getQuery(), deleteRootByQuery.getEntityType());
+	}
+
+	<T> void excuteDeleteByQuery(DbAction.DeleteByQuery<T> deleteByQuery) {
+
+		accessStrategy.deleteByQuery(deleteByQuery.getQuery(), deleteByQuery.getPropertyPath());
+	}
+
 	<T> void executeAcquireLock(DbAction.AcquireLockRoot<T> acquireLock) {
 		accessStrategy.acquireLockById(acquireLock.getId(), LockMode.PESSIMISTIC_WRITE, acquireLock.getEntityType());
 	}
@@ -177,6 +188,10 @@ <T> void executeAcquireLockAllRoot(DbAction.AcquireLockAllRoot<T> acquireLock) {
 		accessStrategy.acquireLockAll(LockMode.PESSIMISTIC_WRITE, acquireLock.getEntityType());
 	}
 
+	<T> void executeAcquireLockRootByQuery(DbAction.AcquireLockAllRootByQuery<T> acquireLock) {
+		accessStrategy.acquireLockByQuery(acquireLock.getQuery(), LockMode.PESSIMISTIC_WRITE, acquireLock.getEntityType());
+	}
+
 	private void add(DbActionExecutionResult result) {
 		results.put(result.getAction(), result);
 	}

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

@@ -37,6 +37,7 @@
  * @author Diego Krupitza
  * @author Myeonghyeon Lee
  * @author Sergey Korotaev
+ * @author Jaeyeon Kim
  */
 public interface JdbcAggregateOperations {
 
@@ -324,4 +325,13 @@ public interface JdbcAggregateOperations {
 	 * @param <T> the type of the aggregate roots.
 	 */
 	<T> void deleteAll(Iterable<? extends T> aggregateRoots);
+
+	/**
+	 * Deletes all aggregates of the given type that match the provided query.
+	 *
+	 * @param query Must not be {@code null}.
+	 * @param domainType the type of the aggregate root. Must not be {@code null}.
+	 * @param <T> the type of the aggregate root.
+	 */
+	<T> void deleteAllByQuery(Query query, Class<T> domainType);
 }

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

@@ -70,6 +70,7 @@
  * @author Diego Krupitza
  * @author Sergey Korotaev
  * @author Mikhail Polivakha
+ * @author Jaeyeon Kim
  */
 public class JdbcAggregateTemplate implements JdbcAggregateOperations {
 
@@ -461,6 +462,17 @@ public <T> void deleteAll(Iterable<? extends T> instances) {
 		}
 	}
 
+	@Override
+	public <T> void deleteAllByQuery(Query query, Class<T> domainType) {
+
+		Assert.notNull(query, "Query must not be null");
+		Assert.notNull(domainType, "Domain type must not be null");
+
+		MutableAggregateChange<?> change = createDeletingChange(query, domainType);
+
+		executor.executeDelete(change);
+	}
+
 	private <T> void verifyIdProperty(T instance) {
 		// accessing the id property just to raise an exception in the case it does not exist.
 		context.getRequiredPersistentEntity(instance.getClass()).getRequiredIdProperty();
@@ -639,6 +651,13 @@ private MutableAggregateChange<?> createDeletingChange(Class<?> domainType) {
 		return aggregateChange;
 	}
 
+	private MutableAggregateChange<?> createDeletingChange(Query query, Class<?> domainType) {
+
+		MutableAggregateChange<?> aggregateChange = MutableAggregateChange.forDelete(domainType);
+		jdbcEntityDeleteWriter.writeForQuery(query, aggregateChange);
+		return aggregateChange;
+	}
+
 	private <T> List<T> triggerAfterConvert(Iterable<T> all) {
 
 		List<T> result = new ArrayList<>();

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

@@ -44,6 +44,7 @@
  * @author Chirag Tailor
  * @author Diego Krupitza
  * @author Sergey Korotaev
+ * @author Jaeyeon Kim
  * @since 1.1
  */
 public class CascadingDataAccessStrategy implements DataAccessStrategy {
@@ -109,6 +110,16 @@ public void deleteAll(PersistentPropertyPath<RelationalPersistentProperty> prope
 		collectVoid(das -> das.deleteAll(propertyPath));
 	}
 
+	@Override
+	public void deleteRootByQuery(Query query, Class<?> domainType) {
+		collectVoid(das -> das.deleteRootByQuery(query, domainType));
+	}
+
+	@Override
+	public void deleteByQuery(Query query, PersistentPropertyPath<RelationalPersistentProperty> propertyPath) {
+		collectVoid(das -> das.deleteByQuery(query, propertyPath));
+	}
+
 	@Override
 	public <T> void acquireLockById(Object id, LockMode lockMode, Class<T> domainType) {
 		collectVoid(das -> das.acquireLockById(id, lockMode, domainType));
@@ -119,6 +130,11 @@ public <T> void acquireLockAll(LockMode lockMode, Class<T> domainType) {
 		collectVoid(das -> das.acquireLockAll(lockMode, domainType));
 	}
 
+	@Override
+	public <T> void acquireLockByQuery(Query query, LockMode lockMode, Class<T> domainType) {
+		collectVoid(das -> das.acquireLockByQuery(query, lockMode, domainType));
+	}
+
 	@Override
 	public long count(Class<?> domainType) {
 		return collect(das -> das.count(domainType));

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

@@ -43,6 +43,7 @@
  * @author Chirag Tailor
  * @author Diego Krupitza
  * @author Sergey Korotaev
+ * @author Jaeyeon Kim
  */
 public interface DataAccessStrategy extends ReadingDataAccessStrategy, RelationResolver {
 
@@ -177,6 +178,22 @@ public interface DataAccessStrategy extends ReadingDataAccessStrategy, RelationR
 	 */
 	void deleteAll(PersistentPropertyPath<RelationalPersistentProperty> propertyPath);
 
+	/**
+	 * Deletes all root entities of the given domain type that match the given {@link Query}.
+	 *
+	 * @param query the query specifying which rows to delete. Must not be {@code null}.
+	 * @param domainType the domain type of the entity. Must not be {@code null}.
+	 */
+	void deleteRootByQuery(Query query, Class<?> domainType);
+
+	/**
+	 * Deletes entities reachable via the given {@link PersistentPropertyPath} from root entities that match the given {@link Query}.
+	 *
+	 * @param query the query specifying which root entities to consider for deleting related entities. Must not be {@code null}.
+	 * @param propertyPath Leading from the root object to the entities to be deleted. Must not be {@code null}.
+	 */
+	void deleteByQuery(Query query, PersistentPropertyPath<RelationalPersistentProperty> propertyPath);
+
 	/**
 	 * Acquire a lock on the aggregate specified by id.
 	 *
@@ -194,6 +211,16 @@ public interface DataAccessStrategy extends ReadingDataAccessStrategy, RelationR
 	 */
 	<T> void acquireLockAll(LockMode lockMode, Class<T> domainType);
 
+	/**
+	 * Acquire a lock on all aggregates that match the given {@link Query}.
+	 *
+	 * @param query the query specifying which entities to lock. Must not be {@code null}.
+	 * @param lockMode the lock mode to apply to the query (e.g. {@code FOR UPDATE}). Must not be {@code null}.
+	 * @param domainType the domain type of the entities to be locked. Must not be {@code null}.
+	 * @param <T> the type of the domain entity.
+	 */
+	<T> void acquireLockByQuery(Query query, LockMode lockMode, Class<T> domainType);
+
 	/**
 	 * Counts the rows in the table representing the given domain type.
 	 *

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

@@ -63,6 +63,7 @@
  * @author Diego Krupitza
  * @author Sergey Korotaev
  * @author Mikhail Polivakha
+ * @author Jaeyeon Kim
  * @since 1.1
  */
 public class DefaultDataAccessStrategy implements DataAccessStrategy {
@@ -243,6 +244,30 @@ public void deleteAll(PersistentPropertyPath<RelationalPersistentProperty> prope
 		operations.getJdbcOperations().update(sql(getBaseType(propertyPath)).createDeleteAllSql(propertyPath));
 	}
 
+	@Override
+	public void deleteRootByQuery(Query query, Class<?> domainType) {
+
+		MapSqlParameterSource parameterSource = new MapSqlParameterSource();
+		String deleteSql = sql(domainType).createDeleteByQuery(query, parameterSource);
+
+		operations.update(deleteSql, parameterSource);
+	}
+
+	@Override
+	public void deleteByQuery(Query query, PersistentPropertyPath<RelationalPersistentProperty> propertyPath) {
+
+		RelationalPersistentEntity<?> rootEntity = context.getRequiredPersistentEntity(getBaseType(propertyPath));
+
+		RelationalPersistentProperty referencingProperty = propertyPath.getLeafProperty();
+
+		Assert.notNull(referencingProperty, "No property found matching the PropertyPath " + propertyPath);
+
+		MapSqlParameterSource parameterSource = new MapSqlParameterSource();
+		String deleteSql = sql(rootEntity.getType()).createDeleteInSubselectByPath(query, parameterSource, propertyPath);
+
+		operations.update(deleteSql, parameterSource);
+	}
+
 	@Override
 	public <T> void acquireLockById(Object id, LockMode lockMode, Class<T> domainType) {
 
@@ -259,6 +284,15 @@ public <T> void acquireLockAll(LockMode lockMode, Class<T> domainType) {
 		operations.getJdbcOperations().query(acquireLockAllSql, ResultSet::next);
 	}
 
+	@Override
+	public <T> void acquireLockByQuery(Query query, LockMode lockMode, Class<T> domainType) {
+
+		MapSqlParameterSource parameterSource = new MapSqlParameterSource();
+		String acquireLockByQuerySql = sql(domainType).getAcquireLockByQuery(query, parameterSource, lockMode);
+
+		operations.query(acquireLockByQuerySql, parameterSource, ResultSet::next);
+	}
+
 	@Override
 	public long count(Class<?> domainType) {
 

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

@@ -39,6 +39,7 @@
  * @author Chirag Tailor
  * @author Diego Krupitza
  * @author Sergey Korotaev
+ * @author Jaeyeon Kim
  * @since 1.1
  */
 public class DelegatingDataAccessStrategy implements DataAccessStrategy {
@@ -109,6 +110,16 @@ public void deleteAll(PersistentPropertyPath<RelationalPersistentProperty> prope
 		delegate.deleteAll(propertyPath);
 	}
 
+	@Override
+	public void deleteRootByQuery(Query query, Class<?> domainType) {
+		delegate.deleteRootByQuery(query, domainType);
+	}
+
+	@Override
+	public void deleteByQuery(Query query, PersistentPropertyPath<RelationalPersistentProperty> propertyPath) {
+		delegate.deleteByQuery(query, propertyPath);
+	}
+
 	@Override
 	public <T> void acquireLockById(Object id, LockMode lockMode, Class<T> domainType) {
 		delegate.acquireLockById(id, lockMode, domainType);
@@ -119,6 +130,11 @@ public <T> void acquireLockAll(LockMode lockMode, Class<T> domainType) {
 		delegate.acquireLockAll(lockMode, domainType);
 	}
 
+	@Override
+	public <T> void acquireLockByQuery(Query query, LockMode lockMode, Class<T> domainType) {
+		delegate.acquireLockByQuery(query, lockMode, domainType);
+	}
+
 	@Override
 	public long count(Class<?> domainType) {
 		return delegate.count(domainType);