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 <ghgh415263@naver.com>

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

@@ -30,6 +30,7 @@ import org.springframework.data.relational.core.conversion.MutableAggregateChang
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
  * @author Mikhail Polivakha
+ * @author Jaeyeon Kim
  * @since 2.0
  */
 class AggregateChangeExecutor {
@@ -101,10 +102,16 @@ class AggregateChangeExecutor {
 			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

@@ -51,6 +51,7 @@ import org.springframework.util.Assert;
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
  * @author Mark Paluch
+ * @author Jaeyeon Kim
  */
 @SuppressWarnings("rawtypes")
 class JdbcAggregateChangeExecutionContext {
@@ -160,6 +161,16 @@ class JdbcAggregateChangeExecutionContext {
 		accessStrategy.deleteAll(delete.propertyPath());
 	}
 
+	<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.propertyPath());
+	}
+
 	<T> void executeAcquireLock(DbAction.AcquireLockRoot<T> acquireLock) {
 		accessStrategy.acquireLockById(acquireLock.getId(), LockMode.PESSIMISTIC_WRITE, acquireLock.getEntityType());
 	}
@@ -168,6 +179,10 @@ class JdbcAggregateChangeExecutionContext {
 		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 @@ import org.springframework.data.relational.core.query.Query;
  * @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 @@ import org.springframework.util.ClassUtils;
  * @author Diego Krupitza
  * @author Sergey Korotaev
  * @author Mikhail Polivakha
+ * @author Jaeyeon Kim
  */
 public class JdbcAggregateTemplate implements JdbcAggregateOperations, ApplicationContextAware {
 
@@ -484,6 +485,17 @@ public class JdbcAggregateTemplate implements JdbcAggregateOperations, Applicati
 		}
 	}
 
+	@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 @@ public class JdbcAggregateTemplate implements JdbcAggregateOperations, Applicati
 		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

@@ -49,6 +49,7 @@ import org.springframework.util.Assert;
  * @author Chirag Tailor
  * @author Diego Krupitza
  * @author Sergey Korotaev
+ * @author Jaeyeon Kim
  * @since 1.1
  */
 public class CascadingDataAccessStrategy implements DataAccessStrategy {
@@ -132,6 +133,16 @@ public class CascadingDataAccessStrategy implements DataAccessStrategy {
 		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));
@@ -142,6 +153,11 @@ public class CascadingDataAccessStrategy implements DataAccessStrategy {
 		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 @@ import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
  * @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 @@ import org.springframework.util.Assert;
  * @author Diego Krupitza
  * @author Sergey Korotaev
  * @author Mikhail Polivakha
+ * @author Jaeyeon Kim
  * @since 1.1
  */
 public class DefaultDataAccessStrategy implements DataAccessStrategy {
@@ -256,6 +257,29 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 		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) {
 
@@ -272,6 +296,15 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 		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

@@ -42,6 +42,7 @@ import org.springframework.util.Assert;
  * @author Chirag Tailor
  * @author Diego Krupitza
  * @author Sergey Korotaev
+ * @author Jaeyeon Kim
  * @since 1.1
  */
 public class DelegatingDataAccessStrategy implements DataAccessStrategy {
@@ -126,6 +127,16 @@ public class DelegatingDataAccessStrategy implements DataAccessStrategy {
 		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);
@@ -136,6 +147,11 @@ public class DelegatingDataAccessStrategy implements DataAccessStrategy {
 		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);

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

@@ -61,6 +61,7 @@ import org.springframework.util.Assert;
  * @author Hari Ohm Prasath
  * @author Viktor Ardelean
  * @author Kurt Niemi
+ * @author Jaeyeon Kim
  */
 public class SqlGenerator {
 
@@ -374,6 +375,18 @@ public class SqlGenerator {
 		return this.createAcquireLockAll(lockMode);
 	}
 
+	/**
+	 * Create a {@code SELECT id FROM … WHERE … (LOCK CLAUSE)} statement based on the given query.
+	 *
+	 * @param query the query to base the select on. Must not be null.
+	 * @param parameterSource the source for holding the bindings.
+	 * @param lockMode Lock clause mode.
+	 * @return the SQL statement as a {@link String}. Guaranteed to be not {@literal null}.
+	 */
+	String getAcquireLockByQuery(Query query, MapSqlParameterSource parameterSource, LockMode lockMode) {
+		return this.createAcquireLockByQuery(query, parameterSource, lockMode);
+	}
+
 	/**
 	 * Create a {@code INSERT INTO … (…) VALUES(…)} statement.
 	 *
@@ -489,6 +502,72 @@ public class SqlGenerator {
 		return createDeleteByPathAndCriteria(mappingContext.getAggregatePath(path), this::inCondition);
 	}
 
+	/**
+	 * Create a {@code DELETE FROM ... WHERE ...} SQL statement based on the given {@link Query}.
+	 *
+	 * @param query the query object defining filter criteria; must not be {@literal null}.
+	 * @param parameterSource the parameter bindings for the query; must not be {@literal null}.
+	 * @return the SQL DELETE statement as a {@link String}; guaranteed to be not {@literal null}.
+	 */
+	public String createDeleteByQuery(Query query, MapSqlParameterSource parameterSource) {
+		Assert.notNull(parameterSource, "parameterSource must not be null");
+
+		Table table = this.getTable();
+
+		DeleteBuilder.DeleteWhere builder = Delete.builder()
+				.from(table);
+
+		query.getCriteria()
+				.filter(criteria -> !criteria.isEmpty())
+				.map(criteria -> queryMapper.getMappedObject(parameterSource, criteria, table, entity))
+				.ifPresent(builder::where);
+
+		return render(builder.build());
+	}
+
+	/**
+	 * Creates a {@code DELETE} SQL query that targets a specific table defined by the given {@link PersistentPropertyPath},
+	 * and applies filtering using a subselect based on the provided {@link Query}.
+	 *
+	 * @param query the query object containing the filtering criteria; must not be {@literal null}.
+	 * @param parameterSource the source for parameter bindings used in the query; must not be {@literal null}.
+	 * @param propertyPath must not be {@literal null}.
+	 * @return the DELETE SQL statement as a {@link String}. Guaranteed to be not {@literal null}.
+	 */
+	public String createDeleteInSubselectByPath(Query query, MapSqlParameterSource parameterSource,
+												PersistentPropertyPath<RelationalPersistentProperty> propertyPath) {
+
+		Assert.notNull(parameterSource, "parameterSource must not be null");
+
+		AggregatePath path = mappingContext.getAggregatePath(propertyPath);
+
+        return createDeleteByPathAndCriteria(path, columnMap -> {
+            Select subSelect = createRootIdSubSelect(query, parameterSource);
+			Collection<Column> columns = columnMap.values();
+            Expression expression = columns.size() == 1 ? columns.iterator().next() : TupleExpression.create(columns);
+            return Conditions.in(expression, subSelect);
+        });
+	}
+
+	/**
+	 * Creates a subselect that retrieves root entity IDs filtered by the given query.
+	 */
+	private Select createRootIdSubSelect(Query query, MapSqlParameterSource parameterSource) {
+
+		Table table = this.getTable();
+
+		SelectBuilder.SelectWhere selectBuilder = StatementBuilder
+				.select(getIdColumns())
+				.from(table);
+
+		query.getCriteria()
+				.filter(criteria -> !criteria.isEmpty())
+				.map(criteria -> queryMapper.getMappedObject(parameterSource, criteria, table, entity))
+				.ifPresent(selectBuilder::where);
+
+		return selectBuilder.build();
+	}
+
 	/**
 	 * Constructs a where condition. The where condition will be of the form {@literal <columns> IN :bind-marker}
 	 */
@@ -596,6 +675,28 @@ public class SqlGenerator {
 		return render(select);
 	}
 
+	private String createAcquireLockByQuery(Query query, MapSqlParameterSource parameterSource, LockMode lockMode) {
+
+		Assert.notNull(parameterSource, "parameterSource must not be null");
+
+		Table table = this.getTable();
+
+		SelectBuilder.SelectWhere selectBuilder = StatementBuilder
+				.select(getSingleNonNullColumn())
+				.from(table);
+
+		query.getCriteria()
+				.filter(criteria -> !criteria.isEmpty())
+				.map(criteria -> queryMapper.getMappedObject(parameterSource, criteria, table, entity))
+				.ifPresent(selectBuilder::where);
+
+		Select select = selectBuilder
+				.lock(lockMode)
+				.build();
+
+		return render(select);
+	}
+
 	private String createFindAllSql() {
 		return render(selectBuilder().build());
 	}

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/mybatis/MyBatisDataAccessStrategy.java

@@ -74,6 +74,7 @@ import org.springframework.util.Assert;
  * @author Christopher Klein
  * @author Mikhail Polivakha
  * @author Sergey Korotaev
+ * @author Jaeyeon Kim
  */
 public class MyBatisDataAccessStrategy implements DataAccessStrategy {
 
@@ -255,6 +256,16 @@ public class MyBatisDataAccessStrategy implements DataAccessStrategy {
 		sqlSession().delete(statement, parameter);
 	}
 
+	@Override
+	public void deleteByQuery(Query query, Class<?> domainType) {
+		throw new UnsupportedOperationException("Not implemented");
+	}
+
+	@Override
+	public void deleteByQuery(Query query, PersistentPropertyPath<RelationalPersistentProperty> propertyPath) {
+		throw new UnsupportedOperationException("Not implemented");
+	}
+
 	@Override
 	public <T> void acquireLockById(Object id, LockMode lockMode, Class<T> domainType) {
 
@@ -278,6 +289,11 @@ public class MyBatisDataAccessStrategy implements DataAccessStrategy {
 		sqlSession().selectOne(statement, parameter);
 	}
 
+	@Override
+	public <T> void acquireLockByQuery(Query query, LockMode lockMode, Class<T> domainType) {
+		throw new UnsupportedOperationException("Not implemented");
+	}
+
 	@Override
 	public <T extends @Nullable Object> T findById(Object id, Class<T> domainType) {
 

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

@@ -32,6 +32,7 @@ import org.springframework.data.jdbc.testing.EnabledOnDatabase;
 import org.springframework.data.jdbc.testing.IntegrationTest;
 import org.springframework.data.jdbc.testing.TestConfiguration;
 import org.springframework.data.relational.core.mapping.Embedded;
+import org.springframework.data.relational.core.query.Criteria;
 import org.springframework.data.relational.core.query.Query;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
 
@@ -39,6 +40,7 @@ import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
  * Integration tests for {@link JdbcAggregateTemplate} and it's handling of entities with embedded entities as keys.
  *
  * @author Jens Schauder
+ * @author Jaeyeon Kim
  */
 @IntegrationTest
 @EnabledOnDatabase(DatabaseType.HSQL)
@@ -129,6 +131,25 @@ class CompositeIdAggregateTemplateHsqlIntegrationTests {
 		assertThat(reloaded).containsExactly(entities.get(2));
 	}
 
+	@Test // GH-1978
+	void deleteAllByQueryWithEmbeddedPk() {
+
+		List<SimpleEntityWithEmbeddedPk> entities = (List<SimpleEntityWithEmbeddedPk>) template
+				.insertAll(List.of(new SimpleEntityWithEmbeddedPk(new EmbeddedPk(1L, "a"), "alpha"),
+						new SimpleEntityWithEmbeddedPk(new EmbeddedPk(2L, "b"), "beta"),
+						new SimpleEntityWithEmbeddedPk(new EmbeddedPk(3L, "b"), "gamma")));
+
+		Query query = Query.query(Criteria.where("name").is("beta"));
+		template.deleteAllByQuery(query, SimpleEntityWithEmbeddedPk.class);
+
+		assertThat(
+				template.findAll(SimpleEntityWithEmbeddedPk.class))
+				.containsExactlyInAnyOrder(
+						entities.get(0), // alpha
+						entities.get(2)  // gamma
+				);
+	}
+
 	@Test // GH-574
 	void existsSingleSimpleEntityWithEmbeddedPk() {
 

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

@@ -73,6 +73,7 @@ import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
  * @author Diego Krupitza
  * @author Hari Ohm Prasath
  * @author Viktor Ardelean
+ * @author Jaeyeon Kim
  */
 @SuppressWarnings("Convert2MethodRef")
 class SqlGeneratorUnitTests {
@@ -165,6 +166,16 @@ class SqlGeneratorUnitTests {
 				.doesNotContain("Element AS elements"));
 	}
 
+	@Test // GH-1978
+	void getAcquireLockByQuery(){
+
+		Query query = Query.query(Criteria.where("id").is(23L));
+
+		String sql = sqlGenerator.getAcquireLockByQuery(query, new MapSqlParameterSource(), LockMode.PESSIMISTIC_WRITE);
+
+		assertThat(sql).isEqualTo("SELECT dummy_entity.id1 AS id1 FROM dummy_entity WHERE dummy_entity.id1 = :id1 FOR UPDATE");
+	}
+
 	@Test // DATAJDBC-112
 	void cascadingDeleteFirstLevel() {
 
@@ -240,6 +251,47 @@ class SqlGeneratorUnitTests {
 		assertThat(sql).isEqualTo("DELETE FROM element WHERE element.dummy_entity = :id1");
 	}
 
+	@Test // GH-1978
+	void deleteByQuery() {
+
+		Query query = Query.query(Criteria.where("id").greaterThan(23L));
+		MapSqlParameterSource parameterSource = new MapSqlParameterSource();
+
+		String sql = sqlGenerator.createDeleteByQuery(query, parameterSource);
+
+		assertThat(sql).isEqualTo("DELETE FROM dummy_entity WHERE dummy_entity.id1 > :id1");
+	}
+
+	@Test // GH-1978
+	void cascadingDeleteInSubselectByPathFirstLevel() {
+
+		Query query = Query.query(Criteria.where("id").is(23L));
+		MapSqlParameterSource parameterSource = new MapSqlParameterSource();
+
+		String sql = sqlGenerator.createDeleteInSubselectByPath(query, parameterSource,
+				getPath("ref", DummyEntity.class));
+
+		assertThat(sql).isEqualTo(
+				"DELETE FROM referenced_entity WHERE referenced_entity.dummy_entity IN " +
+						"(SELECT dummy_entity.id1 AS id1 FROM dummy_entity WHERE dummy_entity.id1 = :id1)");
+	}
+
+	@Test // GH-1978
+	void cascadingDeleteInSubselectByPathSecondLevel() {
+
+		Query query = Query.query(Criteria.where("id").is(23L));
+		MapSqlParameterSource parameterSource = new MapSqlParameterSource();
+
+		String sql = sqlGenerator.createDeleteInSubselectByPath(query, parameterSource,
+				getPath("ref.further", DummyEntity.class));
+
+		assertThat(sql).isEqualTo(
+				"DELETE FROM second_level_referenced_entity " +
+						"WHERE second_level_referenced_entity.referenced_entity IN " +
+						"(SELECT referenced_entity.x_l1id FROM referenced_entity WHERE referenced_entity.dummy_entity IN " +
+						"(SELECT dummy_entity.id1 AS id1 FROM dummy_entity WHERE dummy_entity.id1 = :id1))");
+	}
+
 	@Test // DATAJDBC-101
 	void findAllSortedByUnsorted() {
 

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

@@ -26,6 +26,7 @@ import java.util.function.Function;
 import org.jspecify.annotations.Nullable;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+import org.springframework.data.relational.core.query.Query;
 import org.springframework.data.util.Pair;
 import org.springframework.util.Assert;
 
@@ -39,6 +40,7 @@ import org.springframework.util.Assert;
  * @author Tyler Van Gorder
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
+ * @author Jaeyeon Kim
  */
 public interface DbAction<T> {
 
@@ -219,6 +221,67 @@ public interface DbAction<T> {
 			}
 		}
 
+	/**
+	 * Represents a delete statement for aggregate root entities matching a given {@link Query}.
+	 *
+	 * @param <T> type of the entity for which this represents a database interaction.
+	 */
+	final class DeleteRootByQuery<T> implements DbAction<T> {
+
+		private final Class<T> entityType;
+
+		private final Query query;
+
+		DeleteRootByQuery(Class<T> entityType, Query query) {
+			this.entityType = entityType;
+			this.query = query;
+		}
+
+		@Override
+		public Class<T> getEntityType() {
+			return this.entityType;
+		}
+
+		public Query getQuery() {
+			return query;
+		}
+
+		public String toString() {
+			return "DbAction.DeleteRootByQuery(entityType=" + this.entityType + ", query=" + this.query + ")";
+		}
+	}
+
+	/**
+	 * Represents a delete statement for all entities that are reachable via a given path from the aggregate root,
+	 * filtered by a {@link Query}.
+	 *
+	 * @param <T> type of the entity for which this represents a database interaction.
+	 */
+	final class DeleteByQuery<T> implements WithPropertyPath<T> {
+
+		private final Query query;
+
+		private final PersistentPropertyPath<RelationalPersistentProperty> propertyPath;
+
+		DeleteByQuery(Query query, PersistentPropertyPath<RelationalPersistentProperty> propertyPath) {
+			this.query = query;
+			this.propertyPath = propertyPath;
+		}
+
+		@Override
+		public PersistentPropertyPath<RelationalPersistentProperty> propertyPath() {
+			return this.propertyPath;
+		}
+
+		public Query getQuery() {
+			return query;
+		}
+
+		public String toString() {
+			return "DbAction.DeleteByQuery(propertyPath=" + this.propertyPath() + ", query=" + this.query + ")";
+		}
+	}
+
 	/**
 	 * Represents an acquire lock statement for a aggregate root when only the ID is known.
 	 *
@@ -269,6 +332,37 @@ public interface DbAction<T> {
 		}
 	}
 
+	/**
+	 * Represents a {@code SELECT ... FOR UPDATE} statement on all aggregate roots of a given type,
+	 * filtered by a {@link Query}.
+	 *
+	 * @param <T> type of the root entity for which this represents a database interaction.
+	 */
+	final class AcquireLockAllRootByQuery<T> implements DbAction<T> {
+
+		private final Class<T> entityType;
+
+		private final Query query;
+
+		AcquireLockAllRootByQuery(Class<T> entityType, Query query) {
+			this.entityType = entityType;
+            this.query = query;
+        }
+
+		@Override
+		public Class<T> getEntityType() {
+			return this.entityType;
+		}
+
+		public Query getQuery() {
+			return query;
+		}
+
+		public String toString() {
+			return "DbAction.AcquireLockAllRootByQuery(entityType=" + this.entityType + ", query=" + this.query + ")";
+		}
+	}
+
 	/**
 	 * Represents a batch of {@link DbAction} that share a common value for a property of the action.
 	 *

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

@@ -26,6 +26,8 @@ import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.core.mapping.RelationalPredicates;
+import org.springframework.data.relational.core.query.CriteriaDefinition;
+import org.springframework.data.relational.core.query.Query;
 import org.springframework.util.Assert;
 
 /**
@@ -40,6 +42,7 @@ import org.springframework.util.Assert;
  * @author Tyler Van Gorder
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
+ * @author Jaeyeon Kim
  */
 public class RelationalEntityDeleteWriter implements EntityWriter<Object, MutableAggregateChange<?>> {
 
@@ -70,6 +73,42 @@ public class RelationalEntityDeleteWriter implements EntityWriter<Object, Mutabl
 		}
 	}
 
+	/**
+	 * Fills the provided {@link MutableAggregateChange} with the necessary {@link DbAction}s
+	 * to delete all aggregate roots matching the given {@link Query}.
+	 * This includes acquiring locks, deleting referenced entities, and deleting the root entities themselves.
+	 *
+	 * @param query the query used to select aggregate root IDs to delete. Must not be {@code null}.
+	 * @param aggregateChange The change object to which delete actions will be added. Must not be {@code null}.
+	 */
+	public void writeForQuery(Query query, MutableAggregateChange<?> aggregateChange) {
+
+		Class<?> entityType = aggregateChange.getEntityType();
+
+		CriteriaDefinition criteria = query.getCriteria().orElse(null);
+		if (criteria == null || criteria.isEmpty()) {
+			deleteAll(entityType).forEach(aggregateChange::addAction);
+			return;
+		}
+
+		List<DbAction<?>> deleteReferencedActions = new ArrayList<>();
+
+		forAllTableRepresentingPaths(entityType, p -> deleteReferencedActions.add(new DbAction.DeleteByQuery<>(query, p)));
+
+		Collections.reverse(deleteReferencedActions);
+
+		List<DbAction<?>> actions = new ArrayList<>();
+		if (!deleteReferencedActions.isEmpty()) {
+			actions.add(new DbAction.AcquireLockAllRootByQuery<>(entityType, query));
+		}
+		actions.addAll(deleteReferencedActions);
+
+		DbAction.DeleteRootByQuery<?> deleteRootByQuery = new DbAction.DeleteRootByQuery<>(entityType, query);
+		actions.add(deleteRootByQuery);
+
+		actions.forEach(aggregateChange::addAction);
+	}
+
 	private List<DbAction<?>> deleteAll(Class<?> entityType) {
 
 		List<DbAction<?>> deleteReferencedActions = new ArrayList<>();

spring-data-relational/src/test/java/org/springframework/data/relational/core/conversion/RelationalEntityDeleteWriterUnitTests.java

@@ -28,6 +28,8 @@ import org.springframework.data.relational.core.conversion.DbAction.DeleteAll;
 import org.springframework.data.relational.core.conversion.DbAction.DeleteAllRoot;
 import org.springframework.data.relational.core.conversion.DbAction.DeleteRoot;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
+import org.springframework.data.relational.core.query.Criteria;
+import org.springframework.data.relational.core.query.Query;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -40,6 +42,7 @@ import static org.assertj.core.api.Assertions.*;
  * @author Jens Schauder
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
+ * @author Jaeyeon Kim
  */
 @ExtendWith(MockitoExtension.class)
 public class RelationalEntityDeleteWriterUnitTests {
@@ -142,6 +145,42 @@ public class RelationalEntityDeleteWriterUnitTests {
 				);
 	}
 
+	@Test // GH-1978
+	void writeForQueryDeletesEntitiesByQueryAndReferencedEntities() {
+
+		MutableAggregateChange<SomeEntity> aggregateChange = MutableAggregateChange.forDelete(SomeEntity.class);
+		Query query = Query.query(Criteria.where("id").is(23L));
+
+		converter.writeForQuery(query, aggregateChange);
+
+		assertThat(extractActions(aggregateChange))
+				.extracting(DbAction::getClass, DbAction::getEntityType, DbActionTestSupport::extractPath)
+				.containsExactly(
+						Tuple.tuple(DbAction.AcquireLockAllRootByQuery.class, SomeEntity.class, ""),
+						Tuple.tuple(DbAction.DeleteByQuery.class, YetAnother.class, "other.yetAnother"),
+						Tuple.tuple(DbAction.DeleteByQuery.class, OtherEntity.class, "other"),
+						Tuple.tuple(DbAction.DeleteRootByQuery.class, SomeEntity.class, "")
+				);
+	}
+
+	@Test // GH-1978
+	void writeForQueryDeletesEntitiesByEmptyQueryAndReferencedEntities() {
+
+		MutableAggregateChange<SomeEntity> aggregateChange = MutableAggregateChange.forDelete(SomeEntity.class);
+		Query query = Query.query(Criteria.empty());
+
+		converter.writeForQuery(query, aggregateChange);
+
+		assertThat(extractActions(aggregateChange))
+				.extracting(DbAction::getClass, DbAction::getEntityType, DbActionTestSupport::extractPath)
+				.containsExactly(
+						Tuple.tuple(DbAction.AcquireLockAllRoot.class, SomeEntity.class, ""),
+						Tuple.tuple(DbAction.DeleteAll.class, YetAnother.class, "other.yetAnother"),
+						Tuple.tuple(DbAction.DeleteAll.class, OtherEntity.class, "other"),
+						Tuple.tuple(DbAction.DeleteAllRoot.class, SomeEntity.class, "")
+				);
+	}
+
 	private List<DbAction<?>> extractActions(MutableAggregateChange<?> aggregateChange) {
 
 		List<DbAction<?>> actions = new ArrayList<>();