Introduce Update annotation.

Switch update execution to an annotation based model that allows usage of both the classic update as well as the aggregation pipeline variant. Add the reactive variant of it.
Make sure to allow parameter binding for update expressions and verify method return types.
Update Javadoc and reference documentation.

See: #2107
Original Pull Request: #284

Author: christophstrobl

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/Update.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb.repository;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.core.annotation.AliasFor;
+
+/**
+ * Annotation to declare update operators directly on repository methods. Both attributes allow using a placeholder
+ * notation of {@code ?0}, {@code ?1} and so on. The update will be applied to documents matching the either method name
+ * derived or annotated query, but not to any custom implementation methods.
+ *
+ * @author Christoph Strobl
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE })
+@Documented
+public @interface Update {
+
+	/**
+	 * Takes a MongoDB JSON string to define the actual update to be executed.
+	 *
+	 * @return the MongoDB JSON string representation of the update. Empty string by default.
+	 * @see #update()
+	 */
+	@AliasFor("update")
+	String value() default "";
+
+	/**
+	 * Takes a MongoDB JSON string to define the actual update to be executed.
+	 *
+	 * @return the MongoDB JSON string representation of the update. Empty string by default.
+	 * @see <a href=
+	 *      "https://docs.mongodb.com/manual/tutorial/update-documents/">https://docs.mongodb.com/manual/tutorial/update-documents/</a>
+	 */
+	@AliasFor("value")
+	String update() default "";
+
+	/**
+	 * Takes a MongoDB JSON string representation of an aggregation pipeline to define the update stages to be executed.
+	 * <p>
+	 * This allows to e.g. define update statement that can evaluate conditionals based on a field value, etc.
+	 *
+	 * @return the MongoDB JSON string representation of the update pipeline. Empty array by default.
+	 * @see <a href=
+	 *      "https://docs.mongodb.com/manual/tutorial/update-documents-with-aggregation-pipeline/">https://docs.mongodb.com/manual/tutorial/update-documents-with-aggregation-pipeline</a>
+	 */
+	String[] pipeline() default {};
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/AbstractMongoQuery.java

@@ -15,30 +15,44 @@
  */
 package org.springframework.data.mongodb.repository.query;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import org.bson.Document;
 import org.bson.codecs.configuration.CodecRegistry;
-import org.springframework.data.domain.Pageable;
 import org.springframework.data.mapping.model.SpELExpressionEvaluator;
 import org.springframework.data.mongodb.core.ExecutableFindOperation.ExecutableFind;
 import org.springframework.data.mongodb.core.ExecutableFindOperation.FindWithQuery;
 import org.springframework.data.mongodb.core.ExecutableFindOperation.TerminatingFind;
+import org.springframework.data.mongodb.core.ExecutableUpdateOperation.ExecutableUpdate;
 import org.springframework.data.mongodb.core.MongoOperations;
+import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
+import org.springframework.data.mongodb.core.aggregation.AggregationUpdate;
+import org.springframework.data.mongodb.core.query.BasicUpdate;
 import org.springframework.data.mongodb.core.query.Query;
-import org.springframework.data.mongodb.core.query.Update;
+import org.springframework.data.mongodb.core.query.UpdateDefinition;
+import org.springframework.data.mongodb.repository.Update;
 import org.springframework.data.mongodb.repository.query.MongoQueryExecution.DeleteExecution;
 import org.springframework.data.mongodb.repository.query.MongoQueryExecution.GeoNearExecution;
 import org.springframework.data.mongodb.repository.query.MongoQueryExecution.PagedExecution;
 import org.springframework.data.mongodb.repository.query.MongoQueryExecution.PagingGeoNearExecution;
 import org.springframework.data.mongodb.repository.query.MongoQueryExecution.SlicedExecution;
+import org.springframework.data.mongodb.repository.query.MongoQueryExecution.UpdateExecution;
+import org.springframework.data.mongodb.util.json.ParameterBindingContext;
+import org.springframework.data.mongodb.util.json.ParameterBindingDocumentCodec;
 import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
 import org.springframework.data.spel.ExpressionDependencies;
+import org.springframework.data.util.Lazy;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.ExpressionParser;
+import org.springframework.lang.NonNull;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
 
 import com.mongodb.client.MongoDatabase;
 
@@ -55,8 +69,11 @@ public abstract class AbstractMongoQuery implements RepositoryQuery {
 	private final MongoQueryMethod method;
 	private final MongoOperations operations;
 	private final ExecutableFind<?> executableFind;
+	private final ExecutableUpdate<?> executableUpdate;
 	private final ExpressionParser expressionParser;
 	private final QueryMethodEvaluationContextProvider evaluationContextProvider;
+	private final Lazy<ParameterBindingDocumentCodec> codec = Lazy
+			.of(() -> new ParameterBindingDocumentCodec(getCodecRegistry()));
 
 	/**
 	 * Creates a new {@link AbstractMongoQuery} from the given {@link MongoQueryMethod} and {@link MongoOperations}.
@@ -81,6 +98,7 @@ public AbstractMongoQuery(MongoQueryMethod method, MongoOperations operations, E
 		Class<?> type = metadata.getCollectionEntity().getType();
 
 		this.executableFind = operations.query(type);
+		this.executableUpdate = operations.update(type);
 		this.expressionParser = expressionParser;
 		this.evaluationContextProvider = evaluationContextProvider;
 	}
@@ -130,7 +148,17 @@ private MongoQueryExecution getExecution(ConvertingParameterAccessor accessor, F
 
 		if (isDeleteQuery()) {
 			return new DeleteExecution(operations, method);
-		} else if (method.isGeoNearQuery() && method.isPageQuery()) {
+		}
+
+		if (method.isModifyingQuery()) {
+			if (isLimiting()) {
+				throw new IllegalStateException(
+						String.format("Update method must not be limiting. Offending method: %s", method));
+			}
+			return new UpdateExecution(executableUpdate, method, () -> createUpdate(accessor), accessor);
+		}
+
+		if (method.isGeoNearQuery() && method.isPageQuery()) {
 			return new PagingGeoNearExecution(operation, method, accessor, this);
 		} else if (method.isGeoNearQuery()) {
 			return new GeoNearExecution(operation, method, accessor);
@@ -139,11 +167,6 @@ private MongoQueryExecution getExecution(ConvertingParameterAccessor accessor, F
 		} else if (method.isStreamQuery()) {
 			return q -> operation.matching(q).stream();
 		} else if (method.isCollectionQuery()) {
-
-			if (method.isModifyingQuery()) {
-				return q -> new UpdatingCollectionExecution(accessor.getPageable(), accessor.getUpdate()).execute(q);
-			}
-
 			return q -> operation.matching(q.with(accessor.getPageable()).with(accessor.getSort())).all();
 		} else if (method.isPageQuery()) {
 			return new PagedExecution(operation, accessor.getPageable());
@@ -153,11 +176,6 @@ private MongoQueryExecution getExecution(ConvertingParameterAccessor accessor, F
 			return q -> operation.matching(q).exists();
 		} else {
 			return q -> {
-
-				if (method.isModifyingQuery()) {
-					return new UpdatingSingleEntityExecution(accessor.getUpdate()).execute(q);
-				}
-
 				TerminatingFind<?> find = operation.matching(q);
 				return isLimiting() ? find.firstValue() : find.oneValue();
 			};
@@ -217,6 +235,94 @@ protected Query createCountQuery(ConvertingParameterAccessor accessor) {
 		return applyQueryMetaAttributesWhenPresent(createQuery(accessor));
 	}
 
+	/**
+	 * Retrieves the {@link UpdateDefinition update} from the given
+	 * {@link org.springframework.data.mongodb.repository.query.MongoParameterAccessor#getUpdate() accessor} or creates
+	 * one via by parsing the annotated statement extracted from {@link Update}.
+	 *
+	 * @param accessor never {@literal null}.
+	 * @return the computed {@link UpdateDefinition}.
+	 * @throws IllegalStateException if no update could be found.
+	 * @since 3.4
+	 */
+	protected UpdateDefinition createUpdate(ConvertingParameterAccessor accessor) {
+
+		if (accessor.getUpdate() != null) {
+			return accessor.getUpdate();
+		}
+
+		if (method.hasAnnotatedUpdate()) {
+
+			Update updateSource = method.getUpdateSource();
+			if (StringUtils.hasText(updateSource.update())) {
+				return new BasicUpdate(bindParameters(updateSource.update(), accessor));
+			}
+			if (!ObjectUtils.isEmpty(updateSource.pipeline())) {
+				return AggregationUpdate.from(parseAggregationPipeline(updateSource.pipeline(), accessor));
+			}
+		}
+
+		throw new IllegalStateException(String.format("No Update provided for method %s.", method));
+	}
+
+	/**
+	 * Parse the given aggregation pipeline stages applying values to placeholders to compute the actual list of
+	 * {@link AggregationOperation operations}.
+	 *
+	 * @param sourcePipeline must not be {@literal null}.
+	 * @param accessor must not be {@literal null}.
+	 * @return the parsed aggregation pipeline.
+	 * @since 3.4
+	 */
+	protected List<AggregationOperation> parseAggregationPipeline(String[] sourcePipeline,
+			ConvertingParameterAccessor accessor) {
+
+		List<AggregationOperation> stages = new ArrayList<>(sourcePipeline.length);
+		for (String source : sourcePipeline) {
+			stages.add(computePipelineStage(source, accessor));
+		}
+		return stages;
+	}
+
+	private AggregationOperation computePipelineStage(String source, ConvertingParameterAccessor accessor) {
+		return ctx -> ctx.getMappedObject(bindParameters(source, accessor), getQueryMethod().getDomainClass());
+	}
+
+	protected Document decode(String source, ParameterBindingContext bindingContext) {
+		return getParameterBindingCodec().decode(source, bindingContext);
+	}
+
+	private Document bindParameters(String source, ConvertingParameterAccessor accessor) {
+		return decode(source, prepareBindingContext(source, accessor));
+	}
+
+	/**
+	 * Create the {@link ParameterBindingContext binding context} used for SpEL evaluation.
+	 *
+	 * @param source the JSON source.
+	 * @param accessor value provider for parameter binding.
+	 * @return never {@literal null}.
+	 * @since 3.4
+	 */
+	protected ParameterBindingContext prepareBindingContext(String source, ConvertingParameterAccessor accessor) {
+
+		ExpressionDependencies dependencies = getParameterBindingCodec().captureExpressionDependencies(source,
+				accessor::getBindableValue, expressionParser);
+
+		SpELExpressionEvaluator evaluator = getSpELExpressionEvaluatorFor(dependencies, accessor);
+		return new ParameterBindingContext(accessor::getBindableValue, evaluator);
+	}
+
+	/**
+	 * Obtain the {@link ParameterBindingDocumentCodec} used for parsing JSON expressions.
+	 *
+	 * @return never {@literal null}.
+	 * @since 3.4
+	 */
+	protected ParameterBindingDocumentCodec getParameterBindingCodec() {
+		return codec.get();
+	}
+
 	/**
 	 * Obtain a the {@link EvaluationContext} suitable to evaluate expressions backed by the given dependencies.
 	 *
@@ -278,53 +384,4 @@ protected CodecRegistry getCodecRegistry() {
 	 * @since 2.0.4
 	 */
 	protected abstract boolean isLimiting();
-
-	/**
-	 * {@link MongoQueryExecution} for collection returning find and update queries.
-	 *
-	 * @author Thomas Darimont
-	 */
-	final class UpdatingCollectionExecution implements MongoQueryExecution {
-
-		private final Pageable pageable;
-		private final Update update;
-
-		UpdatingCollectionExecution(Pageable pageable, Update update) {
-			this.pageable = pageable;
-			this.update = update;
-		}
-
-		@Override
-		public Object execute(Query query) {
-
-			MongoEntityMetadata<?> metadata = method.getEntityInformation();
-			return operations.findAndModify(query.with(pageable), update, metadata.getJavaType(),
-					metadata.getCollectionName());
-		}
-	}
-
-	/**
-	 * {@link MongoQueryExecution} to return a single entity with update.
-	 *
-	 * @author Thomas Darimont
-	 */
-	final class UpdatingSingleEntityExecution implements MongoQueryExecution {
-
-		private final Update update;
-
-		private UpdatingSingleEntityExecution(Update update) {
-			this.update = update;
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.mongodb.repository.AbstractMongoQuery.Execution#execute(org.springframework.data.mongodb.core.core.query.Query)
-		 */
-		@Override
-		public Object execute(Query query) {
-
-			MongoEntityMetadata<?> metadata = method.getEntityInformation();
-			return operations.findAndModify(query.limit(1), update, metadata.getJavaType(), metadata.getCollectionName());
-		}
-	}
 }