GH-1978 JdbcAggregateOperations delete by query

Changed the steps as follows:
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 same subquery criteria.

Signed-off-by: JaeYeon Kim <JaeYeon.Kim@ibm.com>

Author: JaeYeon Kim

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

@@ -102,10 +102,10 @@ 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.DeleteRootByIdIn<?> deleteRootByIdIn) {
-			executionContext.executeDeleteRootByIdIn(deleteRootByIdIn);
-		} else if (action instanceof DbAction.DeleteByRootIdIn<?> deleteByRootIdIn) {
-			executionContext.executeDeleteByRootIdIn(deleteByRootIdIn);
+		} 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) {

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

@@ -40,7 +40,6 @@
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.mapping.PersistentPropertyPathAccessor;
 import org.springframework.data.relational.core.conversion.DbAction;
-import org.springframework.data.relational.core.conversion.SelectIdsDbActionExecutionResult;
 import org.springframework.data.relational.core.conversion.DbActionExecutionResult;
 import org.springframework.data.relational.core.conversion.IdValueSource;
 import org.springframework.data.relational.core.mapping.AggregatePath;
@@ -74,7 +73,6 @@ class JdbcAggregateChangeExecutionContext {
 	private final DataAccessStrategy accessStrategy;
 
 	private final Map<DbAction<?>, DbActionExecutionResult> results = new LinkedHashMap<>();
-	private final Map<DbAction.SelectIds<?>, SelectIdsDbActionExecutionResult> selectIdsDbActionExecutionResult = new LinkedHashMap<>();
 
 	JdbcAggregateChangeExecutionContext(JdbcConverter converter, DataAccessStrategy accessStrategy) {
 
@@ -172,32 +170,14 @@ <T> void executeDeleteAll(DbAction.DeleteAll<T> delete) {
 		accessStrategy.deleteAll(delete.getPropertyPath());
 	}
 
-	<T> void executeDeleteRootByIdIn(DbAction.DeleteRootByIdIn<T> deleteRootByIdIn) {
-		SelectIdsDbActionExecutionResult result = getRequiredSelectIdsResult(deleteRootByIdIn.getSelectIdsAction());
+	<T> void excuteDeleteRootByQuery(DbAction.DeleteRootByQuery<T> deleteRootByQuery) {
 
-		List<Object> rootIds = new ArrayList<>(result.getSelectedIds());
-		if (rootIds.isEmpty()) {
-			return;
-		}
-		accessStrategy.delete(rootIds, deleteRootByIdIn.getEntityType());
+		accessStrategy.deleteRootByQuery(deleteRootByQuery.getQuery(), deleteRootByQuery.getEntityType());
 	}
 
-	<T> void executeDeleteByRootIdIn(DbAction.DeleteByRootIdIn<T> deleteByRootIdIn) {
-		SelectIdsDbActionExecutionResult result = getRequiredSelectIdsResult(deleteByRootIdIn.getSelectIdsAction());
+	<T> void excuteDeleteByQuery(DbAction.DeleteByQuery<T> deleteByQuery) {
 
-		List<Object> rootIds = new ArrayList<>(result.getSelectedIds());
-		if (rootIds.isEmpty()) {
-			return;
-		}
-		accessStrategy.delete(rootIds, deleteByRootIdIn.getPropertyPath());
-	}
-
-	private SelectIdsDbActionExecutionResult getRequiredSelectIdsResult(DbAction.SelectIds selectIdsAction) {
-		SelectIdsDbActionExecutionResult result = selectIdsDbActionExecutionResult.get(selectIdsAction);
-		if (result == null) {
-			throw new IllegalArgumentException("Expected SelectIdsDbActionExecutionResult for given selectIdsAction but found none");
-		}
-		return result;
+		accessStrategy.deleteByQuery(deleteByQuery.getQuery(), deleteByQuery.getPropertyPath());
 	}
 
 	<T> void executeAcquireLock(DbAction.AcquireLockRoot<T> acquireLock) {
@@ -209,10 +189,7 @@ <T> void executeAcquireLockAllRoot(DbAction.AcquireLockAllRoot<T> acquireLock) {
 	}
 
 	<T> void executeAcquireLockRootByQuery(DbAction.AcquireLockAllRootByQuery<T> acquireLock) {
-
-		List<?> rootIds = accessStrategy.acquireLockAndFindIdsByQuery(acquireLock.getQuery(), LockMode.PESSIMISTIC_WRITE, acquireLock.getEntityType());
-
-		selectIdsDbActionExecutionResult.put(acquireLock, new SelectIdsDbActionExecutionResult(rootIds, acquireLock));
+		accessStrategy.acquireLockByQuery(acquireLock.getQuery(), LockMode.PESSIMISTIC_WRITE, acquireLock.getEntityType());
 	}
 
 	private void add(DbActionExecutionResult result) {

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

@@ -110,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));
@@ -121,8 +131,8 @@ public <T> void acquireLockAll(LockMode lockMode, Class<T> domainType) {
 	}
 
 	@Override
-	public <T> List<?> acquireLockAndFindIdsByQuery(Query query, LockMode lockMode, Class<T> domainType) {
-		return collect(das -> das.acquireLockAndFindIdsByQuery(query, lockMode, domainType));
+	public <T> void acquireLockByQuery(Query query, LockMode lockMode, Class<T> domainType) {
+		collectVoid(das -> das.acquireLockByQuery(query, lockMode, domainType));
 	}
 
 	@Override

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

@@ -178,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.
 	 *
@@ -196,16 +212,14 @@ 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} and return their identifiers.
-	 * The resulting SQL will include a {@code SELECT id FROM … WHERE … (LOCK CLAUSE)} to retrieve and lock the matching rows.
+	 * 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.
-	 * @return a {@link List} of ids corresponding to the rows locked by the query.
 	 */
-	<T> List<?> acquireLockAndFindIdsByQuery(Query query, LockMode lockMode, Class<T> domainType);
+	<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

@@ -39,7 +39,6 @@
 import org.springframework.data.relational.core.sql.LockMode;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.jdbc.core.RowMapper;
-import org.springframework.jdbc.core.SingleColumnRowMapper;
 import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
 import org.springframework.jdbc.core.namedparam.SqlParameterSource;
@@ -245,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) {
 
@@ -262,25 +285,12 @@ public <T> void acquireLockAll(LockMode lockMode, Class<T> domainType) {
 	}
 
 	@Override
-	public <T> List<?> acquireLockAndFindIdsByQuery(Query query, LockMode lockMode, Class<T> domainType) {
+	public <T> void acquireLockByQuery(Query query, LockMode lockMode, Class<T> domainType) {
 
 		MapSqlParameterSource parameterSource = new MapSqlParameterSource();
-		String acquireLockByQuerySql = sql(domainType).getAcquireLockAndFindIdsByQuery(query, parameterSource, lockMode);
+		String acquireLockByQuerySql = sql(domainType).getAcquireLockByQuery(query, parameterSource, lockMode);
 
-		RelationalPersistentEntity<?> entity = context.getRequiredPersistentEntity(domainType);
-		RelationalPersistentProperty idProperty = entity.getRequiredIdProperty();
-
-		return operations.query(acquireLockByQuerySql, parameterSource, getIdRowMapper(idProperty));
-	}
-
-	private RowMapper<?> getIdRowMapper(RelationalPersistentProperty idProperty) {
-		RelationalPersistentEntity<?> complexId = context.getPersistentEntity(idProperty.getType());
-
-		if (complexId == null) {
-			return SingleColumnRowMapper.newInstance(idProperty.getType(), converter.getConversionService());
-		} else {
-			return new EntityRowMapper<>(context.getRequiredPersistentEntity(idProperty.getType()), converter);
-		}
+		operations.query(acquireLockByQuerySql, parameterSource, ResultSet::next);
 	}
 
 	@Override

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

@@ -110,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);
@@ -121,8 +131,8 @@ public <T> void acquireLockAll(LockMode lockMode, Class<T> domainType) {
 	}
 
 	@Override
-	public <T> List<?> acquireLockAndFindIdsByQuery(Query query, LockMode lockMode, Class<T> domainType) {
-		return delegate.acquireLockAndFindIdsByQuery(query, lockMode, domainType);
+	public <T> void acquireLockByQuery(Query query, LockMode lockMode, Class<T> domainType) {
+		delegate.acquireLockByQuery(query, lockMode, domainType);
 	}
 
 	@Override

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

@@ -386,7 +386,7 @@ String getAcquireLockAll(LockMode lockMode) {
 	 * @param lockMode Lock clause mode.
 	 * @return the SQL statement as a {@link String}. Guaranteed to be not {@literal null}.
 	 */
-	String getAcquireLockAndFindIdsByQuery(Query query, MapSqlParameterSource parameterSource, LockMode lockMode) {
+	String getAcquireLockByQuery(Query query, MapSqlParameterSource parameterSource, LockMode lockMode) {
 		return this.createAcquireLockByQuery(query, parameterSource, lockMode);
 	}
 
@@ -505,6 +505,76 @@ 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())
+				.ifPresent(criteria ->
+						builder.where(queryMapper.getMappedObject(parameterSource, criteria, table, entity))
+				);
+
+		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 = createRootIdSubSelectByQuery(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 createRootIdSubSelectByQuery(Query query, MapSqlParameterSource parameterSource) {
+
+		Table table = this.getTable();
+
+		SelectBuilder.SelectWhere selectBuilder = StatementBuilder
+				.select(getIdColumns())
+				.from(table);
+
+		query.getCriteria()
+				.filter(criteria -> !criteria.isEmpty())
+				.ifPresent(criteria ->
+					selectBuilder.where(
+							queryMapper.getMappedObject(parameterSource, criteria, table, entity)
+					)
+				);
+
+		return selectBuilder.build();
+	}
+
 	/**
 	 * Constructs a where condition. The where condition will be of the form {@literal <columns> IN :bind-marker}
 	 */
@@ -614,10 +684,18 @@ private String createAcquireLockByQuery(Query query, MapSqlParameterSource param
 		Table table = this.getTable();
 
 		SelectBuilder.SelectWhere selectBuilder = StatementBuilder
-				.select(getIdColumns())
+				.select(getSingleNonNullColumn())
 				.from(table);
 
-		Select select = applyQueryOnSelect(query, parameterSource, selectBuilder)
+		query.getCriteria()
+				.filter(criteria -> !criteria.isEmpty())
+				.ifPresent(criteria ->
+						selectBuilder.where(
+								queryMapper.getMappedObject(parameterSource, criteria, table, entity)
+						)
+				);
+
+		Select select = selectBuilder
 				.lock(lockMode)
 				.build();
 

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

@@ -231,6 +231,16 @@ public void deleteAll(PersistentPropertyPath<RelationalPersistentProperty> prope
 		sqlSession().delete(statement, parameter);
 	}
 
+	@Override
+	public void deleteRootByQuery(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) {
 
@@ -255,7 +265,7 @@ public <T> void acquireLockAll(LockMode lockMode, Class<T> domainType) {
 	}
 
 	@Override
-	public <T> List<?> acquireLockAndFindIdsByQuery(Query query, LockMode lockMode, Class<T> domainType) {
+	public <T> void acquireLockByQuery(Query query, LockMode lockMode, Class<T> domainType) {
 		throw new UnsupportedOperationException("Not implemented");
 	}