JdbcAggregateOperations delete by query

Issue link: #1978

Add deleteAllByQuery method to JdbcAggregateOperations

This method enables deleting aggregates based on a query by performing the following steps:
1. Lock the target rows using SELECT ... FOR UPDATE based on the query conditions.
2. Delete sub-entities by leveraging a subquery that selects the matching root rows.
3. Delete the root entities using the query conditions.

But if the query has no criteria, deletion is performed in the same way as deleteAll method of JdbcAggregateOperations

Signed-off-by: JaeYeon Kim <[email protected]>

Author: ghgh415263

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.deleteByQuery(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

@@ -40,6 +40,7 @@
  * @author Myeonghyeon Lee
  * @author Sergey Korotaev
  * @author Tomohiko Ozawa
+ * @author Jaeyeon Kim
  */
 public interface JdbcAggregateOperations {
 
@@ -328,6 +329,15 @@ public interface JdbcAggregateOperations {
 	 */
 	<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);
+
 	/**
 	 * Returns the {@link JdbcConverter}.
 	 *

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

@@ -72,6 +72,7 @@
  * @author Diego Krupitza
  * @author Sergey Korotaev
  * @author Mikhail Polivakha
+ * @author Jaeyeon Kim
  */
 public class JdbcAggregateTemplate implements JdbcAggregateOperations, ApplicationContextAware {
 
@@ -484,6 +485,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);
+	}
+
 	@Override
 	public DataAccessStrategy getDataAccessStrategy() {
 		return accessStrategy;
@@ -672,6 +684,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

@@ -47,6 +47,7 @@
  * @author Chirag Tailor
  * @author Diego Krupitza
  * @author Sergey Korotaev
+ * @author Jaeyeon Kim
  * @since 1.1
  */
 public class CascadingDataAccessStrategy implements DataAccessStrategy {
@@ -128,6 +129,16 @@ public void deleteAll(PersistentPropertyPath<RelationalPersistentProperty> prope
 		collectVoid(das -> das.deleteAll(propertyPath));
 	}
 
+	@Override
+	public void deleteByQuery(Query query, Class<?> domainType) {
+		collectVoid(das -> das.deleteByQuery(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));
@@ -138,6 +149,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

@@ -45,6 +45,7 @@
  * @author Chirag Tailor
  * @author Diego Krupitza
  * @author Sergey Korotaev
+ * @author Jaeyeon Kim
  */
 public interface DataAccessStrategy extends ReadingDataAccessStrategy, RelationResolver {
 
@@ -191,6 +192,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 deleteByQuery(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.
 	 *
@@ -208,6 +225,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

@@ -64,6 +64,7 @@
  * @author Diego Krupitza
  * @author Sergey Korotaev
  * @author Mikhail Polivakha
+ * @author Jaeyeon Kim
  * @since 1.1
  */
 public class DefaultDataAccessStrategy implements DataAccessStrategy {
@@ -254,6 +255,29 @@ public void deleteAll(PersistentPropertyPath<RelationalPersistentProperty> prope
 		operations.getJdbcOperations().update(sql(getBaseType(propertyPath)).createDeleteAllSql(propertyPath));
 	}
 
+	@Override
+	public void deleteByQuery(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) {
 
@@ -270,6 +294,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

@@ -41,6 +41,7 @@
  * @author Chirag Tailor
  * @author Diego Krupitza
  * @author Sergey Korotaev
+ * @author Jaeyeon Kim
  * @since 1.1
  */
 public class DelegatingDataAccessStrategy implements DataAccessStrategy {
@@ -121,6 +122,16 @@ public void deleteAll(PersistentPropertyPath<RelationalPersistentProperty> prope
 		delegate.deleteAll(propertyPath);
 	}
 
+	@Override
+	public void deleteByQuery(Query query, Class<?> domainType) {
+		delegate.deleteByQuery(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);
@@ -131,6 +142,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);