diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/AggregateChangeExecutor.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/AggregateChangeExecutor.java
index 45b139b7ab..c547b6bfba 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/AggregateChangeExecutor.java
+++ b/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");
 		}
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateChangeExecutionContext.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateChangeExecutionContext.java
index b1f33efcaa..eaad0b3434 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateChangeExecutionContext.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateChangeExecutionContext.java
@@ -51,6 +51,7 @@
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
  * @author Mark Paluch
+ * @author Jaeyeon Kim
  */
 @SuppressWarnings("rawtypes")
 class JdbcAggregateChangeExecutionContext {
@@ -160,6 +161,16 @@ <T> void executeDeleteAll(DbAction.DeleteAll<T> delete) {
 		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 @@ <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);
 	}
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateOperations.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateOperations.java
index f98aad06c0..1a26009216 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateOperations.java
+++ b/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}.
 	 *
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateTemplate.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateTemplate.java
index b1c6e8b2da..dc3db167c8 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateTemplate.java
+++ b/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<>();
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/CascadingDataAccessStrategy.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/CascadingDataAccessStrategy.java
index 283612d8bf..50c574b455 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/CascadingDataAccessStrategy.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/CascadingDataAccessStrategy.java
@@ -49,6 +49,7 @@
  * @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 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));
@@ -142,6 +153,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));
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DataAccessStrategy.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DataAccessStrategy.java
index 7a41626841..66eee6bdf2 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DataAccessStrategy.java
+++ b/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.
 	 *
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategy.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategy.java
index e8cdee7dae..806614185a 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategy.java
+++ b/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 {
@@ -256,6 +257,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) {
 
@@ -272,6 +296,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) {
 
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DelegatingDataAccessStrategy.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DelegatingDataAccessStrategy.java
index f3d625c4a4..5190c9ab60 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DelegatingDataAccessStrategy.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DelegatingDataAccessStrategy.java
@@ -42,6 +42,7 @@
  * @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 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);
@@ -136,6 +147,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);
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/SqlGenerator.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/SqlGenerator.java
index 0b4d12abb9..04f69c2da1 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/SqlGenerator.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/SqlGenerator.java
@@ -61,6 +61,7 @@
  * @author Hari Ohm Prasath
  * @author Viktor Ardelean
  * @author Kurt Niemi
+ * @author Jaeyeon Kim
  */
 public class SqlGenerator {
 
@@ -374,6 +375,18 @@ String getAcquireLockAll(LockMode lockMode) {
 		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 @@ String createDeleteInByPath(PersistentPropertyPath<RelationalPersistentProperty>
 		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 @@ private String createAcquireLockAll(LockMode lockMode) {
 		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());
 	}
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/mybatis/MyBatisDataAccessStrategy.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/mybatis/MyBatisDataAccessStrategy.java
index 0ccc36e9f2..e77b133ade 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/mybatis/MyBatisDataAccessStrategy.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/mybatis/MyBatisDataAccessStrategy.java
@@ -74,6 +74,7 @@
  * @author Christopher Klein
  * @author Mikhail Polivakha
  * @author Sergey Korotaev
+ * @author Jaeyeon Kim
  */
 public class MyBatisDataAccessStrategy implements DataAccessStrategy {
 
@@ -255,6 +256,16 @@ public void deleteAll(PersistentPropertyPath<RelationalPersistentProperty> prope
 		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 <T> void acquireLockAll(LockMode lockMode, Class<T> domainType) {
 		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) {
 
diff --git a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/CompositeIdAggregateTemplateHsqlIntegrationTests.java b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/CompositeIdAggregateTemplateHsqlIntegrationTests.java
index 286f4604f5..41e732deae 100644
--- a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/CompositeIdAggregateTemplateHsqlIntegrationTests.java
+++ b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/CompositeIdAggregateTemplateHsqlIntegrationTests.java
@@ -32,6 +32,7 @@
 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 @@
  * 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 @@ void deleteMultipleSimpleEntityWithEmbeddedPk() {
 		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() {
 
diff --git a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/SqlGeneratorUnitTests.java b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/SqlGeneratorUnitTests.java
index 659b008035..8699ca388b 100644
--- a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/SqlGeneratorUnitTests.java
+++ b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/SqlGeneratorUnitTests.java
@@ -73,6 +73,7 @@
  * @author Diego Krupitza
  * @author Hari Ohm Prasath
  * @author Viktor Ardelean
+ * @author Jaeyeon Kim
  */
 @SuppressWarnings("Convert2MethodRef")
 class SqlGeneratorUnitTests {
@@ -165,6 +166,16 @@ void getAcquireLockAll() {
 				.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 @@ void deleteMapByPath() {
 		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() {
 
diff --git a/spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/DbAction.java b/spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/DbAction.java
index 8d2913f050..a2ce6e6ce4 100644
--- a/spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/DbAction.java
+++ b/spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/DbAction.java
@@ -26,6 +26,7 @@
 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 @@
  * @author Tyler Van Gorder
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
+ * @author Jaeyeon Kim
  */
 public interface DbAction<T> {
 
@@ -219,6 +221,67 @@ public String toString() {
 			}
 		}
 
+	/**
+	 * 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 String toString() {
 		}
 	}
 
+	/**
+	 * 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.
 	 *
diff --git a/spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/RelationalEntityDeleteWriter.java b/spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/RelationalEntityDeleteWriter.java
index d5ad10cd75..c457cb2d26 100644
--- a/spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/RelationalEntityDeleteWriter.java
+++ b/spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/RelationalEntityDeleteWriter.java
@@ -26,6 +26,8 @@
 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 @@
  * @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 void write(@Nullable Object id, MutableAggregateChange<?> aggregateChange
 		}
 	}
 
+	/**
+	 * 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<>();
diff --git a/spring-data-relational/src/test/java/org/springframework/data/relational/core/conversion/RelationalEntityDeleteWriterUnitTests.java b/spring-data-relational/src/test/java/org/springframework/data/relational/core/conversion/RelationalEntityDeleteWriterUnitTests.java
index 11e0238b95..16a28b0d8d 100644
--- a/spring-data-relational/src/test/java/org/springframework/data/relational/core/conversion/RelationalEntityDeleteWriterUnitTests.java
+++ b/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.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 @@
  * @author Jens Schauder
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
+ * @author Jaeyeon Kim
  */
 @ExtendWith(MockitoExtension.class)
 public class RelationalEntityDeleteWriterUnitTests {
@@ -142,6 +145,42 @@ public void deleteAllDoesNotDeleteReadOnlyReferences() {
 				);
 	}
 
+	@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<>();