switch to commons branch

Reformat code. Merge if-statements.

Original pull request: #889.

Jenkinsfile

@@ -93,7 +93,7 @@ pipeline {
 
 		stage("Test other configurations") {
 			when {
-				anyOf {
+				allOf {
 					branch 'master'
 					not { triggeredBy 'UpstreamCause' }
 				}

README.adoc

@@ -10,7 +10,7 @@ Key functional areas of Spring Data MongoDB are a POJO centric model for interac
 
 == Code of Conduct
 
-This project is governed by the link:CODE_OF_CONDUCT.adoc[Spring Code of Conduct]. By participating, you are expected to uphold this code of conduct. Please report unacceptable behavior to spring-code-of-conduct@pivotal.io.
+This project is governed by the https://github.com/spring-projects/.github/blob/e3cc2ff230d8f1dca06535aa6b5a4a23815861d4/CODE_OF_CONDUCT.md[Spring Code of Conduct]. By participating, you are expected to uphold this code of conduct. Please report unacceptable behavior to spring-code-of-conduct@pivotal.io.
 
 == Getting Started
 

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>3.1.0-SNAPSHOT</version>
+	<version>3.1.0-STATIC-METADATA-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>
@@ -26,8 +26,8 @@
 	<properties>
 		<project.type>multi</project.type>
 		<dist.id>spring-data-mongodb</dist.id>
-		<springdata.commons>2.4.0-SNAPSHOT</springdata.commons>
-		<mongo>4.0.4</mongo>
+		<springdata.commons>2.4.0-BUILD-TIME-DOMAIN-TYPE-METADATA-SNAPSHOT</springdata.commons>
+		<mongo>4.1.0</mongo>
 		<mongo.reactivestreams>${mongo}</mongo.reactivestreams>
 		<jmh.version>1.19</jmh.version>
 	</properties>

spring-data-mongodb-benchmarks/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>3.1.0-SNAPSHOT</version>
+		<version>3.1.0-STATIC-METADATA-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb-distribution/pom.xml

@@ -14,7 +14,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>3.1.0-SNAPSHOT</version>
+		<version>3.1.0-STATIC-METADATA-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/pom.xml

@@ -11,7 +11,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>3.1.0-SNAPSHOT</version>
+		<version>3.1.0-STATIC-METADATA-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 
@@ -136,6 +136,13 @@
 			<optional>true</optional>
 		</dependency>
 
+		<dependency>
+			<groupId>io.reactivex.rxjava3</groupId>
+			<artifactId>rxjava</artifactId>
+			<version>${rxjava3}</version>
+			<optional>true</optional>
+		</dependency>
+
 		<!-- CDI -->
 		<!-- Dependency order required to build against CDI 1.0 and test with CDI 2.0 -->
 		<dependency>
@@ -192,7 +199,14 @@
 		<dependency>
 			<groupId>org.hibernate</groupId>
 			<artifactId>hibernate-validator</artifactId>
-			<version>5.2.4.Final</version>
+			<version>5.4.3.Final</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.glassfish</groupId>
+			<artifactId>javax.el</artifactId>
+			<version>3.0.1-b11</version>
 			<scope>test</scope>
 		</dependency>
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/MongoDatabaseUtils.java

@@ -79,7 +79,7 @@ public class MongoDatabaseUtils {
 	 * @param factory the {@link MongoDatabaseFactory} to get the {@link MongoDatabase} from.
 	 * @return the {@link MongoDatabase} that is potentially associated with a transactional {@link ClientSession}.
 	 */
-	public static MongoDatabase getDatabase(String dbName, MongoDatabaseFactory factory) {
+	public static MongoDatabase getDatabase(@Nullable String dbName, MongoDatabaseFactory factory) {
 		return doGetMongoDatabase(dbName, factory, SessionSynchronization.ON_ACTUAL_TRANSACTION);
 	}
 
@@ -94,7 +94,7 @@ public class MongoDatabaseUtils {
 	 * @param sessionSynchronization the synchronization to use. Must not be {@literal null}.
 	 * @return the {@link MongoDatabase} that is potentially associated with a transactional {@link ClientSession}.
 	 */
-	public static MongoDatabase getDatabase(String dbName, MongoDatabaseFactory factory,
+	public static MongoDatabase getDatabase(@Nullable String dbName, MongoDatabaseFactory factory,
 			SessionSynchronization sessionSynchronization) {
 		return doGetMongoDatabase(dbName, factory, sessionSynchronization);
 	}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/EnableMongoAuditing.java

@@ -61,8 +61,8 @@ public @interface EnableMongoAuditing {
 	boolean modifyOnCreate() default true;
 
 	/**
-	 * Configures a {@link DateTimeProvider} bean name that allows customizing the {@link org.joda.time.DateTime} to be
-	 * used for setting creation and modification dates.
+	 * Configures a {@link DateTimeProvider} bean name that allows customizing the timestamp to be used for setting
+	 * creation and modification dates.
 	 *
 	 * @return empty {@link String} by default.
 	 */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/EnableReactiveMongoAuditing.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2020 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.config;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.context.annotation.Import;
+import org.springframework.data.auditing.DateTimeProvider;
+import org.springframework.data.domain.ReactiveAuditorAware;
+
+/**
+ * Annotation to enable auditing in MongoDB using reactive infrastructure via annotation configuration.
+ *
+ * @author Mark Paluch
+ * @since 3.1
+ */
+@Inherited
+@Documented
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Import(ReactiveMongoAuditingRegistrar.class)
+public @interface EnableReactiveMongoAuditing {
+
+	/**
+	 * Configures the {@link ReactiveAuditorAware} bean to be used to lookup the current principal.
+	 *
+	 * @return empty {@link String} by default.
+	 */
+	String auditorAwareRef() default "";
+
+	/**
+	 * Configures whether the creation and modification dates are set. Defaults to {@literal true}.
+	 *
+	 * @return {@literal true} by default.
+	 */
+	boolean setDates() default true;
+
+	/**
+	 * Configures whether the entity shall be marked as modified on creation. Defaults to {@literal true}.
+	 *
+	 * @return {@literal true} by default.
+	 */
+	boolean modifyOnCreate() default true;
+
+	/**
+	 * Configures a {@link DateTimeProvider} bean name that allows customizing the timestamp to be used for setting
+	 * creation and modification dates.
+	 *
+	 * @return empty {@link String} by default.
+	 */
+	String dateTimeProviderRef() default "";
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoAuditingRegistrar.java

@@ -17,7 +17,6 @@ package org.springframework.data.mongodb.config;
 
 import java.lang.annotation.Annotation;
 
-import org.springframework.beans.factory.FactoryBean;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
@@ -28,14 +27,8 @@ import org.springframework.data.auditing.IsNewAwareAuditingHandler;
 import org.springframework.data.auditing.config.AuditingBeanDefinitionRegistrarSupport;
 import org.springframework.data.auditing.config.AuditingConfiguration;
 import org.springframework.data.config.ParsingUtils;
-import org.springframework.data.mapping.context.MappingContext;
-import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
-import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
-import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
 import org.springframework.data.mongodb.core.mapping.event.AuditingEntityCallback;
-import org.springframework.data.mongodb.core.mapping.event.ReactiveAuditingEntityCallback;
 import org.springframework.util.Assert;
-import org.springframework.util.ClassUtils;
 
 /**
  * {@link ImportBeanDefinitionRegistrar} to enable {@link EnableMongoAuditing} annotation.
@@ -46,9 +39,6 @@ import org.springframework.util.ClassUtils;
  */
 class MongoAuditingRegistrar extends AuditingBeanDefinitionRegistrarSupport {
 
-	private static boolean PROJECT_REACTOR_AVAILABLE = ClassUtils.isPresent("reactor.core.publisher.Mono",
-			MongoAuditingRegistrar.class.getClassLoader());
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.auditing.config.AuditingBeanDefinitionRegistrarSupport#getAnnotation()
@@ -91,7 +81,7 @@ class MongoAuditingRegistrar extends AuditingBeanDefinitionRegistrarSupport {
 
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(IsNewAwareAuditingHandler.class);
 
-		BeanDefinitionBuilder definition = BeanDefinitionBuilder.genericBeanDefinition(MongoMappingContextLookup.class);
+		BeanDefinitionBuilder definition = BeanDefinitionBuilder.genericBeanDefinition(PersistentEntitiesFactoryBean.class);
 		definition.setAutowireMode(AbstractBeanDefinition.AUTOWIRE_CONSTRUCTOR);
 
 		builder.addConstructorArgValue(definition.getBeanDefinition());
@@ -116,68 +106,6 @@ class MongoAuditingRegistrar extends AuditingBeanDefinitionRegistrarSupport {
 
 		registerInfrastructureBeanWithId(listenerBeanDefinitionBuilder.getBeanDefinition(),
 				AuditingEntityCallback.class.getName(), registry);
-
-		if (PROJECT_REACTOR_AVAILABLE) {
-			registerReactiveAuditingEntityCallback(registry, auditingHandlerDefinition.getSource());
-		}
 	}
 
-	private void registerReactiveAuditingEntityCallback(BeanDefinitionRegistry registry, Object source) {
-
-		BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(ReactiveAuditingEntityCallback.class);
-
-		builder.addConstructorArgValue(ParsingUtils.getObjectFactoryBeanDefinition(getAuditingHandlerBeanName(), registry));
-		builder.getRawBeanDefinition().setSource(source);
-
-		registerInfrastructureBeanWithId(builder.getBeanDefinition(), ReactiveAuditingEntityCallback.class.getName(),
-				registry);
-	}
-
-	/**
-	 * Simple helper to be able to wire the {@link MappingContext} from a {@link MappingMongoConverter} bean available in
-	 * the application context.
-	 *
-	 * @author Oliver Gierke
-	 */
-	static class MongoMappingContextLookup
-			implements FactoryBean<MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty>> {
-
-		private final MappingMongoConverter converter;
-
-		/**
-		 * Creates a new {@link MongoMappingContextLookup} for the given {@link MappingMongoConverter}.
-		 *
-		 * @param converter must not be {@literal null}.
-		 */
-		public MongoMappingContextLookup(MappingMongoConverter converter) {
-			this.converter = converter;
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.beans.factory.FactoryBean#getObject()
-		 */
-		@Override
-		public MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty> getObject() throws Exception {
-			return converter.getMappingContext();
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.beans.factory.FactoryBean#getObjectType()
-		 */
-		@Override
-		public Class<?> getObjectType() {
-			return MappingContext.class;
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.beans.factory.FactoryBean#isSingleton()
-		 */
-		@Override
-		public boolean isSingleton() {
-			return true;
-		}
-	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/PersistentEntitiesFactoryBean.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2020 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.config;
+
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.data.mapping.context.PersistentEntities;
+import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
+
+/**
+ * Simple helper to be able to wire the {@link PersistentEntities} from a {@link MappingMongoConverter} bean available
+ * in the application context.
+ *
+ * @author Oliver Gierke
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ * @since 3.1
+ */
+class PersistentEntitiesFactoryBean implements FactoryBean<PersistentEntities> {
+
+	private final MappingMongoConverter converter;
+
+	/**
+	 * Creates a new {@link PersistentEntitiesFactoryBean} for the given {@link MappingMongoConverter}.
+	 *
+	 * @param converter must not be {@literal null}.
+	 */
+	public PersistentEntitiesFactoryBean(MappingMongoConverter converter) {
+		this.converter = converter;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.FactoryBean#getObject()
+	 */
+	@Override
+	public PersistentEntities getObject() {
+		return PersistentEntities.of(converter.getMappingContext());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.FactoryBean#getObjectType()
+	 */
+	@Override
+	public Class<?> getObjectType() {
+		return PersistentEntities.class;
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/ReactiveMongoAuditingRegistrar.java

@@ -0,0 +1,97 @@
+/*
+ * Copyright 2020 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.config;
+
+import java.lang.annotation.Annotation;
+
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
+import org.springframework.core.type.AnnotationMetadata;
+import org.springframework.data.auditing.ReactiveIsNewAwareAuditingHandler;
+import org.springframework.data.auditing.config.AuditingBeanDefinitionRegistrarSupport;
+import org.springframework.data.auditing.config.AuditingConfiguration;
+import org.springframework.data.config.ParsingUtils;
+import org.springframework.data.mongodb.core.mapping.event.ReactiveAuditingEntityCallback;
+import org.springframework.util.Assert;
+
+/**
+ * {@link ImportBeanDefinitionRegistrar} to enable {@link EnableReactiveMongoAuditing} annotation.
+ *
+ * @author Mark Paluch
+ * @since 3.1
+ */
+class ReactiveMongoAuditingRegistrar extends AuditingBeanDefinitionRegistrarSupport {
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.auditing.config.AuditingBeanDefinitionRegistrarSupport#getAnnotation()
+	 */
+	@Override
+	protected Class<? extends Annotation> getAnnotation() {
+		return EnableReactiveMongoAuditing.class;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.auditing.config.AuditingBeanDefinitionRegistrarSupport#getAuditingHandlerBeanName()
+	 */
+	@Override
+	protected String getAuditingHandlerBeanName() {
+		return "reactiveMongoAuditingHandler";
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.auditing.config.AuditingBeanDefinitionRegistrarSupport#getAuditHandlerBeanDefinitionBuilder(org.springframework.data.auditing.config.AuditingConfiguration)
+	 */
+	@Override
+	protected BeanDefinitionBuilder getAuditHandlerBeanDefinitionBuilder(AuditingConfiguration configuration) {
+
+		Assert.notNull(configuration, "AuditingConfiguration must not be null!");
+
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(ReactiveIsNewAwareAuditingHandler.class);
+
+		BeanDefinitionBuilder definition = BeanDefinitionBuilder.genericBeanDefinition(PersistentEntitiesFactoryBean.class);
+		definition.setAutowireMode(AbstractBeanDefinition.AUTOWIRE_CONSTRUCTOR);
+
+		builder.addConstructorArgValue(definition.getBeanDefinition());
+		return configureDefaultAuditHandlerAttributes(configuration, builder);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.auditing.config.AuditingBeanDefinitionRegistrarSupport#registerAuditListener(org.springframework.beans.factory.config.BeanDefinition, org.springframework.beans.factory.support.BeanDefinitionRegistry)
+	 */
+	@Override
+	protected void registerAuditListenerBeanDefinition(BeanDefinition auditingHandlerDefinition,
+			BeanDefinitionRegistry registry) {
+
+		Assert.notNull(auditingHandlerDefinition, "BeanDefinition must not be null!");
+		Assert.notNull(registry, "BeanDefinitionRegistry must not be null!");
+
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(ReactiveAuditingEntityCallback.class);
+
+		builder.addConstructorArgValue(ParsingUtils.getObjectFactoryBeanDefinition(getAuditingHandlerBeanName(), registry));
+		builder.getRawBeanDefinition().setSource(auditingHandlerDefinition.getSource());
+
+		registerInfrastructureBeanWithId(builder.getBeanDefinition(), ReactiveAuditingEntityCallback.class.getName(),
+				registry);
+	}
+
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/AggregationUtil.java

@@ -28,6 +28,7 @@ import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
 import org.springframework.data.mongodb.core.aggregation.CountOperation;
+import org.springframework.data.mongodb.core.aggregation.RelaxedTypeBasedAggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.TypeBasedAggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
 import org.springframework.data.mongodb.core.convert.QueryMapper;
@@ -75,12 +76,17 @@ class AggregationUtil {
 			return context;
 		}
 
-		if (aggregation instanceof TypedAggregation) {
-			return new TypeBasedAggregationOperationContext(((TypedAggregation) aggregation).getInputType(), mappingContext,
-					queryMapper);
+		if (!(aggregation instanceof TypedAggregation)) {
+			return Aggregation.DEFAULT_CONTEXT;
 		}
 
-		return Aggregation.DEFAULT_CONTEXT;
+		Class<?> inputType = ((TypedAggregation) aggregation).getInputType();
+
+		if (aggregation.getPipeline().containsUnionWith()) {
+			return new RelaxedTypeBasedAggregationOperationContext(inputType, mappingContext, queryMapper);
+		}
+
+		return new TypeBasedAggregationOperationContext(inputType, mappingContext, queryMapper);
 	}
 
 	/**

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/DefaultBulkOperations.java

@@ -24,8 +24,9 @@ import java.util.stream.Collectors;
 import org.bson.Document;
 import org.bson.conversions.Bson;
 import org.springframework.context.ApplicationEventPublisher;
-import org.springframework.dao.support.PersistenceExceptionTranslator;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.data.mapping.callback.EntityCallbacks;
+import org.springframework.data.mongodb.BulkOperationException;
 import org.springframework.data.mongodb.core.convert.QueryMapper;
 import org.springframework.data.mongodb.core.convert.UpdateMapper;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
@@ -46,6 +47,7 @@ import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 
+import com.mongodb.MongoBulkWriteException;
 import com.mongodb.WriteConcern;
 import com.mongodb.bulk.BulkWriteResult;
 import com.mongodb.client.MongoCollection;
@@ -62,6 +64,7 @@ import com.mongodb.client.model.*;
  * @author Jens Schauder
  * @author Michail Nikolaev
  * @author Roman Puchkovskiy
+ * @author Jacob Botuck
  * @since 1.9
  */
 class DefaultBulkOperations implements BulkOperations {
@@ -71,7 +74,6 @@ class DefaultBulkOperations implements BulkOperations {
 	private final BulkOperationContext bulkOperationContext;
 	private final List<SourceAwareWriteModelHolder> models = new ArrayList<>();
 
-	private PersistenceExceptionTranslator exceptionTranslator;
 	private @Nullable WriteConcern defaultWriteConcern;
 
 	private BulkWriteOptions bulkOptions;
@@ -95,19 +97,9 @@ class DefaultBulkOperations implements BulkOperations {
 		this.mongoOperations = mongoOperations;
 		this.collectionName = collectionName;
 		this.bulkOperationContext = bulkOperationContext;
-		this.exceptionTranslator = new MongoExceptionTranslator();
 		this.bulkOptions = getBulkWriteOptions(bulkOperationContext.getBulkMode());
 	}
 
-	/**
-	 * Configures the {@link PersistenceExceptionTranslator} to be used. Defaults to {@link MongoExceptionTranslator}.
-	 *
-	 * @param exceptionTranslator can be {@literal null}.
-	 */
-	public void setExceptionTranslator(@Nullable PersistenceExceptionTranslator exceptionTranslator) {
-		this.exceptionTranslator = exceptionTranslator == null ? new MongoExceptionTranslator() : exceptionTranslator;
-	}
-
 	/**
 	 * Configures the default {@link WriteConcern} to be used. Defaults to {@literal null}.
 	 *
@@ -314,11 +306,26 @@ class DefaultBulkOperations implements BulkOperations {
 			collection = collection.withWriteConcern(defaultWriteConcern);
 		}
 
-		return collection.bulkWrite( //
-				models.stream() //
-						.map(this::extractAndMapWriteModel) //
-						.collect(Collectors.toList()), //
-				bulkOptions);
+		try {
+
+			return collection.bulkWrite( //
+					models.stream() //
+							.map(this::extractAndMapWriteModel) //
+							.collect(Collectors.toList()), //
+					bulkOptions);
+		} catch (RuntimeException ex) {
+
+			if (ex instanceof MongoBulkWriteException) {
+
+				MongoBulkWriteException mongoBulkWriteException = (MongoBulkWriteException) ex;
+				if (mongoBulkWriteException.getWriteConcernError() != null) {
+					throw new DataIntegrityViolationException(ex.getMessage(), ex);
+				}
+				throw new BulkOperationException(ex.getMessage(), mongoBulkWriteException);
+			}
+
+			throw ex;
+		}
 	}
 
 	private WriteModel<Document> extractAndMapWriteModel(SourceAwareWriteModelHolder it) {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java

@@ -1184,6 +1184,29 @@ public interface MongoOperations extends FluentMongoOperations {
 	 */
 	long count(Query query, String collectionName);
 
+	/**
+	 * Estimate the number of documents, in the collection {@link #getCollectionName(Class) identified by the given type},
+	 * based on collection statistics.
+	 *
+	 * @param entityClass must not be {@literal null}.
+	 * @return the estimated number of documents.
+	 * @since 3.1
+	 */
+	default long estimatedCount(Class<?> entityClass) {
+
+		Assert.notNull(entityClass, "Entity class must not be null!");
+		return estimatedCount(getCollectionName(entityClass));
+	}
+
+	/**
+	 * Estimate the number of documents in the given collection based on collection statistics.
+	 *
+	 * @param collectionName must not be {@literal null}.
+	 * @return the estimated number of documents.
+	 * @since 3.1
+	 */
+	long estimatedCount(String collectionName);
+
 	/**
 	 * Returns the number of documents for the given {@link Query} by querying the given collection using the given entity
 	 * class to map the given {@link Query}. <br />

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java

@@ -28,6 +28,7 @@ import org.bson.Document;
 import org.bson.conversions.Bson;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -154,6 +155,7 @@ import com.mongodb.client.result.UpdateResult;
  * @author Cimon Lucas
  * @author Michael J. Simons
  * @author Roman Puchkovskiy
+ * @author Yadhukrishna S Pai
  */
 public class MongoTemplate implements MongoOperations, ApplicationContextAware, IndexOperationsProvider {
 
@@ -757,7 +759,6 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 				new BulkOperationContext(mode, Optional.ofNullable(getPersistentEntity(entityType)), queryMapper, updateMapper,
 						eventPublisher, entityCallbacks));
 
-		operations.setExceptionTranslator(exceptionTranslator);
 		operations.setDefaultWriteConcern(writeConcern);
 
 		return operations;
@@ -1134,6 +1135,19 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 				collection -> collection.countDocuments(CountQuery.of(filter).toQueryDocument(), options));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.MongoOperations#estimatedCount(java.lang.String)
+	 */
+	@Override
+	public long estimatedCount(String collectionName) {
+		return doEstimatedCount(collectionName, new EstimatedDocumentCountOptions());
+	}
+
+	protected long doEstimatedCount(String collectionName, EstimatedDocumentCountOptions options) {
+		return execute(collectionName, collection -> collection.estimatedDocumentCount(options));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.MongoOperations#insert(java.lang.Object)
@@ -1963,9 +1977,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 
 		Assert.notNull(aggregation, "Aggregation pipeline must not be null!");
 
-		AggregationOperationContext context = new TypeBasedAggregationOperationContext(aggregation.getInputType(),
-				mappingContext, queryMapper);
-		return aggregate(aggregation, inputCollectionName, outputType, context);
+		return aggregate(aggregation, inputCollectionName, outputType, null);
 	}
 
 	/* (non-Javadoc)
@@ -2119,7 +2131,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 
 			List<Document> rawResult = new ArrayList<>();
 
-			Class<?> domainType = aggregation instanceof TypedAggregation ? ((TypedAggregation) aggregation).getInputType()
+			Class<?> domainType = aggregation instanceof TypedAggregation ? ((TypedAggregation<?>) aggregation).getInputType()
 					: null;
 
 			Optional<Collation> collation = Optionals.firstNonEmpty(options::getCollation,
@@ -2135,11 +2147,23 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 			}
 
 			options.getComment().ifPresent(aggregateIterable::comment);
+			options.getHint().ifPresent(aggregateIterable::hint);
 
 			if (options.hasExecutionTimeLimit()) {
 				aggregateIterable = aggregateIterable.maxTime(options.getMaxTime().toMillis(), TimeUnit.MILLISECONDS);
 			}
 
+			if (options.isSkipResults()) {
+
+				// toCollection only allowed for $out and $merge if those are the last stages
+				if (aggregation.getPipeline().isOutOrMerge()) {
+					aggregateIterable.toCollection();
+				} else {
+					aggregateIterable.first();
+				}
+				return new AggregationResults<>(Collections.emptyList(), new Document());
+			}
+
 			MongoIterable<O> iterable = aggregateIterable.map(val -> {
 
 				rawResult.add(val);
@@ -2182,6 +2206,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 			}
 
 			options.getComment().ifPresent(cursor::comment);
+			options.getHint().ifPresent(cursor::hint);
 
 			Class<?> domainType = aggregation instanceof TypedAggregation ? ((TypedAggregation) aggregation).getInputType()
 					: null;
@@ -3002,8 +3027,8 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 		private final List<Document> arrayFilters;
 		private final FindAndModifyOptions options;
 
-		FindAndModifyCallback(Document query, Document fields, Document sort, Object update,
-				List<Document> arrayFilters, FindAndModifyOptions options) {
+		FindAndModifyCallback(Document query, Document fields, Document sort, Object update, List<Document> arrayFilters,
+				FindAndModifyOptions options) {
 
 			this.query = query;
 			this.fields = fields;
@@ -3360,8 +3385,8 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 		 * @param exceptionTranslator
 		 * @param objectReadCallback
 		 */
-		CloseableIterableCursorAdapter(MongoIterable<Document> cursor,
-				PersistenceExceptionTranslator exceptionTranslator, DocumentCallback<T> objectReadCallback) {
+		CloseableIterableCursorAdapter(MongoIterable<Document> cursor, PersistenceExceptionTranslator exceptionTranslator,
+				DocumentCallback<T> objectReadCallback) {
 
 			this.cursor = cursor.iterator();
 			this.exceptionTranslator = exceptionTranslator;

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoContext.java

@@ -15,11 +15,15 @@
  */
 package org.springframework.data.mongodb.core;
 
-import org.reactivestreams.Publisher;
-import org.springframework.util.Assert;
 import reactor.core.publisher.Mono;
 import reactor.util.context.Context;
 
+import java.util.function.Function;
+
+import org.reactivestreams.Publisher;
+
+import org.springframework.util.Assert;
+
 import com.mongodb.reactivestreams.client.ClientSession;
 
 /**
@@ -29,7 +33,7 @@ import com.mongodb.reactivestreams.client.ClientSession;
  * @author Christoph Strobl
  * @author Mark Paluch
  * @since 2.1
- * @see Mono#subscriberContext()
+ * @see Mono#deferContextual(Function)
  * @see Context
  */
 public class ReactiveMongoContext {
@@ -46,8 +50,14 @@ public class ReactiveMongoContext {
 	 */
 	public static Mono<ClientSession> getSession() {
 
-		return Mono.subscriberContext().filter(ctx -> ctx.hasKey(SESSION_KEY))
-				.flatMap(ctx -> ctx.<Mono<ClientSession>> get(SESSION_KEY));
+		return Mono.deferContextual(ctx -> {
+
+			if (ctx.hasKey(SESSION_KEY)) {
+				return ctx.<Mono<ClientSession>> get(SESSION_KEY);
+			}
+
+			return Mono.empty();
+		});
 	}
 
 	/**

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java

@@ -980,6 +980,29 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
 	 */
 	Mono<Long> count(Query query, @Nullable Class<?> entityClass, String collectionName);
 
+	/**
+	 * Estimate the number of documents, in the collection {@link #getCollectionName(Class) identified by the given type},
+	 * based on collection statistics.
+	 *
+	 * @param entityClass must not be {@literal null}.
+	 * @return a {@link Mono} emitting the estimated number of documents.
+	 * @since 3.1
+	 */
+	default Mono<Long> estimatedCount(Class<?> entityClass) {
+
+		Assert.notNull(entityClass, "Entity class must not be null!");
+		return estimatedCount(getCollectionName(entityClass));
+	}
+
+	/**
+	 * Estimate the number of documents in the given collection based on collection statistics.
+	 *
+	 * @param collectionName must not be {@literal null}.
+	 * @return a {@link Mono} emitting the estimated number of documents.
+	 * @since 3.1
+	 */
+	Mono<Long> estimatedCount(String collectionName);
+
 	/**
 	 * Insert the object into the collection for the entity type of the object to save.
 	 * <p/>

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java

@@ -36,6 +36,7 @@ import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -116,16 +117,7 @@ import com.mongodb.CursorType;
 import com.mongodb.MongoException;
 import com.mongodb.ReadPreference;
 import com.mongodb.WriteConcern;
-import com.mongodb.client.model.CountOptions;
-import com.mongodb.client.model.CreateCollectionOptions;
-import com.mongodb.client.model.DeleteOptions;
-import com.mongodb.client.model.FindOneAndDeleteOptions;
-import com.mongodb.client.model.FindOneAndReplaceOptions;
-import com.mongodb.client.model.FindOneAndUpdateOptions;
-import com.mongodb.client.model.ReplaceOptions;
-import com.mongodb.client.model.ReturnDocument;
-import com.mongodb.client.model.UpdateOptions;
-import com.mongodb.client.model.ValidationOptions;
+import com.mongodb.client.model.*;
 import com.mongodb.client.model.changestream.FullDocument;
 import com.mongodb.client.result.DeleteResult;
 import com.mongodb.client.result.InsertOneResult;
@@ -154,6 +146,7 @@ import com.mongodb.reactivestreams.client.MongoDatabase;
  * @author Christoph Strobl
  * @author Roman Puchkovskiy
  * @author Mathieu Ouellet
+ * @author Yadhukrishna S Pai
  * @since 2.0
  */
 public class ReactiveMongoTemplate implements ReactiveMongoOperations, ApplicationContextAware {
@@ -580,7 +573,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 				ReactiveMongoTemplate.this);
 
 		return Flux.from(action.doInSession(operations)) //
-				.subscriberContext(ctx -> ReactiveMongoContext.setSession(ctx, Mono.just(session)));
+				.contextWrite(ctx -> ReactiveMongoContext.setSession(ctx, Mono.just(session)));
 	}
 
 	/*
@@ -1013,11 +1006,14 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		}
 
 		ReadDocumentCallback<O> readCallback = new ReadDocumentCallback<>(mongoConverter, outputType, collectionName);
-		return execute(collectionName, collection -> aggregateAndMap(collection, pipeline, options, readCallback,
-				aggregation instanceof TypedAggregation ? ((TypedAggregation) aggregation).getInputType() : null));
+		return execute(collectionName,
+				collection -> aggregateAndMap(collection, pipeline, aggregation.getPipeline().isOutOrMerge(), options,
+						readCallback,
+						aggregation instanceof TypedAggregation ? ((TypedAggregation<?>) aggregation).getInputType() : null));
 	}
 
 	private <O> Flux<O> aggregateAndMap(MongoCollection<Document> collection, List<Document> pipeline,
+			boolean isOutOrMerge,
 			AggregationOptions options, ReadDocumentCallback<O> readCallback, @Nullable Class<?> inputType) {
 
 		AggregatePublisher<Document> cursor = collection.aggregate(pipeline, Document.class)
@@ -1028,6 +1024,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		}
 
 		options.getComment().ifPresent(cursor::comment);
+		options.getHint().ifPresent(cursor::hint);
 
 		Optionals.firstNonEmpty(options::getCollation, () -> operations.forType(inputType).getCollation()) //
 				.map(Collation::toMongoCollation) //
@@ -1037,6 +1034,10 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 			cursor = cursor.maxTime(options.getMaxTime().toMillis(), TimeUnit.MILLISECONDS);
 		}
 
+		if (options.isSkipResults()) {
+			return (isOutOrMerge ? Flux.from(cursor.toCollection()) : Flux.from(cursor.first())).thenMany(Mono.empty());
+		}
+
 		return Flux.from(cursor).concatMap(readCallback::doWith);
 	}
 
@@ -1247,6 +1248,15 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		});
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#estimatedCount(java.lang.String)
+	 */
+	@Override
+	public Mono<Long> estimatedCount(String collectionName) {
+		return doEstimatedCount(collectionName, new EstimatedDocumentCountOptions());
+	}
+
 	/**
 	 * Run the actual count operation against the collection with given name.
 	 *
@@ -1261,6 +1271,11 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 				collection -> collection.countDocuments(CountQuery.of(filter).toQueryDocument(), options));
 	}
 
+	protected Mono<Long> doEstimatedCount(String collectionName, EstimatedDocumentCountOptions options) {
+
+		return createMono(collectionName, collection -> collection.estimatedDocumentCount(options));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#insert(reactor.core.publisher.Mono)

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/AbstractAggregationExpression.java

@@ -29,8 +29,11 @@ import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 
 /**
+ * Support class for {@link AggregationExpression} implementations.
+ *
  * @author Christoph Strobl
  * @author Matt Morrissette
+ * @author Mark Paluch
  * @since 1.10
  */
 abstract class AbstractAggregationExpression implements AggregationExpression {
@@ -49,7 +52,6 @@ abstract class AbstractAggregationExpression implements AggregationExpression {
 		return toDocument(this.value, context);
 	}
 
-	@SuppressWarnings("unchecked")
 	public Document toDocument(Object value, AggregationOperationContext context) {
 		return new Document(getMongoMethod(), unpack(value, context));
 	}
@@ -101,17 +103,19 @@ abstract class AbstractAggregationExpression implements AggregationExpression {
 		return value;
 	}
 
+	@SuppressWarnings("unchecked")
 	protected List<Object> append(Object value, Expand expandList) {
 
 		if (this.value instanceof List) {
 
-			List<Object> clone = new ArrayList<Object>((List) this.value);
+			List<Object> clone = new ArrayList<>((List<Object>) this.value);
 
 			if (value instanceof Collection && Expand.EXPAND_VALUES.equals(expandList)) {
 				clone.addAll((Collection<?>) value);
 			} else {
 				clone.add(value);
 			}
+
 			return clone;
 		}
 
@@ -129,25 +133,72 @@ abstract class AbstractAggregationExpression implements AggregationExpression {
 		return append(value, Expand.EXPAND_VALUES);
 	}
 
-	@SuppressWarnings("unchecked")
-	protected java.util.Map<String, Object> append(String key, Object value) {
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	protected Map<String, Object> append(String key, Object value) {
 
 		Assert.isInstanceOf(Map.class, this.value, "Value must be a type of Map!");
 
-		java.util.Map<String, Object> clone = new LinkedHashMap<>((java.util.Map) this.value);
+		Map<String, Object> clone = new LinkedHashMap<>((java.util.Map) this.value);
 		clone.put(key, value);
 		return clone;
 
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	protected Map<String, Object> remove(String key) {
+
+		Assert.isInstanceOf(Map.class, this.value, "Value must be a type of Map!");
+
+		Map<String, Object> clone = new LinkedHashMap<>((java.util.Map) this.value);
+		clone.remove(key);
+		return clone;
+	}
+
+	/**
+	 * Append the given key at the position in the underlying {@link LinkedHashMap}.
+	 *
+	 * @param index
+	 * @param key
+	 * @param value
+	 * @return
+	 * @since 3.1
+	 */
+	@SuppressWarnings({ "unchecked" })
+	protected Map<String, Object> appendAt(int index, String key, Object value) {
+
+		Assert.isInstanceOf(Map.class, this.value, "Value must be a type of Map!");
+
+		Map<String, Object> clone = new LinkedHashMap<>();
+
+		int i = 0;
+		for (Map.Entry<String, Object> entry : ((Map<String, Object>) this.value).entrySet()) {
+
+			if (i == index) {
+				clone.put(key, value);
+			}
+			if (!entry.getKey().equals(key)) {
+				clone.put(entry.getKey(), entry.getValue());
+			}
+			i++;
+		}
+		if (i <= index) {
+			clone.put(key, value);
+		}
+		return clone;
+
+	}
+
+	@SuppressWarnings({ "rawtypes" })
 	protected List<Object> values() {
 
 		if (value instanceof List) {
 			return new ArrayList<Object>((List) value);
 		}
+
 		if (value instanceof java.util.Map) {
 			return new ArrayList<Object>(((java.util.Map) value).values());
 		}
+
 		return new ArrayList<>(Collections.singletonList(value));
 	}
 
@@ -177,7 +228,7 @@ abstract class AbstractAggregationExpression implements AggregationExpression {
 
 		Assert.isInstanceOf(Map.class, this.value, "Value must be a type of Map!");
 
-		return (T) ((java.util.Map<String, Object>) this.value).get(key);
+		return (T) ((Map<String, Object>) this.value).get(key);
 	}
 
 	/**
@@ -187,11 +238,11 @@ abstract class AbstractAggregationExpression implements AggregationExpression {
 	 * @return
 	 */
 	@SuppressWarnings("unchecked")
-	protected java.util.Map<String, Object> argumentMap() {
+	protected Map<String, Object> argumentMap() {
 
 		Assert.isInstanceOf(Map.class, this.value, "Value must be a type of Map!");
 
-		return Collections.unmodifiableMap((java.util.Map) value);
+		return Collections.unmodifiableMap((java.util.Map<String, Object>) value);
 	}
 
 	/**
@@ -208,7 +259,7 @@ abstract class AbstractAggregationExpression implements AggregationExpression {
 			return false;
 		}
 
-		return ((java.util.Map<String, Object>) this.value).containsKey(key);
+		return ((Map<String, Object>) this.value).containsKey(key);
 	}
 
 	protected abstract String getMongoMethod();

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/AddFieldsOperation.java

@@ -99,6 +99,10 @@ public class AddFieldsOperation extends DocumentEnhancingOperation {
 		return new AddFieldsOperationBuilder(getValueMap());
 	}
 
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.DocumentEnhancingOperation#mongoOperator()
+	 */
 	@Override
 	protected String mongoOperator() {
 		return "$addFields";

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/Aggregation.java

@@ -96,7 +96,7 @@ public class Aggregation {
 	public static final AggregationOperationContext DEFAULT_CONTEXT = AggregationOperationRenderer.DEFAULT_CONTEXT;
 	public static final AggregationOptions DEFAULT_OPTIONS = newAggregationOptions().build();
 
-	protected final List<AggregationOperation> operations;
+	protected final AggregationPipeline pipeline;
 	private final AggregationOptions options;
 
 	/**
@@ -139,7 +139,7 @@ public class Aggregation {
 	public Aggregation withOptions(AggregationOptions options) {
 
 		Assert.notNull(options, "AggregationOptions must not be null.");
-		return new Aggregation(this.operations, options);
+		return new Aggregation(this.pipeline.getOperations(), options);
 	}
 
 	/**
@@ -202,26 +202,10 @@ public class Aggregation {
 		Assert.notNull(aggregationOperations, "AggregationOperations must not be null!");
 		Assert.notNull(options, "AggregationOptions must not be null!");
 
-		// check $out/$merge is the last operation if it exists
-		for (AggregationOperation aggregationOperation : aggregationOperations) {
-
-			if (aggregationOperation instanceof OutOperation && !isLast(aggregationOperation, aggregationOperations)) {
-				throw new IllegalArgumentException("The $out operator must be the last stage in the pipeline.");
-			}
-
-			if (aggregationOperation instanceof MergeOperation && !isLast(aggregationOperation, aggregationOperations)) {
-				throw new IllegalArgumentException("The $merge operator must be the last stage in the pipeline.");
-			}
-		}
-
-		this.operations = aggregationOperations;
+		this.pipeline = new AggregationPipeline(aggregationOperations);
 		this.options = options;
 	}
 
-	private boolean isLast(AggregationOperation aggregationOperation, List<AggregationOperation> aggregationOperations) {
-		return aggregationOperations.indexOf(aggregationOperation) == aggregationOperations.size() - 1;
-	}
-
 	/**
 	 * Get the {@link AggregationOptions}.
 	 *
@@ -661,9 +645,9 @@ public class Aggregation {
 	/**
 	 * Creates a new {@link RedactOperation} that can restrict the content of a document based on information stored
 	 * within the document itself.
-	 * 
+	 *
 	 * <pre class="code">
-	 * 
+	 *
 	 * Aggregation.redact(ConditionalOperators.when(Criteria.where("level").is(5)) //
 	 * 		.then(RedactOperation.PRUNE) //
 	 * 		.otherwise(RedactOperation.DESCEND));
@@ -718,7 +702,15 @@ public class Aggregation {
 	 * @since 2.1
 	 */
 	public List<Document> toPipeline(AggregationOperationContext rootContext) {
-		return AggregationOperationRenderer.toDocument(operations, rootContext);
+		return pipeline.toDocuments(rootContext);
+	}
+
+	/**
+	 * @return the {@link AggregationPipeline}.
+	 * @since 3.0.2
+	 */
+	public AggregationPipeline getPipeline() {
+		return pipeline;
 	}
 
 	/**

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/AggregationOperation.java

@@ -54,4 +54,15 @@ public interface AggregationOperation {
 	default List<Document> toPipelineStages(AggregationOperationContext context) {
 		return Collections.singletonList(toDocument(context));
 	}
+
+	/**
+	 * Return the MongoDB operator that is used for this {@link AggregationOperation}. Aggregation operations should
+	 * implement this method to avoid document rendering.
+	 *
+	 * @return the operator used for this {@link AggregationOperation}.
+	 * @since 3.0.2
+	 */
+	default String getOperator() {
+		return toDocument(Aggregation.DEFAULT_CONTEXT).keySet().iterator().next();
+	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/AggregationOptions.java

@@ -32,6 +32,7 @@ import org.springframework.util.Assert;
  * @author Oliver Gierke
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Yadhukrishna S Pai
  * @see Aggregation#withOptions(AggregationOptions)
  * @see TypedAggregation#withOptions(AggregationOptions)
  * @since 1.6
@@ -45,13 +46,16 @@ public class AggregationOptions {
 	private static final String COLLATION = "collation";
 	private static final String COMMENT = "comment";
 	private static final String MAX_TIME = "maxTimeMS";
+	private static final String HINT = "hint";
 
 	private final boolean allowDiskUse;
 	private final boolean explain;
 	private final Optional<Document> cursor;
 	private final Optional<Collation> collation;
 	private final Optional<String> comment;
+	private final Optional<Document> hint;
 	private Duration maxTime = Duration.ZERO;
+	private ResultOptions resultOptions = ResultOptions.READ;
 
 	/**
 	 * Creates a new {@link AggregationOptions}.
@@ -70,13 +74,13 @@ public class AggregationOptions {
 	 * @param allowDiskUse whether to off-load intensive sort-operations to disk.
 	 * @param explain whether to get the execution plan for the aggregation instead of the actual results.
 	 * @param cursor can be {@literal null}, used to pass additional options (such as {@code batchSize}) to the
-	 * aggregation.
+	 *          aggregation.
 	 * @param collation collation for string comparison. Can be {@literal null}.
 	 * @since 2.0
 	 */
 	public AggregationOptions(boolean allowDiskUse, boolean explain, @Nullable Document cursor,
 			@Nullable Collation collation) {
-		this(allowDiskUse, explain, cursor, collation, null);
+		this(allowDiskUse, explain, cursor, collation, null, null);
 	}
 
 	/**
@@ -85,19 +89,37 @@ public class AggregationOptions {
 	 * @param allowDiskUse whether to off-load intensive sort-operations to disk.
 	 * @param explain whether to get the execution plan for the aggregation instead of the actual results.
 	 * @param cursor can be {@literal null}, used to pass additional options (such as {@code batchSize}) to the
-	 * aggregation.
+	 *          aggregation.
 	 * @param collation collation for string comparison. Can be {@literal null}.
 	 * @param comment execution comment. Can be {@literal null}.
 	 * @since 2.2
 	 */
 	public AggregationOptions(boolean allowDiskUse, boolean explain, @Nullable Document cursor,
 			@Nullable Collation collation, @Nullable String comment) {
+		this(allowDiskUse, explain, cursor, collation, comment, null);
+	}
+
+	/**
+	 * Creates a new {@link AggregationOptions}.
+	 *
+	 * @param allowDiskUse whether to off-load intensive sort-operations to disk.
+	 * @param explain whether to get the execution plan for the aggregation instead of the actual results.
+	 * @param cursor can be {@literal null}, used to pass additional options (such as {@code batchSize}) to the
+	 *          aggregation.
+	 * @param collation collation for string comparison. Can be {@literal null}.
+	 * @param comment execution comment. Can be {@literal null}.
+	 * @param hint can be {@literal null}, used to provide an index that would be forcibly used by query optimizer.
+	 * @since 3.1
+	 */
+	private AggregationOptions(boolean allowDiskUse, boolean explain, @Nullable Document cursor,
+			@Nullable Collation collation, @Nullable String comment, @Nullable Document hint) {
 
 		this.allowDiskUse = allowDiskUse;
 		this.explain = explain;
 		this.cursor = Optional.ofNullable(cursor);
 		this.collation = Optional.ofNullable(collation);
 		this.comment = Optional.ofNullable(comment);
+		this.hint = Optional.ofNullable(hint);
 	}
 
 	/**
@@ -129,8 +151,9 @@ public class AggregationOptions {
 		Collation collation = document.containsKey(COLLATION) ? Collation.from(document.get(COLLATION, Document.class))
 				: null;
 		String comment = document.getString(COMMENT);
+		Document hint = document.get(HINT, Document.class);
 
-		AggregationOptions options = new AggregationOptions(allowDiskUse, explain, cursor, collation, comment);
+		AggregationOptions options = new AggregationOptions(allowDiskUse, explain, cursor, collation, comment, hint);
 		if (document.containsKey(MAX_TIME)) {
 			options.maxTime = Duration.ofMillis(document.getLong(MAX_TIME));
 		}
@@ -211,6 +234,16 @@ public class AggregationOptions {
 		return comment;
 	}
 
+	/**
+	 * Get the hint used to to fulfill the aggregation.
+	 *
+	 * @return never {@literal null}.
+	 * @since 3.1
+	 */
+	public Optional<Document> getHint() {
+		return hint;
+	}
+
 	/**
 	 * @return the time limit for processing. {@link Duration#ZERO} is used for the default unbounded behavior.
 	 * @since 3.0
@@ -219,6 +252,15 @@ public class AggregationOptions {
 		return maxTime;
 	}
 
+	/**
+	 * @return {@literal true} to skip results when running an aggregation. Useful in combination with {@code $merge} or
+	 *         {@code $out}.
+	 * @since 3.0.2
+	 */
+	public boolean isSkipResults() {
+		return ResultOptions.SKIP.equals(resultOptions);
+	}
+
 	/**
 	 * Returns a new potentially adjusted copy for the given {@code aggregationCommandObject} with the configuration
 	 * applied.
@@ -238,6 +280,10 @@ public class AggregationOptions {
 			result.put(EXPLAIN, explain);
 		}
 
+		if (result.containsKey(HINT)) {
+			hint.ifPresent(val -> result.append(HINT, val));
+		}
+
 		if (!result.containsKey(CURSOR)) {
 			cursor.ifPresent(val -> result.put(CURSOR, val));
 		}
@@ -267,6 +313,7 @@ public class AggregationOptions {
 		cursor.ifPresent(val -> document.put(CURSOR, val));
 		collation.ifPresent(val -> document.append(COLLATION, val.toDocument()));
 		comment.ifPresent(val -> document.append(COMMENT, val));
+		hint.ifPresent(val -> document.append(HINT, val));
 
 		if (hasExecutionTimeLimit()) {
 			document.append(MAX_TIME, maxTime.toMillis());
@@ -308,7 +355,9 @@ public class AggregationOptions {
 		private @Nullable Document cursor;
 		private @Nullable Collation collation;
 		private @Nullable String comment;
+		private @Nullable Document hint;
 		private @Nullable Duration maxTime;
+		private @Nullable ResultOptions resultOptions;
 
 		/**
 		 * Defines whether to off-load intensive sort-operations to disk.
@@ -385,11 +434,24 @@ public class AggregationOptions {
 			return this;
 		}
 
+		/**
+		 * Define a hint that is used by query optimizer to to fulfill the aggregation.
+		 *
+		 * @param hint can be {@literal null}.
+		 * @return this.
+		 * @since 3.1
+		 */
+		public Builder hint(@Nullable Document hint) {
+
+			this.hint = hint;
+			return this;
+		}
+
 		/**
 		 * Set the time limit for processing.
 		 *
 		 * @param maxTime {@link Duration#ZERO} is used for the default unbounded behavior. {@link Duration#isNegative()
-		 * Negative} values will be ignored.
+		 *          Negative} values will be ignored.
 		 * @return this.
 		 * @since 3.0
 		 */
@@ -399,6 +461,20 @@ public class AggregationOptions {
 			return this;
 		}
 
+		/**
+		 * Run the aggregation, but do NOT read the aggregation result from the store. <br />
+		 * If the expected result of the aggregation is rather large, eg. when using an {@literal $out} operation, this
+		 * option allows to execute the aggregation without having the cursor return the operation result.
+		 *
+		 * @return this.
+		 * @since 3.0.2
+		 */
+		public Builder skipOutput() {
+
+			this.resultOptions = ResultOptions.SKIP;
+			return this;
+		}
+
 		/**
 		 * Returns a new {@link AggregationOptions} instance with the given configuration.
 		 *
@@ -406,12 +482,30 @@ public class AggregationOptions {
 		 */
 		public AggregationOptions build() {
 
-			AggregationOptions options = new AggregationOptions(allowDiskUse, explain, cursor, collation, comment);
+			AggregationOptions options = new AggregationOptions(allowDiskUse, explain, cursor, collation, comment, hint);
 			if (maxTime != null) {
 				options.maxTime = maxTime;
 			}
+			if (resultOptions != null) {
+				options.resultOptions = resultOptions;
+			}
 
 			return options;
 		}
 	}
+
+	/**
+	 * @since 3.0
+	 */
+	private enum ResultOptions {
+
+		/**
+		 * Just do it!, and do not read the operation result.
+		 */
+		SKIP,
+		/**
+		 * Read the aggregation result from the cursor.
+		 */
+		READ;
+	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/AggregationPipeline.java

@@ -0,0 +1,162 @@
+/*
+ * Copyright 2020 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.core.aggregation;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Predicate;
+
+import org.bson.Document;
+import org.springframework.util.Assert;
+
+/**
+ * The {@link AggregationPipeline} holds the collection of {@link AggregationOperation aggregation stages}.
+ *
+ * @author Christoph Strobl
+ * @author Mark Paluch
+ * @since 3.0.2
+ */
+public class AggregationPipeline {
+
+	private final List<AggregationOperation> pipeline;
+
+	/**
+	 * Create an empty pipeline
+	 */
+	public AggregationPipeline() {
+		this(new ArrayList<>());
+	}
+
+	/**
+	 * Create a new pipeline with given {@link AggregationOperation stages}.
+	 *
+	 * @param aggregationOperations must not be {@literal null}.
+	 */
+	public AggregationPipeline(List<AggregationOperation> aggregationOperations) {
+
+		Assert.notNull(aggregationOperations, "AggregationOperations must not be null!");
+		pipeline = new ArrayList<>(aggregationOperations);
+	}
+
+	/**
+	 * Append the given {@link AggregationOperation stage} to the pipeline.
+	 *
+	 * @param aggregationOperation must not be {@literal null}.
+	 * @return this.
+	 */
+	public AggregationPipeline add(AggregationOperation aggregationOperation) {
+
+		Assert.notNull(aggregationOperation, "AggregationOperation must not be null!");
+
+		pipeline.add(aggregationOperation);
+		return this;
+	}
+
+	/**
+	 * Get the list of {@link AggregationOperation aggregation stages}.
+	 *
+	 * @return never {@literal null}.
+	 */
+	public List<AggregationOperation> getOperations() {
+		return Collections.unmodifiableList(pipeline);
+	}
+
+	List<Document> toDocuments(AggregationOperationContext context) {
+
+		verify();
+		return AggregationOperationRenderer.toDocument(pipeline, context);
+	}
+
+	/**
+	 * @return {@literal true} if the last aggregation stage is either {@literal $out} or {@literal $merge}.
+	 */
+	public boolean isOutOrMerge() {
+
+		if (isEmpty()) {
+			return false;
+		}
+
+		AggregationOperation operation = pipeline.get(pipeline.size() - 1);
+		return isOut(operation) || isMerge(operation);
+	}
+
+	void verify() {
+
+		// check $out/$merge is the last operation if it exists
+		for (AggregationOperation operation : pipeline) {
+
+			if (isOut(operation) && !isLast(operation)) {
+				throw new IllegalArgumentException("The $out operator must be the last stage in the pipeline.");
+			}
+
+			if (isMerge(operation) && !isLast(operation)) {
+				throw new IllegalArgumentException("The $merge operator must be the last stage in the pipeline.");
+			}
+		}
+	}
+
+	/**
+	 * Return whether this aggregation pipeline defines a {@code $unionWith} stage that may contribute documents from
+	 * other collections. Checking for presence of union stages is useful when attempting to determine the aggregation
+	 * element type for mapping metadata computation.
+	 *
+	 * @return {@literal true} the aggregation pipeline makes use of {@code $unionWith}.
+	 * @since 3.1
+	 */
+	public boolean containsUnionWith() {
+		return containsOperation(AggregationPipeline::isUnionWith);
+	}
+
+	/**
+	 * @return {@literal true} if the pipeline does not contain any stages.
+	 * @since 3.1
+	 */
+	public boolean isEmpty() {
+		return pipeline.isEmpty();
+	}
+
+	private boolean containsOperation(Predicate<AggregationOperation> predicate) {
+
+		if (isEmpty()) {
+			return false;
+		}
+
+		for (AggregationOperation element : pipeline) {
+			if (predicate.test(element)) {
+				return true;
+			}
+		}
+
+		return false;
+	}
+
+	private boolean isLast(AggregationOperation aggregationOperation) {
+		return pipeline.indexOf(aggregationOperation) == pipeline.size() - 1;
+	}
+
+	private static boolean isUnionWith(AggregationOperation operator) {
+		return operator instanceof UnionWithOperation || operator.getOperator().equals("$unionWith");
+	}
+
+	private static boolean isMerge(AggregationOperation operator) {
+		return operator instanceof MergeOperation || operator.getOperator().equals("$merge");
+	}
+
+	private static boolean isOut(AggregationOperation operator) {
+		return operator instanceof OutOperation || operator.getOperator().equals("$out");
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/AggregationUpdate.java

@@ -139,7 +139,7 @@ public class AggregationUpdate extends Aggregation implements UpdateDefinition {
 		setOperation.getFields().forEach(it -> {
 			keysTouched.add(it.getName());
 		});
-		operations.add(setOperation);
+		pipeline.add(setOperation);
 		return this;
 	}
 
@@ -155,7 +155,7 @@ public class AggregationUpdate extends Aggregation implements UpdateDefinition {
 
 		Assert.notNull(unsetOperation, "UnsetOperation must not be null!");
 
-		operations.add(unsetOperation);
+		pipeline.add(unsetOperation);
 		keysTouched.addAll(unsetOperation.removedFieldNames());
 		return this;
 	}
@@ -172,7 +172,7 @@ public class AggregationUpdate extends Aggregation implements UpdateDefinition {
 	public AggregationUpdate replaceWith(ReplaceWithOperation replaceWithOperation) {
 
 		Assert.notNull(replaceWithOperation, "ReplaceWithOperation must not be null!");
-		operations.add(replaceWithOperation);
+		pipeline.add(replaceWithOperation);
 		return this;
 	}
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/BucketAutoOperation.java

@@ -15,12 +15,11 @@
  */
 package org.springframework.data.mongodb.core.aggregation;
 
+import org.bson.Document;
 import org.springframework.data.mongodb.core.aggregation.BucketAutoOperation.BucketAutoOperationOutputBuilder;
 import org.springframework.data.mongodb.core.aggregation.BucketOperationSupport.OutputBuilder;
 import org.springframework.util.Assert;
 
-import org.bson.Document;
-
 /**
  * Encapsulates the aggregation framework {@code $bucketAuto}-operation. <br />
  * Bucket stage is typically used with {@link Aggregation} and {@code $facet}. Categorizes incoming documents into a
@@ -106,7 +105,16 @@ public class BucketAutoOperation extends BucketOperationSupport<BucketAutoOperat
 
 		options.putAll(super.toDocument(context));
 
-		return new Document("$bucketAuto", options);
+		return new Document(getOperator(), options);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$bucketAuto";
 	}
 
 	/**

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/BucketOperation.java

@@ -20,21 +20,19 @@ import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
+import org.bson.Document;
 import org.springframework.data.mongodb.core.aggregation.BucketOperation.BucketOperationOutputBuilder;
 import org.springframework.util.Assert;
 
-import org.bson.Document;
-
 /**
  * Encapsulates the aggregation framework {@code $bucket}-operation. <br />
- *
  * Bucket stage is typically used with {@link Aggregation} and {@code $facet}. Categorizes incoming documents into
  * groups, called buckets, based on a specified expression and bucket boundaries. <br />
- *
  * We recommend to use the static factory method {@link Aggregation#bucket(String)} instead of creating instances of
  * this class directly.
  *
- * @see <a href="https://docs.mongodb.org/manual/reference/aggregation/bucket/">https://docs.mongodb.org/manual/reference/aggregation/bucket/</a>
+ * @see <a href=
+ *      "https://docs.mongodb.org/manual/reference/aggregation/bucket/">https://docs.mongodb.org/manual/reference/aggregation/bucket/</a>
  * @see BucketOperationSupport
  * @author Mark Paluch
  * @since 1.10
@@ -103,7 +101,16 @@ public class BucketOperation extends BucketOperationSupport<BucketOperation, Buc
 
 		options.putAll(super.toDocument(context));
 
-		return new Document("$bucket", options);
+		return new Document(getOperator(), options);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$bucket";
 	}
 
 	/**
@@ -204,8 +211,8 @@ public class BucketOperation extends BucketOperationSupport<BucketOperation, Buc
 			extends ExpressionBucketOperationBuilderSupport<BucketOperationOutputBuilder, BucketOperation> {
 
 		/**
-		 * Creates a new {@link ExpressionBucketOperationBuilderSupport} for the given value, {@link BucketOperation}
-		 * and parameters.
+		 * Creates a new {@link ExpressionBucketOperationBuilderSupport} for the given value, {@link BucketOperation} and
+		 * parameters.
 		 *
 		 * @param expression must not be {@literal null}.
 		 * @param operation must not be {@literal null}.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/CountOperation.java

@@ -49,7 +49,12 @@ public class CountOperation implements FieldsExposingAggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$count", fieldName);
+		return new Document(getOperator(), fieldName);
+	}
+
+	@Override
+	public String getOperator() {
+		return "$count";
 	}
 
 	/* (non-Javadoc)

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/DocumentEnhancingOperation.java

@@ -71,6 +71,15 @@ abstract class DocumentEnhancingOperation implements InheritsFieldsAggregationOp
 	 */
 	protected abstract String mongoOperator();
 
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return mongoOperator();
+	}
+
 	/**
 	 * @return the raw value map
 	 */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/ExposedFieldsAggregationOperationContext.java

@@ -149,4 +149,12 @@ class ExposedFieldsAggregationOperationContext implements AggregationOperationCo
 		}
 		return null;
 	}
+
+	/**
+	 * @return obtain the root context used to resolve references.
+	 * @since 3.1
+	 */
+	AggregationOperationContext getRootContext() {
+		return rootContext;
+	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/FacetOperation.java

@@ -20,12 +20,11 @@ import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
+import org.bson.Document;
 import org.springframework.data.mongodb.core.aggregation.BucketOperationSupport.Output;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.ExposedField;
 import org.springframework.util.Assert;
 
-import org.bson.Document;
-
 /**
  * Encapsulates the aggregation framework {@code $facet}-operation. <br />
  * Facet of {@link AggregationOperation}s to be used in an {@link Aggregation}. Processes multiple
@@ -84,7 +83,16 @@ public class FacetOperation implements FieldsExposingAggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$facet", facets.toDocument(context));
+		return new Document(getOperator(), facets.toDocument(context));
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$facet";
 	}
 
 	/* (non-Javadoc)

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/GeoNearOperation.java

@@ -16,14 +16,12 @@
 package org.springframework.data.mongodb.core.aggregation;
 
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 
 import org.bson.Document;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
-import org.springframework.util.NumberUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -95,21 +93,27 @@ public class GeoNearOperation implements AggregationOperation {
 
 		Document command = context.getMappedObject(nearQuery.toDocument());
 
-		if(command.containsKey("query")) {
+		if (command.containsKey("query")) {
 			command.replace("query", context.getMappedObject(command.get("query", Document.class)));
 		}
 
-		if(command.containsKey("collation")) {
-			command.remove("collation");
-		}
-
+		command.remove("collation");
 		command.put("distanceField", distanceField);
 
 		if (StringUtils.hasText(indexKey)) {
 			command.put("key", indexKey);
 		}
 
-		return new Document("$geoNear", command);
+		return new Document(getOperator(), command);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$geoNear";
 	}
 
 	/*
@@ -125,11 +129,11 @@ public class GeoNearOperation implements AggregationOperation {
 		List<Document> stages = new ArrayList<>();
 		stages.add(command);
 
-		if(nearQuery.getSkip() != null && nearQuery.getSkip() > 0){
+		if (nearQuery.getSkip() != null && nearQuery.getSkip() > 0) {
 			stages.add(new Document("$skip", nearQuery.getSkip()));
 		}
 
-		if(limit != null) {
+		if (limit != null) {
 			stages.add(new Document("$limit", limit.longValue()));
 		}
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/GraphLookupOperation.java

@@ -119,7 +119,12 @@ public class GraphLookupOperation implements InheritsFieldsAggregationOperation
 			graphLookup.put("restrictSearchWithMatch", context.getMappedObject(restrictSearchWithMatch.getCriteriaObject()));
 		}
 
-		return new Document("$graphLookup", graphLookup);
+		return new Document(getOperator(), graphLookup);
+	}
+
+	@Override
+	public String getOperator() {
+		return "$graphLookup";
 	}
 
 	/*

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/GroupOperation.java

@@ -429,7 +429,16 @@ public class GroupOperation implements FieldsExposingAggregationOperation {
 			operationObject.putAll(operation.toDocument(context));
 		}
 
-		return new Document("$group", operationObject);
+		return new Document(getOperator(), operationObject);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$group";
 	}
 
 	interface Keyword {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/LimitOperation.java

@@ -49,6 +49,15 @@ public class LimitOperation implements AggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$limit", Long.valueOf(maxElements));
+		return new Document(getOperator(), Long.valueOf(maxElements));
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$limit";
 	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/LiteralOperators.java

@@ -76,6 +76,10 @@ public class LiteralOperators {
 			super(value);
 		}
 
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.AbstractAggregationExpression#getMongoMethod()
+		 */
 		@Override
 		protected String getMongoMethod() {
 			return "$literal";

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/LookupOperation.java

@@ -83,7 +83,16 @@ public class LookupOperation implements FieldsExposingAggregationOperation, Inhe
 		lookupObject.append("foreignField", foreignField.getTarget());
 		lookupObject.append("as", as.getTarget());
 
-		return new Document("$lookup", lookupObject);
+		return new Document(getOperator(), lookupObject);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$lookup";
 	}
 
 	/**
@@ -168,8 +177,7 @@ public class LookupOperation implements FieldsExposingAggregationOperation, Inhe
 
 			Assert.hasText(name, "'As' must not be null or empty!");
 			as = new ExposedField(Fields.field(name), true);
-			return new LookupOperation(from, localField, foreignField,
-					as);
+			return new LookupOperation(from, localField, foreignField, as);
 		}
 
 		@Override

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/MatchOperation.java

@@ -30,7 +30,8 @@ import org.springframework.util.Assert;
  * @author Thomas Darimont
  * @author Oliver Gierke
  * @since 1.3
- * @see <a href="https://docs.mongodb.com/manual/reference/operator/aggregation/match/">MongoDB Aggregation Framework: $match</a>
+ * @see <a href="https://docs.mongodb.com/manual/reference/operator/aggregation/match/">MongoDB Aggregation Framework:
+ *      $match</a>
  */
 public class MatchOperation implements AggregationOperation {
 
@@ -53,6 +54,15 @@ public class MatchOperation implements AggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$match", context.getMappedObject(criteriaDefinition.getCriteriaObject()));
+		return new Document(getOperator(), context.getMappedObject(criteriaDefinition.getCriteriaObject()));
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$match";
 	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/MergeOperation.java

@@ -100,7 +100,7 @@ public class MergeOperation implements FieldsExposingAggregationOperation, Inher
 	public Document toDocument(AggregationOperationContext context) {
 
 		if (isJustCollection()) {
-			return new Document("$merge", into.collection);
+			return new Document(getOperator(), into.collection);
 		}
 
 		Document $merge = new Document();
@@ -122,7 +122,16 @@ public class MergeOperation implements FieldsExposingAggregationOperation, Inher
 			$merge.putAll(whenNotMatched.toDocument(context));
 		}
 
-		return new Document("$merge", $merge);
+		return new Document(getOperator(), $merge);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$merge";
 	}
 
 	/*

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/OutOperation.java

@@ -16,7 +16,6 @@
 package org.springframework.data.mongodb.core.aggregation;
 
 import org.bson.Document;
-
 import org.springframework.data.mongodb.util.BsonUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -211,7 +210,16 @@ public class OutOperation implements AggregationOperation {
 			$out.append("uniqueKey", uniqueKey);
 		}
 
-		return new Document("$out", $out);
+		return new Document(getOperator(), $out);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$out";
 	}
 
 	private boolean requiresMongoDb42Format() {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/ProjectionOperation.java

@@ -261,7 +261,16 @@ public class ProjectionOperation implements FieldsExposingAggregationOperation {
 			fieldObject.putAll(projection.toDocument(context));
 		}
 
-		return new Document("$project", fieldObject);
+		return new Document(getOperator(), fieldObject);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$project";
 	}
 
 	/**
@@ -1548,7 +1557,7 @@ public class ProjectionOperation implements FieldsExposingAggregationOperation {
 				final Field aliasedField = Fields.field(alias, this.field.getName());
 				return new OperationProjection(aliasedField, operation, values.toArray()) {
 
-					/* 
+					/*
 					 * (non-Javadoc)
 					 * @see org.springframework.data.mongodb.core.aggregation.ProjectionOperation.ProjectionOperationBuilder.OperationProjection#getField()
 					 */
@@ -1749,7 +1758,7 @@ public class ProjectionOperation implements FieldsExposingAggregationOperation {
 			this.expression = expression;
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.mongodb.core.aggregation.ProjectionOperation.Projection#toDocument(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)
 		 */
@@ -1877,7 +1886,7 @@ public class ProjectionOperation implements FieldsExposingAggregationOperation {
 			this.projections = projections;
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.mongodb.core.aggregation.ProjectionOperation.Projection#toDocument(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)
 		 */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/RedactOperation.java

@@ -74,7 +74,16 @@ public class RedactOperation implements AggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$redact", condition.toDocument(context));
+		return new Document(getOperator(), condition.toDocument(context));
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$redact";
 	}
 
 	/**

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/ReplaceRootOperation.java

@@ -86,6 +86,15 @@ public class ReplaceRootOperation implements FieldsExposingAggregationOperation
 		return new Document("$replaceRoot", new Document("newRoot", getReplacement().toDocumentExpression(context)));
 	}
 
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$replaceRoot";
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.aggregation.FieldsExposingAggregationOperation#getFields()
 	 */
@@ -166,7 +175,7 @@ public class ReplaceRootOperation implements FieldsExposingAggregationOperation
 	 *
 	 * @author Mark Paluch
 	 */
-	static class ReplaceRootDocumentOperation extends ReplaceRootOperation {
+	public static class ReplaceRootDocumentOperation extends ReplaceRootOperation {
 
 		private final static ReplacementDocument EMPTY = new ReplacementDocument();
 		private final ReplacementDocument current;

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/SampleOperation.java

@@ -48,6 +48,15 @@ public class SampleOperation implements AggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$sample", new Document("size", this.sampleSize));
+		return new Document(getOperator(), new Document("size", this.sampleSize));
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$sample";
 	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/ScriptOperators.java

@@ -0,0 +1,587 @@
+/*
+ * Copyright 2020 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.core.aggregation;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.data.mongodb.core.aggregation.ScriptOperators.Accumulator.AccumulatorBuilder;
+import org.springframework.data.mongodb.core.aggregation.ScriptOperators.Accumulator.AccumulatorInitBuilder;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * Gateway to {@literal $function} and {@literal $accumulator} aggregation operations.
+ * <p />
+ * Using {@link ScriptOperators} as part of the {@link Aggregation} requires MongoDB server to have
+ * <a href="https://docs.mongodb.com/master/core/server-side-javascript/">server-side JavaScript</a> execution
+ * <a href="https://docs.mongodb.com/master/reference/configuration-options/#security.javascriptEnabled">enabled</a>.
+ *
+ * @author Christoph Strobl
+ * @author Mark Paluch
+ * @since 3.1
+ */
+public class ScriptOperators {
+
+	/**
+	 * Create a custom aggregation
+	 * <a href="https://docs.mongodb.com/master/reference/operator/aggregation/function/">$function<a /> in JavaScript.
+	 *
+	 * @param body The function definition. Must not be {@literal null}.
+	 * @return new instance of {@link Function}.
+	 */
+	public static Function function(String body) {
+		return Function.function(body);
+	}
+
+	/**
+	 * Create a custom <a href="https://docs.mongodb.com/master/reference/operator/aggregation/accumulator/">$accumulator
+	 * operator</a> in Javascript.
+	 *
+	 * @return new instance of {@link AccumulatorInitBuilder}.
+	 */
+	public static AccumulatorInitBuilder accumulatorBuilder() {
+		return new AccumulatorBuilder();
+	}
+
+	/**
+	 * {@link Function} defines a custom aggregation
+	 * <a href="https://docs.mongodb.com/master/reference/operator/aggregation/function/">$function</a> in JavaScript.
+	 * <p />
+	 * <code class="java">
+	 * {
+	 *   $function: {
+	 *     body: ...,
+	 *     args: ...,
+	 *     lang: "js"
+	 *   }
+	 * }
+	 * </code>
+	 * <p />
+	 * {@link Function} cannot be used as part of {@link org.springframework.data.mongodb.core.schema.MongoJsonSchema
+	 * schema} validation query expression. <br />
+	 * <b>NOTE:</b> <a href="https://docs.mongodb.com/master/core/server-side-javascript/">Server-Side JavaScript</a>
+	 * execution must be
+	 * <a href="https://docs.mongodb.com/master/reference/configuration-options/#security.javascriptEnabled">enabled</a>
+	 *
+	 * @see <a href="https://docs.mongodb.com/master/reference/operator/aggregation/function/">MongoDB Documentation:
+	 *      $function</a>
+	 */
+	public static class Function extends AbstractAggregationExpression {
+
+		private Function(Map<String, Object> values) {
+			super(values);
+		}
+
+		/**
+		 * Create a new {@link Function} with the given function definition.
+		 *
+		 * @param body must not be {@literal null}.
+		 * @return new instance of {@link Function}.
+		 */
+		public static Function function(String body) {
+
+			Assert.notNull(body, "Function body must not be null!");
+
+			Map<String, Object> function = new LinkedHashMap<>(2);
+			function.put(Fields.BODY.toString(), body);
+			function.put(Fields.ARGS.toString(), Collections.emptyList());
+			function.put(Fields.LANG.toString(), "js");
+
+			return new Function(function);
+		}
+
+		/**
+		 * Set the arguments passed to the function body.
+		 *
+		 * @param args the arguments passed to the function body. Leave empty if the function does not take any arguments.
+		 * @return new instance of {@link Function}.
+		 */
+		public Function args(Object... args) {
+			return args(Arrays.asList(args));
+		}
+
+		/**
+		 * Set the arguments passed to the function body.
+		 *
+		 * @param args the arguments passed to the function body. Leave empty if the function does not take any arguments.
+		 * @return new instance of {@link Function}.
+		 */
+		public Function args(List<Object> args) {
+
+			Assert.notNull(args, "Args must not be null! Use an empty list instead.");
+
+			return new Function(appendAt(1, Fields.ARGS.toString(), args));
+		}
+
+		/**
+		 * The language used in the body.
+		 *
+		 * @param lang must not be {@literal null} nor empty.
+		 * @return new instance of {@link Function}.
+		 */
+		public Function lang(String lang) {
+
+			Assert.hasText(lang, "Lang must not be null nor empty! The default would be 'js'.");
+
+			return new Function(appendAt(2, Fields.LANG.toString(), lang));
+		}
+
+		@Nullable
+		List<Object> getArgs() {
+			return get(Fields.ARGS.toString());
+		}
+
+		String getBody() {
+			return get(Fields.BODY.toString());
+		}
+
+		String getLang() {
+			return get(Fields.LANG.toString());
+		}
+
+		@Override
+		protected String getMongoMethod() {
+			return "$function";
+		}
+
+		enum Fields {
+
+			BODY, ARGS, LANG;
+
+			@Override
+			public String toString() {
+				return name().toLowerCase();
+			}
+		}
+	}
+
+	/**
+	 * {@link Accumulator} defines a custom aggregation
+	 * <a href="https://docs.mongodb.com/master/reference/operator/aggregation/accumulator/">$accumulator operator</a>,
+	 * one that maintains its state (e.g. totals, maximums, minimums, and related data) as documents progress through the
+	 * pipeline, in JavaScript.
+	 * <p />
+	 * <code class="java">
+	 * {
+	 *   $accumulator: {
+	 *     init: ...,
+	 *     intArgs: ...,
+	 *     accumulate: ...,
+	 *     accumulateArgs: ...,
+	 *     merge: ...,
+	 *     finalize: ...,
+	 *     lang: "js"
+	 *   }
+	 * }
+	 * </code>
+	 * <p />
+	 * {@link Accumulator} can be used as part of {@link GroupOperation $group}, {@link BucketOperation $bucket} and
+	 * {@link BucketAutoOperation $bucketAuto} pipeline stages. <br />
+	 * <b>NOTE:</b> <a href="https://docs.mongodb.com/master/core/server-side-javascript/">Server-Side JavaScript</a>
+	 * execution must be
+	 * <a href="https://docs.mongodb.com/master/reference/configuration-options/#security.javascriptEnabled">enabled</a>
+	 *
+	 * @see <a href="https://docs.mongodb.com/master/reference/operator/aggregation/accumulator/">MongoDB Documentation:
+	 *      $accumulator</a>
+	 */
+	public static class Accumulator extends AbstractAggregationExpression {
+
+		private Accumulator(Map<String, Object> value) {
+			super(value);
+		}
+
+		@Override
+		protected String getMongoMethod() {
+			return "$accumulator";
+		}
+
+		enum Fields {
+
+			ACCUMULATE("accumulate"), //
+			ACCUMULATE_ARGS("accumulateArgs"), //
+			FINALIZE("finalize"), //
+			INIT("init"), //
+			INIT_ARGS("initArgs"), //
+			LANG("lang"), //
+			MERGE("merge"); //
+
+			private String field;
+
+			Fields(String field) {
+				this.field = field;
+			}
+
+			@Override
+			public String toString() {
+				return field;
+			}
+		}
+
+		public interface AccumulatorInitBuilder {
+
+			/**
+			 * Define the {@code init} {@link Function} for the {@link Accumulator accumulators} initial state. The function
+			 * receives its arguments from the {@link Function#args(Object...) initArgs} array expression.
+			 * <p />
+			 * <code class="java">
+			 * function(initArg1, initArg2, ...) {
+			 *   ...
+			 *   return initialState
+			 * }
+			 * </code>
+			 *
+			 * @param function must not be {@literal null}.
+			 * @return this.
+			 */
+			default AccumulatorAccumulateBuilder init(Function function) {
+				return init(function.getBody()).initArgs(function.getArgs());
+			}
+
+			/**
+			 * Define the {@code init} function for the {@link Accumulator accumulators} initial state. The function receives
+			 * its arguments from the {@link AccumulatorInitArgsBuilder#initArgs(Object...)} array expression.
+			 * <p />
+			 * <code class="java">
+			 * function(initArg1, initArg2, ...) {
+			 *   ...
+			 *   return initialState
+			 * }
+			 * </code>
+			 *
+			 * @param function must not be {@literal null}.
+			 * @return this.
+			 */
+			AccumulatorInitArgsBuilder init(String function);
+
+			/**
+			 * The language used in the {@code $accumulator} code.
+			 *
+			 * @param lang must not be {@literal null}. Default is {@literal js}.
+			 * @return this.
+			 */
+			AccumulatorInitBuilder lang(String lang);
+		}
+
+		public interface AccumulatorInitArgsBuilder extends AccumulatorAccumulateBuilder {
+
+			/**
+			 * Define the optional {@code initArgs} for the {@link AccumulatorInitBuilder#init(String)} function.
+			 *
+			 * @param args must not be {@literal null}.
+			 * @return this.
+			 */
+			default AccumulatorAccumulateBuilder initArgs(Object... args) {
+				return initArgs(Arrays.asList(args));
+			}
+
+			/**
+			 * Define the optional {@code initArgs} for the {@link AccumulatorInitBuilder#init(String)} function.
+			 *
+			 * @param args must not be {@literal null}.
+			 * @return this.
+			 */
+			AccumulatorAccumulateBuilder initArgs(List<Object> args);
+		}
+
+		public interface AccumulatorAccumulateBuilder {
+
+			/**
+			 * Set the {@code accumulate} {@link Function} that updates the state for each document. The functions first
+			 * argument is the current {@code state}, additional arguments can be defined via {@link Function#args(Object...)
+			 * accumulateArgs}.
+			 * <p />
+			 * <code class="java">
+			 * function(state, accumArg1, accumArg2, ...) {
+			 *   ...
+			 *   return newState
+			 * }
+			 * </code>
+			 *
+			 * @param function must not be {@literal null}.
+			 * @return this.
+			 */
+			default AccumulatorMergeBuilder accumulate(Function function) {
+				return accumulate(function.getBody()).accumulateArgs(function.getArgs());
+			}
+
+			/**
+			 * Set the {@code accumulate} function that updates the state for each document. The functions first argument is
+			 * the current {@code state}, additional arguments can be defined via
+			 * {@link AccumulatorAccumulateArgsBuilder#accumulateArgs(Object...)}.
+			 * <p />
+			 * <code class="java">
+			 * function(state, accumArg1, accumArg2, ...) {
+			 *   ...
+			 *   return newState
+			 * }
+			 * </code>
+			 *
+			 * @param function must not be {@literal null}.
+			 * @return this.
+			 */
+			AccumulatorAccumulateArgsBuilder accumulate(String function);
+		}
+
+		public interface AccumulatorAccumulateArgsBuilder extends AccumulatorMergeBuilder {
+
+			/**
+			 * Define additional {@code accumulateArgs} for the {@link AccumulatorAccumulateBuilder#accumulate(String)}
+			 * function.
+			 *
+			 * @param args must not be {@literal null}.
+			 * @return this.
+			 */
+			default AccumulatorMergeBuilder accumulateArgs(Object... args) {
+				return accumulateArgs(Arrays.asList(args));
+			}
+
+			/**
+			 * Define additional {@code accumulateArgs} for the {@link AccumulatorAccumulateBuilder#accumulate(String)}
+			 * function.
+			 *
+			 * @param args must not be {@literal null}.
+			 * @return this.
+			 */
+			AccumulatorMergeBuilder accumulateArgs(List<Object> args);
+		}
+
+		public interface AccumulatorMergeBuilder {
+
+			/**
+			 * Set the {@code merge} function used to merge two internal states. <br />
+			 * This might be required because the operation is run on a sharded cluster or when the operator exceeds its
+			 * memory limit.
+			 * <p />
+			 * <code class="java">
+			 * function(state1, state2) {
+			 *   ...
+			 *   return newState
+			 * }
+			 * </code>
+			 *
+			 * @param function must not be {@literal null}.
+			 * @return this.
+			 */
+			AccumulatorFinalizeBuilder merge(String function);
+		}
+
+		public interface AccumulatorFinalizeBuilder {
+
+			/**
+			 * Set the {@code finalize} function used to update the result of the accumulation when all documents have been
+			 * processed.
+			 * <p />
+			 * <code class="java">
+			 * function(state) {
+			 *   ...
+			 *   return finalState
+			 * }
+			 * </code>
+			 *
+			 * @param function must not be {@literal null}.
+			 * @return new instance of {@link Accumulator}.
+			 */
+			Accumulator finalize(String function);
+
+			/**
+			 * Build the {@link Accumulator} object without specifying a {@link #finalize(String) finalize function}.
+			 *
+			 * @return new instance of {@link Accumulator}.
+			 */
+			Accumulator build();
+		}
+
+		static class AccumulatorBuilder
+				implements AccumulatorInitBuilder, AccumulatorInitArgsBuilder, AccumulatorAccumulateBuilder,
+				AccumulatorAccumulateArgsBuilder, AccumulatorMergeBuilder, AccumulatorFinalizeBuilder {
+
+			private List<Object> initArgs;
+			private String initFunction;
+			private List<Object> accumulateArgs;
+			private String accumulateFunction;
+			private String mergeFunction;
+			private String finalizeFunction;
+			private String lang = "js";
+
+			/**
+			 * Define the {@code init} function for the {@link Accumulator accumulators} initial state. The function receives
+			 * its arguments from the {@link #initArgs(Object...)} array expression.
+			 * <p />
+			 * <code class="java">
+			 * function(initArg1, initArg2, ...) {
+			 *   ...
+			 *   return initialState
+			 * }
+			 * </code>
+			 *
+			 * @param function must not be {@literal null}.
+			 * @return this.
+			 */
+			@Override
+			public AccumulatorBuilder init(String function) {
+
+				this.initFunction = function;
+				return this;
+			}
+
+			/**
+			 * Define the optional {@code initArgs} for the {@link #init(String)} function.
+			 *
+			 * @param function must not be {@literal null}.
+			 * @return this.
+			 */
+			@Override
+			public AccumulatorBuilder initArgs(List<Object> args) {
+
+				Assert.notNull(args, "Args must not be null");
+
+				this.initArgs = new ArrayList<>(args);
+				return this;
+			}
+
+			/**
+			 * Set the {@code accumulate} function that updates the state for each document. The functions first argument is
+			 * the current {@code state}, additional arguments can be defined via {@link #accumulateArgs(Object...)}.
+			 * <p />
+			 * <code class="java">
+			 * function(state, accumArg1, accumArg2, ...) {
+			 *   ...
+			 *   return newState
+			 * }
+			 * </code>
+			 *
+			 * @param function must not be {@literal null}.
+			 * @return this.
+			 */
+			@Override
+			public AccumulatorBuilder accumulate(String function) {
+
+				Assert.notNull(function, "Accumulate function must not be null");
+
+				this.accumulateFunction = function;
+				return this;
+			}
+
+			/**
+			 * Define additional {@code accumulateArgs} for the {@link #accumulate(String)} function.
+			 *
+			 * @param args must not be {@literal null}.
+			 * @return this.
+			 */
+			@Override
+			public AccumulatorBuilder accumulateArgs(List<Object> args) {
+
+				Assert.notNull(args, "Args must not be null");
+
+				this.accumulateArgs = new ArrayList<>(args);
+				return this;
+			}
+
+			/**
+			 * Set the {@code merge} function used to merge two internal states. <br />
+			 * This might be required because the operation is run on a sharded cluster or when the operator exceeds its
+			 * memory limit.
+			 * <p />
+			 * <code class="java">
+			 * function(state1, state2) {
+			 *   ...
+			 *   return newState
+			 * }
+			 * </code>
+			 *
+			 * @param function must not be {@literal null}.
+			 * @return this.
+			 */
+			@Override
+			public AccumulatorBuilder merge(String function) {
+
+				Assert.notNull(function, "Merge function must not be null");
+
+				this.mergeFunction = function;
+				return this;
+			}
+
+			/**
+			 * The language used in the {@code $accumulator} code.
+			 *
+			 * @param lang must not be {@literal null}. Default is {@literal js}.
+			 * @return this.
+			 */
+			public AccumulatorBuilder lang(String lang) {
+
+				Assert.hasText(lang, "Lang must not be null nor empty! The default would be 'js'.");
+
+				this.lang = lang;
+				return this;
+			}
+
+			/**
+			 * Set the {@code finalize} function used to update the result of the accumulation when all documents have been
+			 * processed.
+			 * <p />
+			 * <code class="java">
+			 * function(state) {
+			 *   ...
+			 *   return finalState
+			 * }
+			 * </code>
+			 *
+			 * @param function must not be {@literal null}.
+			 * @return new instance of {@link Accumulator}.
+			 */
+			@Override
+			public Accumulator finalize(String function) {
+
+				Assert.notNull(function, "Finalize function must not be null");
+
+				this.finalizeFunction = function;
+
+				Map<String, Object> args = createArgumentMap();
+				args.put(Fields.FINALIZE.toString(), finalizeFunction);
+
+				return new Accumulator(args);
+			}
+
+			@Override
+			public Accumulator build() {
+				return new Accumulator(createArgumentMap());
+			}
+
+			private Map<String, Object> createArgumentMap() {
+
+				Map<String, Object> args = new LinkedHashMap<>();
+				args.put(Fields.INIT.toString(), initFunction);
+				if (!CollectionUtils.isEmpty(initArgs)) {
+					args.put(Fields.INIT_ARGS.toString(), initArgs);
+				}
+				args.put(Fields.ACCUMULATE.toString(), accumulateFunction);
+				if (!CollectionUtils.isEmpty(accumulateArgs)) {
+					args.put(Fields.ACCUMULATE_ARGS.toString(), accumulateArgs);
+				}
+				args.put(Fields.MERGE.toString(), mergeFunction);
+				args.put(Fields.LANG.toString(), lang);
+
+				return args;
+			}
+		}
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/SetOperation.java

@@ -99,6 +99,10 @@ public class SetOperation extends DocumentEnhancingOperation {
 		return new FieldAppender(getValueMap());
 	}
 
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.DocumentEnhancingOperation#mongoOperator()
+	 */
 	@Override
 	protected String mongoOperator() {
 		return "$set";

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/SkipOperation.java

@@ -28,7 +28,8 @@ import org.springframework.util.Assert;
  * @author Oliver Gierke
  * @author Christoph Strobl
  * @since 1.3
- * @see <a href="https://docs.mongodb.com/manual/reference/operator/aggregation/skip/">MongoDB Aggregation Framework: $skip</a>
+ * @see <a href="https://docs.mongodb.com/manual/reference/operator/aggregation/skip/">MongoDB Aggregation Framework:
+ *      $skip</a>
  */
 public class SkipOperation implements AggregationOperation {
 
@@ -51,6 +52,15 @@ public class SkipOperation implements AggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$skip", skipCount);
+		return new Document(getOperator(), skipCount);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$skip";
 	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/SortByCountOperation.java

@@ -67,14 +67,23 @@ public class SortByCountOperation implements AggregationOperation {
 		this.groupByField = null;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#toDocument(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
 
-		return new Document("$sortByCount", groupByExpression == null ? context.getReference(groupByField).toString()
+		return new Document(getOperator(), groupByExpression == null ? context.getReference(groupByField).toString()
 				: groupByExpression.toDocument(context));
 	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$sortByCount";
+	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/SortOperation.java

@@ -33,7 +33,8 @@ import org.springframework.util.Assert;
  * @author Christoph Strobl
  * @author Mark Paluch
  * @since 1.3
- * @see <a href="https://docs.mongodb.com/manual/reference/operator/aggregation/sort/">MongoDB Aggregation Framework: $sort</a>
+ * @see <a href="https://docs.mongodb.com/manual/reference/operator/aggregation/sort/">MongoDB Aggregation Framework:
+ *      $sort</a>
  */
 public class SortOperation implements AggregationOperation {
 
@@ -74,6 +75,15 @@ public class SortOperation implements AggregationOperation {
 			object.put(reference.getRaw(), order.isAscending() ? 1 : -1);
 		}
 
-		return new Document("$sort", object);
+		return new Document(getOperator(), object);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$sort";
 	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/TypeBasedAggregationOperationContext.java

@@ -133,6 +133,19 @@ public class TypeBasedAggregationOperationContext implements AggregationOperatio
 	 */
 	@Override
 	public AggregationOperationContext continueOnMissingFieldReference() {
+		return continueOnMissingFieldReference(type);
+	}
+
+	/**
+	 * This toggle allows the {@link AggregationOperationContext context} to use any given field name without checking for
+	 * its existence. Typically the {@link AggregationOperationContext} fails when referencing unknown fields, those that
+	 * are not present in one of the previous stages or the input source, throughout the pipeline.
+	 *
+	 * @param type The domain type to map fields to.
+	 * @return a more relaxed {@link AggregationOperationContext}.
+	 * @since 3.1
+	 */
+	public AggregationOperationContext continueOnMissingFieldReference(Class<?> type) {
 		return new RelaxedTypeBasedAggregationOperationContext(type, mappingContext, mapper);
 	}
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/TypedAggregation.java

@@ -81,6 +81,6 @@ public class TypedAggregation<I> extends Aggregation {
 	public TypedAggregation<I> withOptions(AggregationOptions options) {
 
 		Assert.notNull(options, "AggregationOptions must not be null.");
-		return new TypedAggregation<I>(inputType, operations, options);
+		return new TypedAggregation<I>(inputType, pipeline.getOperations(), options);
 	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/UnionWithOperation.java

@@ -0,0 +1,168 @@
+/*
+ * Copyright 2020 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.core.aggregation;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.bson.Document;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * The <a href="https://docs.mongodb.com/master/reference/operator/aggregation/unionWith/">$unionWith</a> aggregation
+ * stage (available since MongoDB 4.4) performs a union of two collections by combining pipeline results, potentially
+ * containing duplicates, into a single result set that is handed over to the next stage. <br />
+ * In order to remove duplicates it is possible to append a {@link GroupOperation} right after
+ * {@link UnionWithOperation}.
+ * <p />
+ * If the {@link UnionWithOperation} uses a
+ * <a href="https://docs.mongodb.com/master/reference/operator/aggregation/unionWith/#unionwith-pipeline">pipeline</a>
+ * to process documents, field names within the pipeline will be treated as is. In order to map domain type property
+ * names to actual field names (considering potential {@link org.springframework.data.mongodb.core.mapping.Field}
+ * annotations) make sure the enclosing aggregation is a {@link TypedAggregation} and provide the target type for the
+ * {@code $unionWith} stage via {@link #mapFieldsTo(Class)}.
+ *
+ * @author Christoph Strobl
+ * @see <a href="https://docs.mongodb.com/master/reference/operator/aggregation/unionWith/">Aggregation Pipeline Stage:
+ *      $unionWith</a>
+ * @since 3.1
+ */
+public class UnionWithOperation implements AggregationOperation {
+
+	private final String collection;
+
+	private final @Nullable AggregationPipeline pipeline;
+
+	private final @Nullable Class<?> domainType;
+
+	public UnionWithOperation(String collection, @Nullable AggregationPipeline pipeline, @Nullable Class<?> domainType) {
+
+		Assert.notNull(collection, "Collection must not be null!");
+
+		this.collection = collection;
+		this.pipeline = pipeline;
+		this.domainType = domainType;
+	}
+
+	/**
+	 * Set the name of the collection from which pipeline results should be included in the result set.<br />
+	 * The collection name is used to set the {@code coll} parameter of {@code $unionWith}.
+	 *
+	 * @param collection the MongoDB collection name. Must not be {@literal null}.
+	 * @return new instance of {@link UnionWithOperation}.
+	 * @throws IllegalArgumentException if the required argument is {@literal null}.
+	 */
+	public static UnionWithOperation unionWith(String collection) {
+		return new UnionWithOperation(collection, null, null);
+	}
+
+	/**
+	 * Set the {@link AggregationPipeline} to apply to the specified collection. The pipeline corresponds to the optional
+	 * {@code pipeline} field of the {@code $unionWith} aggregation stage and is used to compute the documents going into
+	 * the result set.
+	 *
+	 * @param pipeline the {@link AggregationPipeline} that computes the documents. Must not be {@literal null}.
+	 * @return new instance of {@link UnionWithOperation}.
+	 * @throws IllegalArgumentException if the required argument is {@literal null}.
+	 */
+	public UnionWithOperation pipeline(AggregationPipeline pipeline) {
+		return new UnionWithOperation(collection, pipeline, domainType);
+	}
+
+	/**
+	 * Set the aggregation pipeline stages to apply to the specified collection. The pipeline corresponds to the optional
+	 * {@code pipeline} field of the {@code $unionWith} aggregation stage and is used to compute the documents going into
+	 * the result set.
+	 *
+	 * @param aggregationStages the aggregation pipeline stages that compute the documents. Must not be {@literal null}.
+	 * @return new instance of {@link UnionWithOperation}.
+	 * @throws IllegalArgumentException if the required argument is {@literal null}.
+	 */
+	public UnionWithOperation pipeline(List<AggregationOperation> aggregationStages) {
+		return new UnionWithOperation(collection, new AggregationPipeline(aggregationStages), domainType);
+	}
+
+	/**
+	 * Set the aggregation pipeline stages to apply to the specified collection. The pipeline corresponds to the optional
+	 * {@code pipeline} field of the {@code $unionWith} aggregation stage and is used to compute the documents going into
+	 * the result set.
+	 *
+	 * @param aggregationStages the aggregation pipeline stages that compute the documents. Must not be {@literal null}.
+	 * @return new instance of {@link UnionWithOperation}.
+	 * @throws IllegalArgumentException if the required argument is {@literal null}.
+	 */
+	public UnionWithOperation pipeline(AggregationOperation... aggregationStages) {
+		return new UnionWithOperation(collection, new AggregationPipeline(Arrays.asList(aggregationStages)), domainType);
+	}
+
+	/**
+	 * Set domain type used for field name mapping of property references used by the {@link AggregationPipeline}.
+	 * Remember to also use a {@link TypedAggregation} in the outer pipeline.<br />
+	 * If not set, field names used within {@link AggregationOperation pipeline operations} are taken as is.
+	 *
+	 * @param domainType the domain type to map field names used in pipeline operations to. Must not be {@literal null}.
+	 * @return new instance of {@link UnionWithOperation}.
+	 * @throws IllegalArgumentException if the required argument is {@literal null}.
+	 */
+	public UnionWithOperation mapFieldsTo(Class<?> domainType) {
+
+		Assert.notNull(domainType, "DomainType must not be null!");
+		return new UnionWithOperation(collection, pipeline, domainType);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#toDocument(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)
+	 */
+	@Override
+	public Document toDocument(AggregationOperationContext context) {
+
+		Document $unionWith = new Document("coll", collection);
+		if (pipeline == null || pipeline.isEmpty()) {
+			return new Document(getOperator(), $unionWith);
+		}
+
+		$unionWith.append("pipeline", pipeline.toDocuments(computeContext(context)));
+		return new Document(getOperator(), $unionWith);
+	}
+
+	private AggregationOperationContext computeContext(AggregationOperationContext source) {
+
+		if (domainType == null) {
+			return Aggregation.DEFAULT_CONTEXT;
+		}
+
+		if (source instanceof TypeBasedAggregationOperationContext) {
+			return ((TypeBasedAggregationOperationContext) source).continueOnMissingFieldReference(domainType);
+		}
+
+		if (source instanceof ExposedFieldsAggregationOperationContext) {
+			return computeContext(((ExposedFieldsAggregationOperationContext) source).getRootContext());
+		}
+
+		return source;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$unionWith";
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/UnsetOperation.java

@@ -40,7 +40,7 @@ public class UnsetOperation implements InheritsFieldsAggregationOperation {
 
 	/**
 	 * Create new instance of {@link UnsetOperation}.
-	 * 
+	 *
 	 * @param fields must not be {@literal null}.
 	 */
 	public UnsetOperation(Collection<Object> fields) {
@@ -117,13 +117,22 @@ public class UnsetOperation implements InheritsFieldsAggregationOperation {
 	public Document toDocument(AggregationOperationContext context) {
 
 		if (fields.size() == 1) {
-			return new Document("$unset", computeFieldName(fields.iterator().next(), context));
+			return new Document(getOperator(), computeFieldName(fields.iterator().next(), context));
 		}
 
-		return new Document("$unset",
+		return new Document(getOperator(),
 				fields.stream().map(it -> computeFieldName(it, context)).collect(Collectors.toList()));
 	}
 
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$unset";
+	}
+
 	private Object computeFieldName(Object field, AggregationOperationContext context) {
 
 		if (field instanceof Field) {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/UnwindOperation.java

@@ -31,7 +31,8 @@ import org.springframework.util.Assert;
  * @author Mark Paluch
  * @author Christoph Strobl
  * @since 1.3
- * @see <a href="https://docs.mongodb.com/manual/reference/operator/aggregation/unwind/">MongoDB Aggregation Framework: $unwind</a>
+ * @see <a href="https://docs.mongodb.com/manual/reference/operator/aggregation/unwind/">MongoDB Aggregation Framework:
+ *      $unwind</a>
  */
 public class UnwindOperation
 		implements AggregationOperation, FieldsExposingAggregationOperation.InheritsFieldsAggregationOperation {
@@ -94,7 +95,7 @@ public class UnwindOperation
 		String path = context.getReference(field).toString();
 
 		if (!preserveNullAndEmptyArrays && arrayIndex == null) {
-			return new Document("$unwind", path);
+			return new Document(getOperator(), path);
 		}
 
 		Document unwindArgs = new Document();
@@ -104,7 +105,16 @@ public class UnwindOperation
 		}
 		unwindArgs.put("preserveNullAndEmptyArrays", preserveNullAndEmptyArrays);
 
-		return new Document("$unwind", unwindArgs);
+		return new Document(getOperator(), unwindArgs);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
+	 */
+	@Override
+	public String getOperator() {
+		return "$unwind";
 	}
 
 	/*

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/DefaultDbRefResolver.java

@@ -45,6 +45,7 @@ import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.mongodb.ClientSessionException;
 import org.springframework.data.mongodb.LazyLoadingException;
 import org.springframework.data.mongodb.MongoDatabaseFactory;
+import org.springframework.data.mongodb.MongoDatabaseUtils;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
 import org.springframework.lang.Nullable;
 import org.springframework.objenesis.ObjenesisStd;
@@ -114,14 +115,16 @@ public class DefaultDbRefResolver implements DbRefResolver {
 	@Override
 	public Document fetch(DBRef dbRef) {
 
+		MongoCollection<Document> mongoCollection = getCollection(dbRef);
+
 		if (LOGGER.isTraceEnabled()) {
 			LOGGER.trace("Fetching DBRef '{}' from {}.{}.", dbRef.getId(),
-					StringUtils.hasText(dbRef.getDatabaseName()) ? dbRef.getDatabaseName() : mongoDbFactory.getMongoDatabase().getName(),
+					StringUtils.hasText(dbRef.getDatabaseName()) ? dbRef.getDatabaseName()
+							: mongoCollection.getNamespace().getDatabaseName(),
 					dbRef.getCollectionName());
 		}
 
-		StringUtils.hasText(dbRef.getDatabaseName());
-		return getCollection(dbRef).find(Filters.eq("_id", dbRef.getId())).first();
+		return mongoCollection.find(Filters.eq("_id", dbRef.getId())).first();
 	}
 
 	/*
@@ -152,15 +155,16 @@ public class DefaultDbRefResolver implements DbRefResolver {
 		}
 
 		DBRef databaseSource = refs.iterator().next();
+		MongoCollection<Document> mongoCollection = getCollection(databaseSource);
 
 		if (LOGGER.isTraceEnabled()) {
 			LOGGER.trace("Bulk fetching DBRefs {} from {}.{}.", ids,
 					StringUtils.hasText(databaseSource.getDatabaseName()) ? databaseSource.getDatabaseName()
-							: mongoDbFactory.getMongoDatabase().getName(),
+							: mongoCollection.getNamespace().getDatabaseName(),
 					databaseSource.getCollectionName());
 		}
 
-		List<Document> result = getCollection(databaseSource) //
+		List<Document> result = mongoCollection //
 				.find(new Document("_id", new Document("$in", ids))) //
 				.into(new ArrayList<>());
 
@@ -497,7 +501,7 @@ public class DefaultDbRefResolver implements DbRefResolver {
 	 */
 	protected MongoCollection<Document> getCollection(DBRef dbref) {
 
-		return (StringUtils.hasText(dbref.getDatabaseName()) ? mongoDbFactory.getMongoDatabase(dbref.getDatabaseName())
-				: mongoDbFactory.getMongoDatabase()).getCollection(dbref.getCollectionName(), Document.class);
+		return MongoDatabaseUtils.getDatabase(dbref.getDatabaseName(), mongoDbFactory)
+				.getCollection(dbref.getCollectionName(), Document.class);
 	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/MongoConverter.java

@@ -127,8 +127,8 @@ public interface MongoConverter
 	@Nullable
 	default Object convertId(@Nullable Object id, Class<?> targetType) {
 
-		if (id == null) {
-			return null;
+		if (id == null || ClassUtils.isAssignableValue(targetType, id)) {
+			return id;
 		}
 
 		if (ClassUtils.isAssignable(ObjectId.class, targetType)) {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/index/CompoundIndex.java

@@ -46,6 +46,7 @@ import java.lang.annotation.Target;
  * @author Philipp Schneider
  * @author Johno Crawford
  * @author Christoph Strobl
+ * @author Dave Perryman
  */
 @Target({ ElementType.TYPE })
 @Documented
@@ -95,7 +96,8 @@ public @interface CompoundIndex {
 	boolean unique() default false;
 
 	/**
-	 * If set to true index will skip over any document that is missing the indexed field.
+	 * If set to true index will skip over any document that is missing the indexed field. <br />
+	 * Must not be used with {@link #partialFilter()}.
 	 *
 	 * @return {@literal false} by default.
 	 * @see <a href=
@@ -170,4 +172,14 @@ public @interface CompoundIndex {
 	 */
 	boolean background() default false;
 
+	/**
+	 * Only index the documents in a collection that meet a specified {@link IndexFilter filter expression}. <br />
+	 * Must not be used with {@link #sparse() sparse = true}.
+	 *
+	 * @return empty by default.
+	 * @see <a href=
+	 *      "https://docs.mongodb.com/manual/core/index-partial/">https://docs.mongodb.com/manual/core/index-partial/</a>
+	 * @since 3.1
+	 */
+	String partialFilter() default "";
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/index/Indexed.java

@@ -53,7 +53,8 @@ public @interface Indexed {
 	IndexDirection direction() default IndexDirection.ASCENDING;
 
 	/**
-	 * If set to true index will skip over any document that is missing the indexed field.
+	 * If set to true index will skip over any document that is missing the indexed field. <br />
+	 * Must not be used with {@link #partialFilter()}.
 	 *
 	 * @return {@literal false} by default.
 	 * @see <a href=
@@ -170,4 +171,15 @@ public @interface Indexed {
 	 * @since 2.2
 	 */
 	String expireAfter() default "";
+
+	/**
+	 * Only index the documents in a collection that meet a specified {@link IndexFilter filter expression}. <br />
+	 * Must not be used with {@link #sparse() sparse = true}.
+	 * 
+	 * @return empty by default.
+	 * @see <a href=
+	 *      "https://docs.mongodb.com/manual/core/index-partial/">https://docs.mongodb.com/manual/core/index-partial/</a>
+	 * @since 3.1
+	 */
+	String partialFilter() default "";
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/index/MongoPersistentEntityIndexResolver.java

@@ -46,6 +46,7 @@ import org.springframework.data.mongodb.core.mapping.BasicMongoPersistentEntity;
 import org.springframework.data.mongodb.core.mapping.Document;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
+import org.springframework.data.mongodb.util.BsonUtils;
 import org.springframework.data.spel.EvaluationContextProvider;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.expression.EvaluationContext;
@@ -69,6 +70,7 @@ import org.springframework.util.StringUtils;
  * @author Thomas Darimont
  * @author Martin Macko
  * @author Mark Paluch
+ * @author Dave Perryman
  * @since 1.5
  */
 public class MongoPersistentEntityIndexResolver implements IndexResolver {
@@ -380,6 +382,10 @@ public class MongoPersistentEntityIndexResolver implements IndexResolver {
 			indexDefinition.background();
 		}
 
+		if (StringUtils.hasText(index.partialFilter())) {
+			indexDefinition.partial(evaluatePartialFilter(index.partialFilter(), entity));
+		}
+
 		return new IndexDefinitionHolder(dotPath, indexDefinition, collection);
 	}
 
@@ -469,9 +475,25 @@ public class MongoPersistentEntityIndexResolver implements IndexResolver {
 			}
 		}
 
+		if (StringUtils.hasText(index.partialFilter())) {
+			indexDefinition.partial(evaluatePartialFilter(index.partialFilter(), persistentProperty.getOwner()));
+		}
+
 		return new IndexDefinitionHolder(dotPath, indexDefinition, collection);
 	}
 
+	private PartialIndexFilter evaluatePartialFilter(String filterExpression, PersistentEntity<?,?> entity) {
+
+		Object result = evaluate(filterExpression, getEvaluationContextForProperty(entity));
+
+		if (result instanceof org.bson.Document) {
+			return PartialIndexFilter.of((org.bson.Document) result);
+		}
+
+		return PartialIndexFilter.of(BsonUtils.parse(filterExpression, null));
+	}
+
+
 	/**
 	 * Creates {@link HashedIndex} wrapped in {@link IndexDefinitionHolder} out of {@link HashIndexed} for a given
 	 * {@link MongoPersistentProperty}.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/mapping/MongoSimpleTypes.java

@@ -48,6 +48,7 @@ import com.mongodb.client.model.geojson.Polygon;
  *
  * @author Oliver Gierke
  * @author Christoph Strobl
+ * @author Mark Paluch
  */
 public abstract class MongoSimpleTypes {
 
@@ -109,6 +110,10 @@ public abstract class MongoSimpleTypes {
 		@Override
 		public boolean isSimpleType(Class<?> type) {
 
+			if (type.isEnum()) {
+				return true;
+			}
+
 			if (type.getName().startsWith("java.time")) {
 				return false;
 			}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/mapping/event/ReactiveAuditingEntityCallback.java

@@ -15,13 +15,12 @@
  */
 package org.springframework.data.mongodb.core.mapping.event;
 
-import reactor.core.publisher.Mono;
-
 import org.reactivestreams.Publisher;
+
 import org.springframework.beans.factory.ObjectFactory;
 import org.springframework.core.Ordered;
 import org.springframework.data.auditing.AuditingHandler;
-import org.springframework.data.auditing.IsNewAwareAuditingHandler;
+import org.springframework.data.auditing.ReactiveIsNewAwareAuditingHandler;
 import org.springframework.data.mapping.callback.EntityCallback;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.util.Assert;
@@ -34,7 +33,7 @@ import org.springframework.util.Assert;
  */
 public class ReactiveAuditingEntityCallback implements ReactiveBeforeConvertCallback<Object>, Ordered {
 
-	private final ObjectFactory<IsNewAwareAuditingHandler> auditingHandlerFactory;
+	private final ObjectFactory<ReactiveIsNewAwareAuditingHandler> auditingHandlerFactory;
 
 	/**
 	 * Creates a new {@link ReactiveAuditingEntityCallback} using the given {@link MappingContext} and
@@ -42,19 +41,19 @@ public class ReactiveAuditingEntityCallback implements ReactiveBeforeConvertCall
 	 *
 	 * @param auditingHandlerFactory must not be {@literal null}.
 	 */
-	public ReactiveAuditingEntityCallback(ObjectFactory<IsNewAwareAuditingHandler> auditingHandlerFactory) {
+	public ReactiveAuditingEntityCallback(ObjectFactory<ReactiveIsNewAwareAuditingHandler> auditingHandlerFactory) {
 
 		Assert.notNull(auditingHandlerFactory, "IsNewAwareAuditingHandler must not be null!");
 		this.auditingHandlerFactory = auditingHandlerFactory;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.mapping.event.ReactiveBeforeConvertCallback#onBeforeConvert(java.lang.Object, java.lang.String)
 	 */
 	@Override
 	public Publisher<Object> onBeforeConvert(Object entity, String collection) {
-		return Mono.just(auditingHandlerFactory.getObject().markAudited(entity));
+		return auditingHandlerFactory.getObject().markAudited(entity);
 	}
 
 	/*

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/query/BasicUpdate.java

@@ -67,6 +67,7 @@ public class BasicUpdate extends Update {
 	}
 
 	@Override
+	@Deprecated
 	public Update pushAll(String key, Object[] values) {
 		Document keyValue = new Document();
 		keyValue.put(key, values);

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/query/Field.java

@@ -20,47 +20,129 @@ import java.util.Map;
 import java.util.Map.Entry;
 
 import org.bson.Document;
+
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 
 /**
+ * Field projection.
+ *
  * @author Thomas Risberg
  * @author Oliver Gierke
  * @author Patryk Wasik
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Owen Q
  */
 public class Field {
 
-	private final Map<String, Integer> criteria = new HashMap<String, Integer>();
-	private final Map<String, Object> slices = new HashMap<String, Object>();
-	private final Map<String, Criteria> elemMatchs = new HashMap<String, Criteria>();
+	private final Map<String, Integer> criteria = new HashMap<>();
+	private final Map<String, Object> slices = new HashMap<>();
+	private final Map<String, Criteria> elemMatchs = new HashMap<>();
 	private @Nullable String positionKey;
 	private int positionValue;
 
-	public Field include(String key) {
-		criteria.put(key, Integer.valueOf(1));
+	/**
+	 * Include a single {@code field} to be returned by the query operation.
+	 *
+	 * @param field the document field name to be included.
+	 * @return {@code this} field projection instance.
+	 */
+	public Field include(String field) {
+
+		Assert.notNull(field, "Key must not be null!");
+
+		criteria.put(field, 1);
+
 		return this;
 	}
 
-	public Field exclude(String key) {
-		criteria.put(key, Integer.valueOf(0));
+	/**
+	 * Include one or more {@code fields} to be returned by the query operation.
+	 *
+	 * @param fields the document field names to be included.
+	 * @return {@code this} field projection instance.
+	 * @since 3.1
+	 */
+	public Field include(String... fields) {
+
+		Assert.notNull(fields, "Keys must not be null!");
+
+		for (String key : fields) {
+			criteria.put(key, 1);
+		}
+
 		return this;
 	}
 
-	public Field slice(String key, int size) {
-		slices.put(key, Integer.valueOf(size));
+	/**
+	 * Exclude a single {@code field} from being returned by the query operation.
+	 *
+	 * @param field the document field name to be included.
+	 * @return {@code this} field projection instance.
+	 */
+	public Field exclude(String field) {
+
+		Assert.notNull(field, "Key must not be null!");
+
+		criteria.put(field, 0);
+
 		return this;
 	}
 
-	public Field slice(String key, int offset, int size) {
-		slices.put(key, new Integer[] { Integer.valueOf(offset), Integer.valueOf(size) });
+	/**
+	 * Exclude one or more {@code fields} from being returned by the query operation.
+	 *
+	 * @param fields the document field names to be included.
+	 * @return {@code this} field projection instance.
+	 * @since 3.1
+	 */
+	public Field exclude(String... fields) {
+
+		Assert.notNull(fields, "Keys must not be null!");
+
+		for (String key : fields) {
+			criteria.put(key, 0);
+		}
+
 		return this;
 	}
 
-	public Field elemMatch(String key, Criteria elemMatchCriteria) {
-		elemMatchs.put(key, elemMatchCriteria);
+	/**
+	 * Project a {@code $slice} of the array {@code field} using the first {@code size} elements.
+	 *
+	 * @param field the document field name to project, must be an array field.
+	 * @param size the number of elements to include.
+	 * @return {@code this} field projection instance.
+	 */
+	public Field slice(String field, int size) {
+
+		Assert.notNull(field, "Key must not be null!");
+
+		slices.put(field, size);
+
+		return this;
+	}
+
+	/**
+	 * Project a {@code $slice} of the array {@code field} using the first {@code size} elements starting at
+	 * {@code offset}.
+	 *
+	 * @param field the document field name to project, must be an array field.
+	 * @param offset the offset to start at.
+	 * @param size the number of elements to include.
+	 * @return {@code this} field projection instance.
+	 */
+	public Field slice(String field, int offset, int size) {
+
+		slices.put(field, new Integer[] { offset, size });
+		return this;
+	}
+
+	public Field elemMatch(String field, Criteria elemMatchCriteria) {
+
+		elemMatchs.put(field, elemMatchCriteria);
 		return this;
 	}
 
@@ -70,7 +152,7 @@ public class Field {
 	 *
 	 * @param field query array field, must not be {@literal null} or empty.
 	 * @param value
-	 * @return
+	 * @return {@code this} field projection instance.
 	 */
 	public Field position(String field, int value) {
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/schema/TypedJsonSchemaObject.java

@@ -41,6 +41,7 @@ import org.springframework.util.StringUtils;
  *
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Michał Kurcius
  * @since 2.1
  */
 public class TypedJsonSchemaObject extends UntypedJsonSchemaObject {
@@ -1203,7 +1204,7 @@ public class TypedJsonSchemaObject extends UntypedJsonSchemaObject {
 			Document doc = new Document(super.toDocument());
 
 			if (!CollectionUtils.isEmpty(items)) {
-				doc.append("items", items.size() == 1 ? items.iterator().next()
+				doc.append("items", items.size() == 1 ? items.iterator().next().toDocument()
 						: items.stream().map(JsonSchemaObject::toDocument).collect(Collectors.toList()));
 			}
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/GridFsOperations.java

@@ -129,7 +129,7 @@ public interface GridFsOperations extends ResourcePatternResolver {
 	 *
 	 * @param content must not be {@literal null}.
 	 * @param filename must not be {@literal null} or empty.
-	 * @param contentType can be {@literal null}.
+	 * @param contentType can be {@literal null}. If not empty, may override content type within {@literal metadata}.
 	 * @param metadata can be {@literal null}.
 	 * @return the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just created.
 	 */
@@ -140,12 +140,12 @@ public interface GridFsOperations extends ResourcePatternResolver {
 		if (StringUtils.hasText(filename)) {
 			uploadBuilder.filename(filename);
 		}
-		if (StringUtils.hasText(contentType)) {
-			uploadBuilder.contentType(contentType);
-		}
 		if (!ObjectUtils.isEmpty(metadata)) {
 			uploadBuilder.metadata(metadata);
 		}
+		if (StringUtils.hasText(contentType)) {
+			uploadBuilder.contentType(contentType);
+		}
 
 		return store(uploadBuilder.build());
 	}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsOperations.java

@@ -135,7 +135,7 @@ public interface ReactiveGridFsOperations {
 	 *
 	 * @param content must not be {@literal null}.
 	 * @param filename must not be {@literal null} or empty.
-	 * @param contentType can be {@literal null}.
+	 * @param contentType can be {@literal null}. If not empty, may override content type within {@literal metadata}.
 	 * @param metadata can be {@literal null}.
 	 * @return a {@link Mono} emitting the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just
 	 *         created.
@@ -148,12 +148,12 @@ public interface ReactiveGridFsOperations {
 		if (StringUtils.hasText(filename)) {
 			uploadBuilder.filename(filename);
 		}
-		if (StringUtils.hasText(contentType)) {
-			uploadBuilder.contentType(contentType);
-		}
 		if (!ObjectUtils.isEmpty(metadata)) {
 			uploadBuilder.metadata(metadata);
 		}
+		if (StringUtils.hasText(contentType)) {
+			uploadBuilder.contentType(contentType);
+		}
 
 		return store(uploadBuilder.build());
 	}

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

@@ -16,6 +16,8 @@
 package org.springframework.data.mongodb.repository.query;
 
 import org.bson.Document;
+import org.bson.codecs.configuration.CodecRegistry;
+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;
@@ -30,10 +32,14 @@ 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.expression.spel.standard.SpelExpressionParser;
+import org.springframework.data.spel.ExpressionDependencies;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
+import com.mongodb.client.MongoDatabase;
+
 /**
  * Base class for {@link RepositoryQuery} implementations for Mongo.
  *
@@ -47,7 +53,7 @@ public abstract class AbstractMongoQuery implements RepositoryQuery {
 	private final MongoQueryMethod method;
 	private final MongoOperations operations;
 	private final ExecutableFind<?> executableFind;
-	private final SpelExpressionParser expressionParser;
+	private final ExpressionParser expressionParser;
 	private final QueryMethodEvaluationContextProvider evaluationContextProvider;
 
 	/**
@@ -58,7 +64,7 @@ public abstract class AbstractMongoQuery implements RepositoryQuery {
 	 * @param expressionParser must not be {@literal null}.
 	 * @param evaluationContextProvider must not be {@literal null}.
 	 */
-	public AbstractMongoQuery(MongoQueryMethod method, MongoOperations operations, SpelExpressionParser expressionParser,
+	public AbstractMongoQuery(MongoQueryMethod method, MongoOperations operations, ExpressionParser expressionParser,
 			QueryMethodEvaluationContextProvider evaluationContextProvider) {
 
 		Assert.notNull(operations, "MongoOperations must not be null!");
@@ -208,6 +214,29 @@ public abstract class AbstractMongoQuery implements RepositoryQuery {
 		return applyQueryMetaAttributesWhenPresent(createQuery(accessor));
 	}
 
+	/**
+	 * Obtain a the {@link EvaluationContext} suitable to evaluate expressions backed by the given dependencies.
+	 *
+	 * @param dependencies must not be {@literal null}.
+	 * @param accessor must not be {@literal null}.
+	 * @return the {@link SpELExpressionEvaluator}.
+	 * @since 2.4
+	 */
+	protected SpELExpressionEvaluator getSpELExpressionEvaluatorFor(ExpressionDependencies dependencies,
+			ConvertingParameterAccessor accessor) {
+
+		return new DefaultSpELExpressionEvaluator(expressionParser, evaluationContextProvider
+				.getEvaluationContext(getQueryMethod().getParameters(), accessor.getValues(), dependencies));
+	}
+
+	/**
+	 * @return the {@link CodecRegistry} used.
+	 * @since 2.4
+	 */
+	protected CodecRegistry getCodecRegistry() {
+		return operations.execute(MongoDatabase::getCodecRegistry);
+	}
+
 	/**
 	 * Creates a {@link Query} instance using the given {@link ParameterAccessor}
 	 *

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

@@ -15,13 +15,15 @@
  */
 package org.springframework.data.mongodb.repository.query;
 
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.bson.Document;
+import org.bson.codecs.configuration.CodecRegistry;
 import org.reactivestreams.Publisher;
+
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.mapping.model.EntityInstantiators;
+import org.springframework.data.mapping.model.SpELExpressionEvaluator;
 import org.springframework.data.mongodb.core.MongoOperations;
 import org.springframework.data.mongodb.core.ReactiveFindOperation.FindWithProjection;
 import org.springframework.data.mongodb.core.ReactiveFindOperation.FindWithQuery;
@@ -33,14 +35,17 @@ import org.springframework.data.mongodb.repository.query.ReactiveMongoQueryExecu
 import org.springframework.data.mongodb.repository.query.ReactiveMongoQueryExecution.ResultProcessingConverter;
 import org.springframework.data.mongodb.repository.query.ReactiveMongoQueryExecution.ResultProcessingExecution;
 import org.springframework.data.repository.query.ParameterAccessor;
-import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.data.repository.query.ReactiveQueryMethodEvaluationContextProvider;
 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.TypeInformation;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
+import com.mongodb.MongoClientSettings;
+
 /**
  * Base class for reactive {@link RepositoryQuery} implementations for MongoDB.
  *
@@ -54,8 +59,8 @@ public abstract class AbstractReactiveMongoQuery implements RepositoryQuery {
 	private final ReactiveMongoOperations operations;
 	private final EntityInstantiators instantiators;
 	private final FindWithProjection<?> findOperationWithProjection;
-	private final SpelExpressionParser expressionParser;
-	private final QueryMethodEvaluationContextProvider evaluationContextProvider;
+	private final ExpressionParser expressionParser;
+	private final ReactiveQueryMethodEvaluationContextProvider evaluationContextProvider;
 
 	/**
 	 * Creates a new {@link AbstractReactiveMongoQuery} from the given {@link MongoQueryMethod} and
@@ -67,12 +72,12 @@ public abstract class AbstractReactiveMongoQuery implements RepositoryQuery {
 	 * @param evaluationContextProvider must not be {@literal null}.
 	 */
 	public AbstractReactiveMongoQuery(ReactiveMongoQueryMethod method, ReactiveMongoOperations operations,
-			SpelExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
+			ExpressionParser expressionParser, ReactiveQueryMethodEvaluationContextProvider evaluationContextProvider) {
 
 		Assert.notNull(method, "MongoQueryMethod must not be null!");
 		Assert.notNull(operations, "ReactiveMongoOperations must not be null!");
 		Assert.notNull(expressionParser, "SpelExpressionParser must not be null!");
-		Assert.notNull(evaluationContextProvider, "QueryMethodEvaluationContextProvider must not be null!");
+		Assert.notNull(evaluationContextProvider, "ReactiveEvaluationContextExtension must not be null!");
 
 		this.method = method;
 		this.operations = operations;
@@ -98,25 +103,21 @@ public abstract class AbstractReactiveMongoQuery implements RepositoryQuery {
 	 * (non-Javadoc)
 	 * @see org.springframework.data.repository.query.RepositoryQuery#execute(java.lang.Object[])
 	 */
-	public Object execute(Object[] parameters) {
+	public Publisher<Object> execute(Object[] parameters) {
 
 		return method.hasReactiveWrapperParameter() ? executeDeferred(parameters)
 				: execute(new MongoParametersParameterAccessor(method, parameters));
 	}
 
 	@SuppressWarnings("unchecked")
-	private Object executeDeferred(Object[] parameters) {
+	private Publisher<Object> executeDeferred(Object[] parameters) {
 
 		ReactiveMongoParameterAccessor parameterAccessor = new ReactiveMongoParameterAccessor(method, parameters);
 
-		if (getQueryMethod().isCollectionQuery()) {
-			return Flux.defer(() -> (Publisher<Object>) execute(parameterAccessor));
-		}
-
-		return Mono.defer(() -> (Mono<Object>) execute(parameterAccessor));
+		return execute(parameterAccessor);
 	}
 
-	private Object execute(MongoParameterAccessor parameterAccessor) {
+	private Publisher<Object> execute(MongoParameterAccessor parameterAccessor) {
 
 		ConvertingParameterAccessor accessor = new ConvertingParameterAccessor(operations.getConverter(),
 				parameterAccessor);
@@ -141,24 +142,26 @@ public abstract class AbstractReactiveMongoQuery implements RepositoryQuery {
 	 * @param accessor for providing invocation arguments. Never {@literal null}.
 	 * @param typeToRead the desired component target type. Can be {@literal null}.
 	 */
-	protected Object doExecute(ReactiveMongoQueryMethod method, ResultProcessor processor,
+	protected Publisher<Object> doExecute(ReactiveMongoQueryMethod method, ResultProcessor processor,
 			ConvertingParameterAccessor accessor, @Nullable Class<?> typeToRead) {
 
-		Query query = createQuery(accessor);
+		return createQuery(accessor).flatMapMany(it -> {
 
-		applyQueryMetaAttributesWhenPresent(query);
-		query = applyAnnotatedDefaultSortIfPresent(query);
-		query = applyAnnotatedCollationIfPresent(query, accessor);
+			Query query = it;
+			applyQueryMetaAttributesWhenPresent(query);
+			query = applyAnnotatedDefaultSortIfPresent(query);
+			query = applyAnnotatedCollationIfPresent(query, accessor);
 
-		FindWithQuery<?> find = typeToRead == null //
-				? findOperationWithProjection //
-				: findOperationWithProjection.as(typeToRead);
+			FindWithQuery<?> find = typeToRead == null //
+					? findOperationWithProjection //
+					: findOperationWithProjection.as(typeToRead);
 
-		String collection = method.getEntityInformation().getCollectionName();
+			String collection = method.getEntityInformation().getCollectionName();
 
-		ReactiveMongoQueryExecution execution = getExecution(accessor,
-				new ResultProcessingConverter(processor, operations, instantiators), find);
-		return execution.execute(query, processor.getReturnedType().getDomainType(), collection);
+			ReactiveMongoQueryExecution execution = getExecution(accessor,
+					new ResultProcessingConverter(processor, operations, instantiators), find);
+			return execution.execute(query, processor.getReturnedType().getDomainType(), collection);
+		});
 	}
 
 	/**
@@ -254,8 +257,37 @@ public abstract class AbstractReactiveMongoQuery implements RepositoryQuery {
 	 * @param accessor must not be {@literal null}.
 	 * @return
 	 */
-	protected Query createCountQuery(ConvertingParameterAccessor accessor) {
-		return applyQueryMetaAttributesWhenPresent(createQuery(accessor));
+	protected Mono<Query> createCountQuery(ConvertingParameterAccessor accessor) {
+		return createQuery(accessor).map(this::applyQueryMetaAttributesWhenPresent);
+	}
+
+	/**
+	 * Obtain a {@link Mono publisher} emitting the {@link SpELExpressionEvaluator} suitable to evaluate expressions
+	 * backed by the given dependencies.
+	 *
+	 * @param dependencies must not be {@literal null}.
+	 * @param accessor must not be {@literal null}.
+	 * @return a {@link Mono} emitting the {@link SpELExpressionEvaluator} when ready.
+	 * @since 2.4
+	 */
+	protected Mono<SpELExpressionEvaluator> getSpelEvaluatorFor(ExpressionDependencies dependencies,
+			ConvertingParameterAccessor accessor) {
+
+		return evaluationContextProvider
+				.getEvaluationContextLater(getQueryMethod().getParameters(), accessor.getValues(), dependencies)
+				.map(evaluationContext -> (SpELExpressionEvaluator) new DefaultSpELExpressionEvaluator(expressionParser,
+						evaluationContext))
+				.defaultIfEmpty(DefaultSpELExpressionEvaluator.unsupported());
+	}
+
+	/**
+	 * @return a {@link Mono} emitting the {@link CodecRegistry} when ready.
+	 * @since 2.4
+	 */
+	protected Mono<CodecRegistry> getCodecRegistry() {
+
+		return Mono.from(operations.execute(db -> Mono.just(db.getCodecRegistry())))
+				.defaultIfEmpty(MongoClientSettings.getDefaultCodecRegistry());
 	}
 
 	/**
@@ -264,7 +296,7 @@ public abstract class AbstractReactiveMongoQuery implements RepositoryQuery {
 	 * @param accessor must not be {@literal null}.
 	 * @return
 	 */
-	protected abstract Query createQuery(ConvertingParameterAccessor accessor);
+	protected abstract Mono<Query> createQuery(ConvertingParameterAccessor accessor);
 
 	/**
 	 * Returns whether the query should get a count projection applied.

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

@@ -16,7 +16,6 @@
 package org.springframework.data.mongodb.repository.query;
 
 import java.time.Duration;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
@@ -25,16 +24,13 @@ import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort.Order;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
-import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
 import org.springframework.data.mongodb.core.convert.MongoConverter;
 import org.springframework.data.mongodb.core.query.Collation;
 import org.springframework.data.mongodb.core.query.Meta;
 import org.springframework.data.mongodb.core.query.Query;
-import org.springframework.data.mongodb.util.json.ParameterBindingContext;
-import org.springframework.data.mongodb.util.json.ParameterBindingDocumentCodec;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.lang.Nullable;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
@@ -49,10 +45,7 @@ import org.springframework.util.StringUtils;
  */
 abstract class AggregationUtils {
 
-	private static final ParameterBindingDocumentCodec CODEC = new ParameterBindingDocumentCodec();
-
-	private AggregationUtils() {
-	}
+	private AggregationUtils() {}
 
 	/**
 	 * Apply a collation extracted from the given {@literal collationExpression} to the given
@@ -64,12 +57,12 @@ abstract class AggregationUtils {
 	 * @param accessor must not be {@literal null}.
 	 * @return the {@link Query} having proper {@link Collation}.
 	 * @see AggregationOptions#getCollation()
-	 * @see CollationUtils#computeCollation(String, ConvertingParameterAccessor, MongoParameters, SpelExpressionParser,
+	 * @see CollationUtils#computeCollation(String, ConvertingParameterAccessor, MongoParameters, ExpressionParser,
 	 *      QueryMethodEvaluationContextProvider)
 	 */
 	static AggregationOptions.Builder applyCollation(AggregationOptions.Builder builder,
 			@Nullable String collationExpression, ConvertingParameterAccessor accessor, MongoParameters parameters,
-			SpelExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
+			ExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
 
 		Collation collation = CollationUtils.computeCollation(collationExpression, accessor, parameters, expressionParser,
 				evaluationContextProvider);
@@ -105,32 +98,6 @@ abstract class AggregationUtils {
 		return builder;
 	}
 
-	/**
-	 * Compute the {@link AggregationOperation aggregation} pipeline for the given {@link MongoQueryMethod}. The raw
-	 * {@link org.springframework.data.mongodb.repository.Aggregation#pipeline()} is parsed with a
-	 * {@link ParameterBindingDocumentCodec} to obtain the MongoDB native {@link Document} representation returned by
-	 * {@link AggregationOperation#toDocument(AggregationOperationContext)} that is mapped against the domain type
-	 * properties.
-	 *
-	 * @param method
-	 * @param accessor
-	 * @param expressionParser
-	 * @param evaluationContextProvider
-	 * @return
-	 */
-	static List<AggregationOperation> computePipeline(MongoQueryMethod method, ConvertingParameterAccessor accessor,
-			SpelExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
-
-		ParameterBindingContext bindingContext = new ParameterBindingContext((accessor::getBindableValue), expressionParser,
-				() -> evaluationContextProvider.getEvaluationContext(method.getParameters(), accessor.getValues()));
-
-		List<AggregationOperation> target = new ArrayList<>(method.getAnnotatedAggregation().length);
-		for (String source : method.getAnnotatedAggregation()) {
-			target.add(ctx -> ctx.getMappedObject(CODEC.decode(source, bindingContext), method.getDomainClass()));
-		}
-		return target;
-	}
-
 	/**
 	 * Append {@code $sort} aggregation stage if {@link ConvertingParameterAccessor#getSort()} is present.
 	 *

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

@@ -24,6 +24,7 @@ import org.springframework.data.mongodb.core.query.Collation;
 import org.springframework.data.mongodb.util.json.ParameterBindingContext;
 import org.springframework.data.mongodb.util.json.ParameterBindingDocumentCodec;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.lang.Nullable;
 import org.springframework.util.NumberUtils;
@@ -59,7 +60,7 @@ abstract class CollationUtils {
 	 */
 	@Nullable
 	static Collation computeCollation(@Nullable String collationExpression, ConvertingParameterAccessor accessor,
-			MongoParameters parameters, SpelExpressionParser expressionParser,
+			MongoParameters parameters, ExpressionParser expressionParser,
 			QueryMethodEvaluationContextProvider evaluationContextProvider) {
 
 		if (accessor.getCollation() != null) {
@@ -72,8 +73,9 @@ abstract class CollationUtils {
 
 		if (StringUtils.trimLeadingWhitespace(collationExpression).startsWith("{")) {
 
-			ParameterBindingContext bindingContext = new ParameterBindingContext((accessor::getBindableValue),
-					expressionParser, () -> evaluationContextProvider.getEvaluationContext(parameters, accessor.getValues()));
+			ParameterBindingContext bindingContext = ParameterBindingContext.forExpressions(accessor::getBindableValue,
+					expressionParser, dependencies -> evaluationContextProvider.getEvaluationContext(parameters,
+							accessor.getValues(), dependencies));
 
 			return Collation.from(CODEC.decode(collationExpression, bindingContext));
 		}

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

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2020 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.query;
+
+import org.springframework.data.mapping.model.SpELExpressionEvaluator;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.ExpressionParser;
+
+/**
+ * Simple {@link SpELExpressionEvaluator} implementation using {@link ExpressionParser} and {@link EvaluationContext}.
+ *
+ * @author Mark Paluch
+ * @since 3.1
+ */
+class DefaultSpELExpressionEvaluator implements SpELExpressionEvaluator {
+
+	private final ExpressionParser parser;
+	private final EvaluationContext context;
+
+	DefaultSpELExpressionEvaluator(ExpressionParser parser, EvaluationContext context) {
+		this.parser = parser;
+		this.context = context;
+	}
+
+	/**
+	 * Return a {@link SpELExpressionEvaluator} that does not support expression evaluation.
+	 *
+	 * @return a {@link SpELExpressionEvaluator} that does not support expression evaluation.
+	 * @since 3.1
+	 */
+	public static SpELExpressionEvaluator unsupported() {
+		return NoOpExpressionEvaluator.INSTANCE;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mapping.model.SpELExpressionEvaluator#evaluate(java.lang.String)
+	 */
+	@Override
+	@SuppressWarnings("unchecked")
+	public <T> T evaluate(String expression) {
+		return (T) parser.parseExpression(expression).getValue(context, Object.class);
+	}
+
+	/**
+	 * {@link SpELExpressionEvaluator} that does not support SpEL evaluation.
+	 *
+	 * @author Mark Paluch
+	 * @since 3.1
+	 */
+	enum NoOpExpressionEvaluator implements SpELExpressionEvaluator {
+
+		INSTANCE;
+
+		@Override
+		public <T> T evaluate(String expression) {
+			throw new UnsupportedOperationException("Expression evaluation not supported");
+		}
+	}
+}

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

@@ -32,7 +32,7 @@ import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
 import org.springframework.data.repository.query.ReturnedType;
 import org.springframework.data.repository.query.parser.PartTree;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.util.StringUtils;
 
 /**
@@ -59,7 +59,7 @@ public class PartTreeMongoQuery extends AbstractMongoQuery {
 	 * @param evaluationContextProvider must not be {@literal null}.
 	 */
 	public PartTreeMongoQuery(MongoQueryMethod method, MongoOperations mongoOperations,
-			SpelExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
+			ExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
 
 		super(method, mongoOperations, expressionParser, evaluationContextProvider);
 

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

@@ -21,6 +21,7 @@ import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.data.mongodb.core.query.Collation;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.lang.Nullable;
 
@@ -76,7 +77,7 @@ class QueryUtils {
 	 * @since 2.2
 	 */
 	static Query applyCollation(Query query, @Nullable String collationExpression, ConvertingParameterAccessor accessor,
-			MongoParameters parameters, SpelExpressionParser expressionParser,
+			MongoParameters parameters, ExpressionParser expressionParser,
 			QueryMethodEvaluationContextProvider evaluationContextProvider) {
 
 		Collation collation = CollationUtils.computeCollation(collationExpression, accessor, parameters, expressionParser,

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

@@ -32,6 +32,7 @@ import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.repository.query.ResultProcessor;
 import org.springframework.data.repository.query.ReturnedType;
 import org.springframework.data.repository.util.ReactiveWrappers;
+import org.springframework.data.util.ReflectionUtils;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -48,7 +49,7 @@ import org.springframework.util.ClassUtils;
  */
 interface ReactiveMongoQueryExecution {
 
-	Object execute(Query query, Class<?> type, String collection);
+	Publisher<? extends Object> execute(Query query, Class<?> type, String collection);
 
 	/**
 	 * {@link MongoQueryExecution} to execute geo-near queries.
@@ -74,7 +75,7 @@ interface ReactiveMongoQueryExecution {
 		 * @see org.springframework.data.mongodb.repository.query.AbstractMongoQuery.Execution#execute(org.springframework.data.mongodb.core.query.Query, java.lang.Class, java.lang.String)
 		 */
 		@Override
-		public Object execute(Query query, Class<?> type, String collection) {
+		public Publisher<? extends Object> execute(Query query, Class<?> type, String collection) {
 
 			Flux<GeoResult<Object>> results = doExecuteQuery(query, type, collection);
 			return isStreamOfGeoResult() ? results : results.map(GeoResult::getContent);
@@ -132,7 +133,7 @@ interface ReactiveMongoQueryExecution {
 		 * @see org.springframework.data.mongodb.repository.query.AbstractMongoQuery.Execution#execute(org.springframework.data.mongodb.core.query.Query, java.lang.Class, java.lang.String)
 		 */
 		@Override
-		public Object execute(Query query, Class<?> type, String collection) {
+		public Publisher<? extends Object> execute(Query query, Class<?> type, String collection) {
 
 			if (method.isCollectionQuery()) {
 				return operations.findAllAndRemove(query, type, collection);
@@ -166,8 +167,8 @@ interface ReactiveMongoQueryExecution {
 		}
 
 		@Override
-		public Object execute(Query query, Class<?> type, String collection) {
-			return converter.convert(delegate.execute(query, type, collection));
+		public Publisher<? extends Object> execute(Query query, Class<?> type, String collection) {
+			return (Publisher) converter.convert(delegate.execute(query, type, collection));
 		}
 	}
 
@@ -203,7 +204,7 @@ interface ReactiveMongoQueryExecution {
 
 			ReturnedType returnedType = processor.getReturnedType();
 
-			if (isVoid(returnedType)) {
+			if (ReflectionUtils.isVoid(returnedType.getReturnedType())) {
 
 				if (source instanceof Mono) {
 					return ((Mono<?>) source).then();
@@ -228,8 +229,4 @@ interface ReactiveMongoQueryExecution {
 			return processor.processResult(source, converter);
 		}
 	}
-
-	static boolean isVoid(ReturnedType returnedType) {
-		return returnedType.getReturnedType().equals(Void.class);
-	}
 }

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

@@ -94,8 +94,8 @@ public class ReactiveMongoQueryMethod extends MongoQueryMethod {
 		}
 
 		this.method = method;
-		this.isCollectionQuery = Lazy.of(() -> !(isPageQuery() || isSliceQuery())
-				&& ReactiveWrappers.isMultiValueType(metadata.getReturnType(method).getType()));
+		this.isCollectionQuery = Lazy.of(() -> (!(isPageQuery() || isSliceQuery())
+				&& ReactiveWrappers.isMultiValueType(metadata.getReturnType(method).getType()) || super.isCollectionQuery()));
 	}
 
 	/*

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

@@ -15,9 +15,10 @@
  */
 package org.springframework.data.mongodb.repository.query;
 
+import reactor.core.publisher.Mono;
+
 import org.bson.Document;
 import org.bson.json.JsonParseException;
-
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.data.mongodb.core.ReactiveMongoOperations;
@@ -26,12 +27,12 @@ import org.springframework.data.mongodb.core.query.BasicQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.TextCriteria;
 import org.springframework.data.repository.query.QueryMethod;
-import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.data.repository.query.ReactiveQueryMethodEvaluationContextProvider;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
 import org.springframework.data.repository.query.ReturnedType;
 import org.springframework.data.repository.query.parser.PartTree;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.util.StringUtils;
 
 /**
@@ -57,7 +58,7 @@ public class ReactivePartTreeMongoQuery extends AbstractReactiveMongoQuery {
 	 * @param evaluationContextProvider must not be {@literal null}.
 	 */
 	public ReactivePartTreeMongoQuery(ReactiveMongoQueryMethod method, ReactiveMongoOperations mongoOperations,
-			SpelExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
+			ExpressionParser expressionParser, ReactiveQueryMethodEvaluationContextProvider evaluationContextProvider) {
 
 		super(method, mongoOperations, expressionParser, evaluationContextProvider);
 
@@ -81,11 +82,28 @@ public class ReactivePartTreeMongoQuery extends AbstractReactiveMongoQuery {
 	 * @see org.springframework.data.mongodb.repository.query.AbstractMongoQuery#createQuery(org.springframework.data.mongodb.repository.query.ConvertingParameterAccessor, boolean)
 	 */
 	@Override
-	protected Query createQuery(ConvertingParameterAccessor accessor) {
+	protected Mono<Query> createQuery(ConvertingParameterAccessor accessor) {
+		return Mono.fromSupplier(() -> createQueryInternal(accessor, false));
+	}
 
-		MongoQueryCreator creator = new MongoQueryCreator(tree, accessor, context, isGeoNearQuery);
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.repository.query.AbstractReactiveMongoQuery#createCountQuery(org.springframework.data.mongodb.repository.query.ConvertingParameterAccessor)
+	 */
+	@Override
+	protected Mono<Query> createCountQuery(ConvertingParameterAccessor accessor) {
+		return Mono.fromSupplier(() -> createQueryInternal(accessor, true));
+	}
+
+	private Query createQueryInternal(ConvertingParameterAccessor accessor, boolean isCountQuery) {
+
+		MongoQueryCreator creator = new MongoQueryCreator(tree, accessor, context, isCountQuery ? false : isGeoNearQuery);
 		Query query = creator.createQuery();
 
+		if (isCountQuery) {
+			return query;
+		}
+
 		if (tree.isLimiting()) {
 			query.limit(tree.getMaxResults());
 		}
@@ -114,22 +132,12 @@ public class ReactivePartTreeMongoQuery extends AbstractReactiveMongoQuery {
 			result.setSortObject(query.getSortObject());
 
 			return result;
-
 		} catch (JsonParseException o_O) {
 			throw new IllegalStateException(String.format("Invalid query or field specification in %s!", getQueryMethod()),
 					o_O);
 		}
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.mongodb.repository.query.AbstractReactiveMongoQuery#createCountQuery(org.springframework.data.mongodb.repository.query.ConvertingParameterAccessor)
-	 */
-	@Override
-	protected Query createCountQuery(ConvertingParameterAccessor accessor) {
-		return new MongoQueryCreator(tree, accessor, context, false).createQuery();
-	}
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.repository.query.AbstractReactiveMongoQuery#isCountQuery()

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

@@ -16,11 +16,13 @@
 package org.springframework.data.mongodb.repository.query;
 
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
+import java.util.ArrayList;
 import java.util.List;
 
 import org.bson.Document;
-
+import org.reactivestreams.Publisher;
 import org.springframework.data.mongodb.core.ReactiveMongoOperations;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
@@ -29,9 +31,12 @@ import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
 import org.springframework.data.mongodb.core.convert.MongoConverter;
 import org.springframework.data.mongodb.core.mapping.MongoSimpleTypes;
 import org.springframework.data.mongodb.core.query.Query;
-import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.data.mongodb.util.json.ParameterBindingContext;
+import org.springframework.data.mongodb.util.json.ParameterBindingDocumentCodec;
+import org.springframework.data.repository.query.ReactiveQueryMethodEvaluationContextProvider;
 import org.springframework.data.repository.query.ResultProcessor;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.data.spel.ExpressionDependencies;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.util.ClassUtils;
 
 /**
@@ -44,8 +49,8 @@ import org.springframework.util.ClassUtils;
  */
 public class ReactiveStringBasedAggregation extends AbstractReactiveMongoQuery {
 
-	private final SpelExpressionParser expressionParser;
-	private final QueryMethodEvaluationContextProvider evaluationContextProvider;
+	private final ExpressionParser expressionParser;
+	private final ReactiveQueryMethodEvaluationContextProvider evaluationContextProvider;
 	private final ReactiveMongoOperations reactiveMongoOperations;
 	private final MongoConverter mongoConverter;
 
@@ -56,8 +61,8 @@ public class ReactiveStringBasedAggregation extends AbstractReactiveMongoQuery {
 	 * @param evaluationContextProvider must not be {@literal null}.
 	 */
 	public ReactiveStringBasedAggregation(ReactiveMongoQueryMethod method,
-			ReactiveMongoOperations reactiveMongoOperations, SpelExpressionParser expressionParser,
-			QueryMethodEvaluationContextProvider evaluationContextProvider) {
+			ReactiveMongoOperations reactiveMongoOperations, ExpressionParser expressionParser,
+			ReactiveQueryMethodEvaluationContextProvider evaluationContextProvider) {
 
 		super(method, reactiveMongoOperations, expressionParser, evaluationContextProvider);
 
@@ -72,52 +77,75 @@ public class ReactiveStringBasedAggregation extends AbstractReactiveMongoQuery {
 	 * @see org.springframework.data.mongodb.repository.query.AbstractReactiveMongoQuery#doExecute(org.springframework.data.mongodb.repository.query.ReactiveMongoQueryMethod, org.springframework.data.repository.query.ResultProcessor, org.springframework.data.mongodb.repository.query.ConvertingParameterAccessor, java.lang.Class)
 	 */
 	@Override
-	protected Object doExecute(ReactiveMongoQueryMethod method, ResultProcessor processor,
+	protected Publisher<Object> doExecute(ReactiveMongoQueryMethod method, ResultProcessor processor,
 			ConvertingParameterAccessor accessor, Class<?> typeToRead) {
 
-		Class<?> sourceType = method.getDomainClass();
-		Class<?> targetType = typeToRead;
+		return computePipeline(accessor).flatMapMany(it -> {
 
-		List<AggregationOperation> pipeline = computePipeline(accessor);
-		AggregationUtils.appendSortIfPresent(pipeline, accessor, typeToRead);
-		AggregationUtils.appendLimitAndOffsetIfPresent(pipeline, accessor);
+			Class<?> sourceType = method.getDomainClass();
+			Class<?> targetType = typeToRead;
 
-		boolean isSimpleReturnType = isSimpleReturnType(typeToRead);
-		boolean isRawReturnType = ClassUtils.isAssignable(org.bson.Document.class, typeToRead);
+			List<AggregationOperation> pipeline = it;
 
-		if (isSimpleReturnType || isRawReturnType) {
-			targetType = Document.class;
-		}
+			AggregationUtils.appendSortIfPresent(pipeline, accessor, typeToRead);
+			AggregationUtils.appendLimitAndOffsetIfPresent(pipeline, accessor);
 
-		AggregationOptions options = computeOptions(method, accessor);
-		TypedAggregation<?> aggregation = new TypedAggregation<>(sourceType, pipeline, options);
+			boolean isSimpleReturnType = isSimpleReturnType(typeToRead);
+			boolean isRawReturnType = ClassUtils.isAssignable(org.bson.Document.class, typeToRead);
 
-		Flux<?> flux = reactiveMongoOperations.aggregate(aggregation, targetType);
+			if (isSimpleReturnType || isRawReturnType) {
+				targetType = Document.class;
+			}
 
-		if (isSimpleReturnType && !isRawReturnType) {
-			flux = flux.handle((it, sink) -> {
+			AggregationOptions options = computeOptions(method, accessor);
+			TypedAggregation<?> aggregation = new TypedAggregation<>(sourceType, pipeline, options);
 
-				Object result = AggregationUtils.extractSimpleTypeResult((Document) it, typeToRead, mongoConverter);
+			Flux<?> flux = reactiveMongoOperations.aggregate(aggregation, targetType);
 
-				if (result != null) {
-					sink.next(result);
-				}
-			});
-		}
+			if (isSimpleReturnType && !isRawReturnType) {
+				flux = flux.handle((item, sink) -> {
 
-		if (method.isCollectionQuery()) {
-			return flux;
-		} else {
-			return flux.next();
-		}
+					Object result = AggregationUtils.extractSimpleTypeResult((Document) item, typeToRead, mongoConverter);
+
+					if (result != null) {
+						sink.next(result);
+					}
+				});
+			}
+
+			return method.isCollectionQuery() ? flux : flux.next();
+		});
 	}
 
 	private boolean isSimpleReturnType(Class<?> targetType) {
 		return MongoSimpleTypes.HOLDER.isSimpleType(targetType);
 	}
 
-	List<AggregationOperation> computePipeline(ConvertingParameterAccessor accessor) {
-		return AggregationUtils.computePipeline(getQueryMethod(), accessor, expressionParser, evaluationContextProvider);
+	private Mono<List<AggregationOperation>> computePipeline(ConvertingParameterAccessor accessor) {
+
+		return getCodecRegistry().map(ParameterBindingDocumentCodec::new).flatMap(codec -> {
+
+			String[] sourcePipeline = getQueryMethod().getAnnotatedAggregation();
+
+			List<Mono<AggregationOperation>> stages = new ArrayList<>(sourcePipeline.length);
+			for (String source : sourcePipeline) {
+				stages.add(computePipelineStage(source, accessor, codec));
+			}
+			return Flux.concat(stages).collectList();
+		});
+	}
+
+	private Mono<AggregationOperation> computePipelineStage(String source, ConvertingParameterAccessor accessor,
+			ParameterBindingDocumentCodec codec) {
+
+		ExpressionDependencies dependencies = codec.captureExpressionDependencies(source, accessor::getBindableValue,
+				expressionParser);
+
+		return getSpelEvaluatorFor(dependencies, accessor).map(it -> {
+
+			ParameterBindingContext bindingContext = new ParameterBindingContext(accessor::getBindableValue, it);
+			return ctx -> ctx.getMappedObject(codec.decode(source, bindingContext), getQueryMethod().getDomainClass());
+		});
 	}
 
 	private AggregationOptions computeOptions(MongoQueryMethod method, ConvertingParameterAccessor accessor) {
@@ -136,7 +164,7 @@ public class ReactiveStringBasedAggregation extends AbstractReactiveMongoQuery {
 	 * @see org.springframework.data.mongodb.repository.query.AbstractReactiveMongoQuery#createQuery(org.springframework.data.mongodb.repository.query.ConvertingParameterAccessor)
 	 */
 	@Override
-	protected Query createQuery(ConvertingParameterAccessor accessor) {
+	protected Mono<Query> createQuery(ConvertingParameterAccessor accessor) {
 		throw new UnsupportedOperationException("No query support for aggregation");
 	}
 

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

@@ -15,17 +15,21 @@
  */
 package org.springframework.data.mongodb.repository.query;
 
+import reactor.core.publisher.Mono;
+
 import org.bson.Document;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import org.springframework.data.mongodb.core.MongoOperations;
 import org.springframework.data.mongodb.core.ReactiveMongoOperations;
 import org.springframework.data.mongodb.core.query.BasicQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.util.json.ParameterBindingContext;
 import org.springframework.data.mongodb.util.json.ParameterBindingDocumentCodec;
-import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.data.repository.query.ReactiveExtensionAwareQueryMethodEvaluationContextProvider;
+import org.springframework.data.repository.query.ReactiveQueryMethodEvaluationContextProvider;
+import org.springframework.data.spel.ExpressionDependencies;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.util.Assert;
 
@@ -40,13 +44,12 @@ public class ReactiveStringBasedMongoQuery extends AbstractReactiveMongoQuery {
 
 	private static final String COUNT_EXISTS_AND_DELETE = "Manually defined query for %s cannot be a count and exists or delete query at the same time!";
 	private static final Logger LOG = LoggerFactory.getLogger(ReactiveStringBasedMongoQuery.class);
-	private static final ParameterBindingDocumentCodec CODEC = new ParameterBindingDocumentCodec();
 
 	private final String query;
 	private final String fieldSpec;
 
-	private final SpelExpressionParser expressionParser;
-	private final QueryMethodEvaluationContextProvider evaluationContextProvider;
+	private final ExpressionParser expressionParser;
+	private final ReactiveQueryMethodEvaluationContextProvider evaluationContextProvider;
 
 	private final boolean isCountQuery;
 	private final boolean isExistsQuery;
@@ -62,13 +65,14 @@ public class ReactiveStringBasedMongoQuery extends AbstractReactiveMongoQuery {
 	 * @param evaluationContextProvider must not be {@literal null}.
 	 */
 	public ReactiveStringBasedMongoQuery(ReactiveMongoQueryMethod method, ReactiveMongoOperations mongoOperations,
-			SpelExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
+			ExpressionParser expressionParser, ReactiveQueryMethodEvaluationContextProvider evaluationContextProvider) {
 		this(method.getAnnotatedQuery(), method, mongoOperations, expressionParser, evaluationContextProvider);
 	}
 
 	/**
 	 * Creates a new {@link ReactiveStringBasedMongoQuery} for the given {@link String}, {@link MongoQueryMethod},
-	 * {@link MongoOperations}, {@link SpelExpressionParser} and {@link QueryMethodEvaluationContextProvider}.
+	 * {@link MongoOperations}, {@link SpelExpressionParser} and
+	 * {@link ReactiveExtensionAwareQueryMethodEvaluationContextProvider}.
 	 *
 	 * @param query must not be {@literal null}.
 	 * @param method must not be {@literal null}.
@@ -76,8 +80,8 @@ public class ReactiveStringBasedMongoQuery extends AbstractReactiveMongoQuery {
 	 * @param expressionParser must not be {@literal null}.
 	 */
 	public ReactiveStringBasedMongoQuery(String query, ReactiveMongoQueryMethod method,
-			ReactiveMongoOperations mongoOperations, SpelExpressionParser expressionParser,
-			QueryMethodEvaluationContextProvider evaluationContextProvider) {
+			ReactiveMongoOperations mongoOperations, ExpressionParser expressionParser,
+			ReactiveQueryMethodEvaluationContextProvider evaluationContextProvider) {
 
 		super(method, mongoOperations, expressionParser, evaluationContextProvider);
 
@@ -114,21 +118,36 @@ public class ReactiveStringBasedMongoQuery extends AbstractReactiveMongoQuery {
 	 * @see org.springframework.data.mongodb.repository.query.AbstractReactiveMongoQuery#createQuery(org.springframework.data.mongodb.repository.query.ConvertingParameterAccessor)
 	 */
 	@Override
-	protected Query createQuery(ConvertingParameterAccessor accessor) {
+	protected Mono<Query> createQuery(ConvertingParameterAccessor accessor) {
 
-		ParameterBindingContext bindingContext = new ParameterBindingContext((accessor::getBindableValue), expressionParser,
-				() -> evaluationContextProvider.getEvaluationContext(getQueryMethod().getParameters(), accessor.getValues()));
+		return getCodecRegistry().map(ParameterBindingDocumentCodec::new).flatMap(codec -> {
 
-		Document queryObject = CODEC.decode(this.query, bindingContext);
-		Document fieldsObject = CODEC.decode(this.fieldSpec, bindingContext);
+			Mono<Document> queryObject = getBindingContext(query, accessor, codec)
+					.map(context -> codec.decode(query, context));
+			Mono<Document> fieldsObject = getBindingContext(fieldSpec, accessor, codec)
+					.map(context -> codec.decode(fieldSpec, context));
 
-		Query query = new BasicQuery(queryObject, fieldsObject).with(accessor.getSort());
+			return queryObject.zipWith(fieldsObject).map(tuple -> {
 
-		if (LOG.isDebugEnabled()) {
-			LOG.debug(String.format("Created query %s for %s fields.", query.getQueryObject(), query.getFieldsObject()));
-		}
+				Query query = new BasicQuery(tuple.getT1(), tuple.getT2()).with(accessor.getSort());
 
-		return query;
+				if (LOG.isDebugEnabled()) {
+					LOG.debug(String.format("Created query %s for %s fields.", query.getQueryObject(), query.getFieldsObject()));
+				}
+
+				return query;
+			});
+		});
+	}
+
+	private Mono<ParameterBindingContext> getBindingContext(String json, ConvertingParameterAccessor accessor,
+			ParameterBindingDocumentCodec codec) {
+
+		ExpressionDependencies dependencies = codec.captureExpressionDependencies(json, accessor::getBindableValue,
+				expressionParser);
+
+		return getSpelEvaluatorFor(dependencies, accessor)
+				.map(it -> new ParameterBindingContext(accessor::getBindableValue, it));
 	}
 
 	/*

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

@@ -15,11 +15,12 @@
  */
 package org.springframework.data.mongodb.repository.query;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
 
 import org.bson.Document;
-
+import org.springframework.data.mapping.model.SpELExpressionEvaluator;
 import org.springframework.data.mongodb.InvalidMongoDbApiUsageException;
 import org.springframework.data.mongodb.core.MongoOperations;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
@@ -30,9 +31,12 @@ import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
 import org.springframework.data.mongodb.core.convert.MongoConverter;
 import org.springframework.data.mongodb.core.mapping.MongoSimpleTypes;
 import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.util.json.ParameterBindingContext;
+import org.springframework.data.mongodb.util.json.ParameterBindingDocumentCodec;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
 import org.springframework.data.repository.query.ResultProcessor;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.data.spel.ExpressionDependencies;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.util.ClassUtils;
 
 /**
@@ -43,7 +47,7 @@ public class StringBasedAggregation extends AbstractMongoQuery {
 
 	private final MongoOperations mongoOperations;
 	private final MongoConverter mongoConverter;
-	private final SpelExpressionParser expressionParser;
+	private final ExpressionParser expressionParser;
 	private final QueryMethodEvaluationContextProvider evaluationContextProvider;
 
 	/**
@@ -55,7 +59,7 @@ public class StringBasedAggregation extends AbstractMongoQuery {
 	 * @param evaluationContextProvider
 	 */
 	public StringBasedAggregation(MongoQueryMethod method, MongoOperations mongoOperations,
-			SpelExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
+			ExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
 		super(method, mongoOperations, expressionParser, evaluationContextProvider);
 
 		this.mongoOperations = mongoOperations;
@@ -73,7 +77,9 @@ public class StringBasedAggregation extends AbstractMongoQuery {
 			ConvertingParameterAccessor accessor, Class<?> typeToRead) {
 
 		if (method.isPageQuery() || method.isSliceQuery()) {
-			throw new InvalidMongoDbApiUsageException(String.format("Repository aggregation method '%s' does not support '%s' return type. Please use eg. 'List' instead.", method.getName(), method.getReturnType().getType().getSimpleName()));
+			throw new InvalidMongoDbApiUsageException(String.format(
+					"Repository aggregation method '%s' does not support '%s' return type. Please use eg. 'List' instead.",
+					method.getName(), method.getReturnType().getType().getSimpleName()));
 		}
 
 		Class<?> sourceType = method.getDomainClass();
@@ -125,7 +131,26 @@ public class StringBasedAggregation extends AbstractMongoQuery {
 	}
 
 	List<AggregationOperation> computePipeline(MongoQueryMethod method, ConvertingParameterAccessor accessor) {
-		return AggregationUtils.computePipeline(method, accessor, expressionParser, evaluationContextProvider);
+
+		ParameterBindingDocumentCodec codec = new ParameterBindingDocumentCodec(getCodecRegistry());
+		String[] sourcePipeline = method.getAnnotatedAggregation();
+
+		List<AggregationOperation> stages = new ArrayList<>(sourcePipeline.length);
+		for (String source : sourcePipeline) {
+			stages.add(computePipelineStage(source, accessor, codec));
+		}
+		return stages;
+	}
+
+	private AggregationOperation computePipelineStage(String source, ConvertingParameterAccessor accessor,
+			ParameterBindingDocumentCodec codec) {
+
+		ExpressionDependencies dependencies = codec.captureExpressionDependencies(source, accessor::getBindableValue,
+				expressionParser);
+
+		SpELExpressionEvaluator evaluator = getSpELExpressionEvaluatorFor(dependencies, accessor);
+		ParameterBindingContext bindingContext = new ParameterBindingContext(accessor::getBindableValue, evaluator);
+		return ctx -> ctx.getMappedObject(codec.decode(source, bindingContext), getQueryMethod().getDomainClass());
 	}
 
 	private AggregationOptions computeOptions(MongoQueryMethod method, ConvertingParameterAccessor accessor) {

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

@@ -16,21 +16,20 @@
 package org.springframework.data.mongodb.repository.query;
 
 import org.bson.Document;
-import org.bson.codecs.configuration.CodecRegistry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.data.mapping.model.SpELExpressionEvaluator;
 import org.springframework.data.mongodb.core.MongoOperations;
 import org.springframework.data.mongodb.core.query.BasicQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.util.json.ParameterBindingContext;
 import org.springframework.data.mongodb.util.json.ParameterBindingDocumentCodec;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.data.spel.ExpressionDependencies;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.util.Assert;
 
-import com.mongodb.MongoClientSettings;
-import com.mongodb.client.MongoDatabase;
-
 /**
  * Query to use a plain JSON String to create the {@link Query} to actually execute.
  *
@@ -47,8 +46,7 @@ public class StringBasedMongoQuery extends AbstractMongoQuery {
 	private final String query;
 	private final String fieldSpec;
 
-	private final ParameterBindingDocumentCodec codec;
-	private final SpelExpressionParser expressionParser;
+	private final ExpressionParser expressionParser;
 	private final QueryMethodEvaluationContextProvider evaluationContextProvider;
 
 	private final boolean isCountQuery;
@@ -65,7 +63,7 @@ public class StringBasedMongoQuery extends AbstractMongoQuery {
 	 * @param evaluationContextProvider must not be {@literal null}.
 	 */
 	public StringBasedMongoQuery(MongoQueryMethod method, MongoOperations mongoOperations,
-			SpelExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
+			ExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
 		this(method.getAnnotatedQuery(), method, mongoOperations, expressionParser, evaluationContextProvider);
 	}
 
@@ -79,7 +77,7 @@ public class StringBasedMongoQuery extends AbstractMongoQuery {
 	 * @param expressionParser must not be {@literal null}.
 	 */
 	public StringBasedMongoQuery(String query, MongoQueryMethod method, MongoOperations mongoOperations,
-			SpelExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
+			ExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
 
 		super(method, mongoOperations, expressionParser, evaluationContextProvider);
 
@@ -109,10 +107,6 @@ public class StringBasedMongoQuery extends AbstractMongoQuery {
 			this.isExistsQuery = false;
 			this.isDeleteQuery = false;
 		}
-
-		CodecRegistry codecRegistry = mongoOperations.execute(MongoDatabase::getCodecRegistry);
-		this.codec = new ParameterBindingDocumentCodec(
-				codecRegistry != null ? codecRegistry : MongoClientSettings.getDefaultCodecRegistry());
 	}
 
 	/*
@@ -122,11 +116,10 @@ public class StringBasedMongoQuery extends AbstractMongoQuery {
 	@Override
 	protected Query createQuery(ConvertingParameterAccessor accessor) {
 
-		ParameterBindingContext bindingContext = new ParameterBindingContext((accessor::getBindableValue), expressionParser,
-				() -> evaluationContextProvider.getEvaluationContext(getQueryMethod().getParameters(), accessor.getValues()));
+		ParameterBindingDocumentCodec codec = getParameterBindingCodec();
 
-		Document queryObject = codec.decode(this.query, bindingContext);
-		Document fieldsObject = codec.decode(this.fieldSpec, bindingContext);
+		Document queryObject = codec.decode(this.query, getBindingContext(this.query, accessor, codec));
+		Document fieldsObject = codec.decode(this.fieldSpec, getBindingContext(this.fieldSpec, accessor, codec));
 
 		Query query = new BasicQuery(queryObject, fieldsObject).with(accessor.getSort());
 
@@ -137,6 +130,16 @@ public class StringBasedMongoQuery extends AbstractMongoQuery {
 		return query;
 	}
 
+	private ParameterBindingContext getBindingContext(String json, ConvertingParameterAccessor accessor,
+			ParameterBindingDocumentCodec codec) {
+
+		ExpressionDependencies dependencies = codec.captureExpressionDependencies(json, accessor::getBindableValue,
+				expressionParser);
+
+		SpELExpressionEvaluator evaluator = getSpELExpressionEvaluatorFor(dependencies, accessor);
+		return new ParameterBindingContext(accessor::getBindableValue, evaluator);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.repository.query.AbstractMongoQuery#isCountQuery()
@@ -177,4 +180,8 @@ public class StringBasedMongoQuery extends AbstractMongoQuery {
 			boolean isDeleteQuery) {
 		return BooleanUtil.countBooleanTrueValues(isCountQuery, isExistsQuery, isDeleteQuery) > 1;
 	}
+
+	private ParameterBindingDocumentCodec getParameterBindingCodec() {
+		return new ParameterBindingDocumentCodec(getCodecRegistry());
+	}
 }

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

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2020 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.support;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.springframework.expression.Expression;
+import org.springframework.expression.ExpressionParser;
+import org.springframework.expression.ParseException;
+import org.springframework.expression.ParserContext;
+
+/**
+ * Caching variant of {@link ExpressionParser}. This implementation does not support
+ * {@link #parseExpression(String, ParserContext) parsing with ParseContext}.
+ *
+ * @author Mark Paluch
+ * @since 3.1
+ */
+class CachingExpressionParser implements ExpressionParser {
+
+	private final ExpressionParser delegate;
+	private final Map<String, Expression> cache = new ConcurrentHashMap<>();
+
+	CachingExpressionParser(ExpressionParser delegate) {
+		this.delegate = delegate;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.expression.ExpressionParser#parseExpression(java.lang.String)
+	 */
+	@Override
+	public Expression parseExpression(String expressionString) throws ParseException {
+		return cache.computeIfAbsent(expressionString, delegate::parseExpression);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.expression.ExpressionParser#parseExpression(java.lang.String, org.springframework.expression.ParserContext)
+	 */
+	@Override
+	public Expression parseExpression(String expressionString, ParserContext context) throws ParseException {
+		throw new UnsupportedOperationException("Parsing using ParserContext is not supported");
+	}
+}

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

@@ -44,6 +44,7 @@ import org.springframework.data.repository.query.QueryLookupStrategy;
 import org.springframework.data.repository.query.QueryLookupStrategy.Key;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
 import org.springframework.data.repository.query.RepositoryQuery;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -162,7 +163,8 @@ public class MongoRepositoryFactory extends RepositoryFactorySupport {
 
 		private final MongoOperations operations;
 		private final QueryMethodEvaluationContextProvider evaluationContextProvider;
-		MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty> mappingContext;
+		private final MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty> mappingContext;
+		private final ExpressionParser expressionParser = new CachingExpressionParser(EXPRESSION_PARSER);
 
 		public MongoQueryLookupStrategy(MongoOperations operations,
 				QueryMethodEvaluationContextProvider evaluationContextProvider,
@@ -186,14 +188,14 @@ public class MongoRepositoryFactory extends RepositoryFactorySupport {
 
 			if (namedQueries.hasQuery(namedQueryName)) {
 				String namedQuery = namedQueries.getQuery(namedQueryName);
-				return new StringBasedMongoQuery(namedQuery, queryMethod, operations, EXPRESSION_PARSER,
+				return new StringBasedMongoQuery(namedQuery, queryMethod, operations, expressionParser,
 						evaluationContextProvider);
 			} else if (queryMethod.hasAnnotatedAggregation()) {
-				return new StringBasedAggregation(queryMethod, operations, EXPRESSION_PARSER, evaluationContextProvider);
+				return new StringBasedAggregation(queryMethod, operations, expressionParser, evaluationContextProvider);
 			} else if (queryMethod.hasAnnotatedQuery()) {
-				return new StringBasedMongoQuery(queryMethod, operations, EXPRESSION_PARSER, evaluationContextProvider);
+				return new StringBasedMongoQuery(queryMethod, operations, expressionParser, evaluationContextProvider);
 			} else {
-				return new PartTreeMongoQuery(queryMethod, operations, EXPRESSION_PARSER, evaluationContextProvider);
+				return new PartTreeMongoQuery(queryMethod, operations, expressionParser, evaluationContextProvider);
 			}
 		}
 	}

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

@@ -42,7 +42,9 @@ import org.springframework.data.repository.core.support.RepositoryFragment;
 import org.springframework.data.repository.query.QueryLookupStrategy;
 import org.springframework.data.repository.query.QueryLookupStrategy.Key;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.data.repository.query.ReactiveQueryMethodEvaluationContextProvider;
 import org.springframework.data.repository.query.RepositoryQuery;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -73,6 +75,7 @@ public class ReactiveMongoRepositoryFactory extends ReactiveRepositoryFactorySup
 
 		this.operations = mongoOperations;
 		this.mappingContext = mongoOperations.getConverter().getMappingContext();
+		setEvaluationContextProvider(ReactiveQueryMethodEvaluationContextProvider.DEFAULT);
 	}
 
 	/*
@@ -127,7 +130,8 @@ public class ReactiveMongoRepositoryFactory extends ReactiveRepositoryFactorySup
 	@Override
 	protected Optional<QueryLookupStrategy> getQueryLookupStrategy(@Nullable Key key,
 			QueryMethodEvaluationContextProvider evaluationContextProvider) {
-		return Optional.of(new MongoQueryLookupStrategy(operations, evaluationContextProvider, mappingContext));
+		return Optional.of(new MongoQueryLookupStrategy(operations,
+				(ReactiveQueryMethodEvaluationContextProvider) evaluationContextProvider, mappingContext));
 	}
 
 	/*
@@ -157,11 +161,12 @@ public class ReactiveMongoRepositoryFactory extends ReactiveRepositoryFactorySup
 	private static class MongoQueryLookupStrategy implements QueryLookupStrategy {
 
 		private final ReactiveMongoOperations operations;
-		private final QueryMethodEvaluationContextProvider evaluationContextProvider;
+		private final ReactiveQueryMethodEvaluationContextProvider evaluationContextProvider;
 		private final MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty> mappingContext;
+		private final ExpressionParser expressionParser = new CachingExpressionParser(EXPRESSION_PARSER);
 
 		MongoQueryLookupStrategy(ReactiveMongoOperations operations,
-				QueryMethodEvaluationContextProvider evaluationContextProvider,
+				ReactiveQueryMethodEvaluationContextProvider evaluationContextProvider,
 				MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty> mappingContext) {
 
 			this.operations = operations;
@@ -182,15 +187,15 @@ public class ReactiveMongoRepositoryFactory extends ReactiveRepositoryFactorySup
 
 			if (namedQueries.hasQuery(namedQueryName)) {
 				String namedQuery = namedQueries.getQuery(namedQueryName);
-				return new ReactiveStringBasedMongoQuery(namedQuery, queryMethod, operations, EXPRESSION_PARSER,
+				return new ReactiveStringBasedMongoQuery(namedQuery, queryMethod, operations, expressionParser,
 						evaluationContextProvider);
 			} else if (queryMethod.hasAnnotatedAggregation()) {
-				return new ReactiveStringBasedAggregation(queryMethod, operations, EXPRESSION_PARSER,
+				return new ReactiveStringBasedAggregation(queryMethod, operations, expressionParser,
 						evaluationContextProvider);
 			} else if (queryMethod.hasAnnotatedQuery()) {
-				return new ReactiveStringBasedMongoQuery(queryMethod, operations, EXPRESSION_PARSER, evaluationContextProvider);
+				return new ReactiveStringBasedMongoQuery(queryMethod, operations, expressionParser, evaluationContextProvider);
 			} else {
-				return new ReactivePartTreeMongoQuery(queryMethod, operations, EXPRESSION_PARSER, evaluationContextProvider);
+				return new ReactivePartTreeMongoQuery(queryMethod, operations, expressionParser, evaluationContextProvider);
 			}
 		}
 	}

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

@@ -16,13 +16,17 @@
 package org.springframework.data.mongodb.repository.support;
 
 import java.io.Serializable;
+import java.util.Optional;
 
+import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mongodb.core.ReactiveMongoOperations;
 import org.springframework.data.mongodb.core.index.IndexOperationsAdapter;
 import org.springframework.data.repository.Repository;
 import org.springframework.data.repository.core.support.RepositoryFactoryBeanSupport;
 import org.springframework.data.repository.core.support.RepositoryFactorySupport;
+import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.data.repository.query.ReactiveExtensionAwareQueryMethodEvaluationContextProvider;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -35,6 +39,7 @@ import org.springframework.util.Assert;
  * @since 2.0
  * @see org.springframework.data.repository.reactive.ReactiveSortingRepository
  * @see org.springframework.data.repository.reactive.RxJava2SortingRepository
+ * @see org.springframework.data.repository.reactive.RxJava3SortingRepository
  */
 public class ReactiveMongoRepositoryFactoryBean<T extends Repository<S, ID>, S, ID extends Serializable>
 		extends RepositoryFactoryBeanSupport<T, S, ID> {
@@ -83,10 +88,7 @@ public class ReactiveMongoRepositoryFactoryBean<T extends Repository<S, ID>, S,
 
 	/*
 	 * (non-Javadoc)
-	 *
-	 * @see
-	 * org.springframework.data.repository.support.RepositoryFactoryBeanSupport
-	 * #createRepositoryFactory()
+	 * @see org.springframework.data.repository.core.support.RepositoryFactoryBeanSupport#createRepositoryFactory()
 	 */
 	@Override
 	protected RepositoryFactorySupport createRepositoryFactory() {
@@ -101,6 +103,16 @@ public class ReactiveMongoRepositoryFactoryBean<T extends Repository<S, ID>, S,
 		return factory;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.core.support.RepositoryFactoryBeanSupport#createDefaultQueryMethodEvaluationContextProvider(ListableBeanFactory)
+	 */
+	@Override
+	protected Optional<QueryMethodEvaluationContextProvider> createDefaultQueryMethodEvaluationContextProvider(
+			ListableBeanFactory beanFactory) {
+		return Optional.of(new ReactiveExtensionAwareQueryMethodEvaluationContextProvider(beanFactory));
+	}
+
 	/**
 	 * Creates and initializes a {@link RepositoryFactorySupport} instance.
 	 *
@@ -113,10 +125,7 @@ public class ReactiveMongoRepositoryFactoryBean<T extends Repository<S, ID>, S,
 
 	/*
 	 * (non-Javadoc)
-	 *
-	 * @see
-	 * org.springframework.data.repository.support.RepositoryFactoryBeanSupport
-	 * #afterPropertiesSet()
+	 * @see org.springframework.data.repository.core.support.RepositoryFactoryBeanSupport#afterPropertiesSet()
 	 */
 	@Override
 	public void afterPropertiesSet() {

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

@@ -50,6 +50,7 @@ import com.mongodb.client.result.DeleteResult;
  * @author Christoph Strobl
  * @author Thomas Darimont
  * @author Mark Paluch
+ * @author Mehran Behnam
  */
 public class SimpleMongoRepository<T, ID> implements MongoRepository<T, ID> {
 
@@ -97,7 +98,7 @@ public class SimpleMongoRepository<T, ID> implements MongoRepository<T, ID> {
 		Assert.notNull(entities, "The given Iterable of entities not be null!");
 
 		Streamable<S> source = Streamable.of(entities);
-		boolean allNew = source.stream().allMatch(it -> entityInformation.isNew(it));
+		boolean allNew = source.stream().allMatch(entityInformation::isNew);
 
 		if (allNew) {
 
@@ -211,6 +212,8 @@ public class SimpleMongoRepository<T, ID> implements MongoRepository<T, ID> {
 	@Override
 	public Iterable<T> findAllById(Iterable<ID> ids) {
 
+		Assert.notNull(ids, "The given Ids of entities not be null!");
+
 		return findAll(new Query(new Criteria(entityInformation.getIdAttribute())
 				.in(Streamable.of(ids).stream().collect(StreamUtils.toUnmodifiableList()))));
 	}
@@ -224,7 +227,7 @@ public class SimpleMongoRepository<T, ID> implements MongoRepository<T, ID> {
 
 		Assert.notNull(pageable, "Pageable must not be null!");
 
-		Long count = count();
+		long count = count();
 		List<T> list = findAll(new Query().with(pageable));
 
 		return new PageImpl<>(list, pageable, count);

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/json/ParameterBindingContext.java

@@ -15,11 +15,15 @@
  */
 package org.springframework.data.mongodb.util.json;
 
+import java.util.function.Function;
 import java.util.function.Supplier;
 
+import org.springframework.data.mapping.model.SpELExpressionEvaluator;
+import org.springframework.data.spel.ExpressionDependencies;
 import org.springframework.data.util.Lazy;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.lang.Nullable;
 
@@ -28,25 +32,21 @@ import org.springframework.lang.Nullable;
  * To be used along with {@link ParameterBindingDocumentCodec#decode(String, ParameterBindingContext)}.
  *
  * @author Christoph Strobl
+ * @author Mark Paluch
  * @since 2.2
  */
 public class ParameterBindingContext {
 
 	private final ValueProvider valueProvider;
-	private final SpelExpressionParser expressionParser;
-	private final Lazy<EvaluationContext> evaluationContext;
+	private final SpELExpressionEvaluator expressionEvaluator;
 
 	/**
 	 * @param valueProvider
 	 * @param expressionParser
 	 * @param evaluationContext
-	 * @deprecated since 2.2.3 - Please use
-	 *             {@link #ParameterBindingContext(ValueProvider, SpelExpressionParser, Supplier)} instead.
 	 */
-	@Deprecated
 	public ParameterBindingContext(ValueProvider valueProvider, SpelExpressionParser expressionParser,
 			EvaluationContext evaluationContext) {
-
 		this(valueProvider, expressionParser, () -> evaluationContext);
 	}
 
@@ -56,12 +56,51 @@ public class ParameterBindingContext {
 	 * @param evaluationContext a {@link Supplier} for {@link Lazy} context retrieval.
 	 * @since 2.2.3
 	 */
-	public ParameterBindingContext(ValueProvider valueProvider, SpelExpressionParser expressionParser,
+	public ParameterBindingContext(ValueProvider valueProvider, ExpressionParser expressionParser,
 			Supplier<EvaluationContext> evaluationContext) {
 
+		this(valueProvider, new SpELExpressionEvaluator() {
+			@Override
+			public <T> T evaluate(String expressionString) {
+				return (T) expressionParser.parseExpression(expressionString).getValue(evaluationContext.get(), Object.class);
+			}
+		});
+	}
+
+	/**
+	 * @param valueProvider
+	 * @param expressionEvaluator
+	 * @since 3.1
+	 */
+	public ParameterBindingContext(ValueProvider valueProvider, SpELExpressionEvaluator expressionEvaluator) {
 		this.valueProvider = valueProvider;
-		this.expressionParser = expressionParser;
-		this.evaluationContext = evaluationContext instanceof Lazy ? (Lazy) evaluationContext : Lazy.of(evaluationContext);
+		this.expressionEvaluator = expressionEvaluator;
+	}
+
+	/**
+	 * Create a new {@link ParameterBindingContext} that is capable of expression parsing and can provide a
+	 * {@link EvaluationContext} based on {@link ExpressionDependencies}.
+	 *
+	 * @param valueProvider
+	 * @param expressionParser
+	 * @param contextFunction
+	 * @return
+	 * @since 3.1
+	 */
+	public static ParameterBindingContext forExpressions(ValueProvider valueProvider,
+			ExpressionParser expressionParser, Function<ExpressionDependencies, EvaluationContext> contextFunction) {
+
+		return new ParameterBindingContext(valueProvider, new SpELExpressionEvaluator() {
+			@Override
+			public <T> T evaluate(String expressionString) {
+
+				Expression expression = expressionParser.parseExpression(expressionString);
+				ExpressionDependencies dependencies = ExpressionDependencies.discover(expression);
+				EvaluationContext evaluationContext = contextFunction.apply(dependencies);
+
+				return (T) expression.getValue(evaluationContext, Object.class);
+			}
+		});
 	}
 
 	@Nullable
@@ -71,17 +110,7 @@ public class ParameterBindingContext {
 
 	@Nullable
 	public Object evaluateExpression(String expressionString) {
-
-		Expression expression = expressionParser.parseExpression(expressionString);
-		return expression.getValue(getEvaluationContext(), Object.class);
-	}
-
-	public EvaluationContext getEvaluationContext() {
-		return this.evaluationContext.get();
-	}
-
-	public SpelExpressionParser getExpressionParser() {
-		return expressionParser;
+		return expressionEvaluator.evaluate(expressionString);
 	}
 
 	public ValueProvider getValueProvider() {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/json/ParameterBindingDocumentCodec.java

@@ -24,6 +24,7 @@ import java.util.Collection;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.UUID;
 
 import org.bson.AbstractBsonReader.State;
@@ -39,7 +40,10 @@ import org.bson.Transformer;
 import org.bson.codecs.*;
 import org.bson.codecs.configuration.CodecRegistry;
 import org.bson.json.JsonParseException;
+import org.springframework.data.mapping.model.SpELExpressionEvaluator;
 import org.springframework.data.spel.EvaluationContextProvider;
+import org.springframework.data.spel.ExpressionDependencies;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.lang.Nullable;
 import org.springframework.util.NumberUtils;
@@ -163,7 +167,7 @@ public class ParameterBindingDocumentCodec implements CollectibleCodec<Document>
 	public Document decode(@Nullable String json, Object[] values) {
 
 		return decode(json, new ParameterBindingContext((index) -> values[index], new SpelExpressionParser(),
-				() -> EvaluationContextProvider.DEFAULT.getEvaluationContext(values)));
+				EvaluationContextProvider.DEFAULT.getEvaluationContext(values)));
 	}
 
 	public Document decode(@Nullable String json, ParameterBindingContext bindingContext) {
@@ -176,6 +180,31 @@ public class ParameterBindingDocumentCodec implements CollectibleCodec<Document>
 		return this.decode(reader, DecoderContext.builder().build());
 	}
 
+	/**
+	 * Determine {@link ExpressionDependencies} from Expressions that are nested in the {@code json} content. Returns
+	 * {@link Optional#empty()} if {@code json} is empty or of it does not contain any SpEL expressions.
+	 *
+	 * @param json
+	 * @param expressionParser
+	 * @return merged {@link ExpressionDependencies} object if expressions were found, otherwise
+	 *         {@link ExpressionDependencies#none()}.
+	 * @since 3.1
+	 */
+	public ExpressionDependencies captureExpressionDependencies(@Nullable String json, ValueProvider valueProvider,
+			ExpressionParser expressionParser) {
+
+		if (StringUtils.isEmpty(json)) {
+			return ExpressionDependencies.none();
+		}
+
+		DependencyCapturingExpressionEvaluator expressionEvaluator = new DependencyCapturingExpressionEvaluator(
+				expressionParser);
+		this.decode(new ParameterBindingJsonReader(json, new ParameterBindingContext(valueProvider, expressionEvaluator)),
+				DecoderContext.builder().build());
+
+		return expressionEvaluator.getCapturedDependencies();
+	}
+
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Override
 	public Document decode(final BsonReader reader, final DecoderContext decoderContext) {
@@ -336,17 +365,38 @@ public class ParameterBindingDocumentCodec implements CollectibleCodec<Document>
 		reader.readStartArray();
 		List<Object> list = new ArrayList<>();
 		while (reader.readBsonType() != BsonType.END_OF_DOCUMENT) {
-
-			// Spring Data Customization START
-			Object listValue = readValue(reader, decoderContext);
-			if (listValue instanceof Collection) {
-				list.addAll((Collection) listValue);
-				break;
-			}
-			list.add(listValue);
-			// Spring Data Customization END
+			list.add(readValue(reader, decoderContext));
 		}
 		reader.readEndArray();
 		return list;
 	}
+
+	/**
+	 * @author Christoph Strobl
+	 * @since 3.1
+	 */
+	static class DependencyCapturingExpressionEvaluator implements SpELExpressionEvaluator {
+
+		private static final Object PLACEHOLDER = new Object();
+
+		private final ExpressionParser expressionParser;
+		private final List<ExpressionDependencies> dependencies = new ArrayList<>();
+
+		DependencyCapturingExpressionEvaluator(ExpressionParser expressionParser) {
+			this.expressionParser = expressionParser;
+		}
+
+		@Nullable
+		@Override
+		public <T> T evaluate(String expression) {
+
+			dependencies.add(ExpressionDependencies.discover(expressionParser.parseExpression(expression)));
+			return (T) PLACEHOLDER;
+		}
+
+		ExpressionDependencies getCapturedDependencies() {
+			return ExpressionDependencies.merged(dependencies);
+		}
+
+	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/json/ParameterBindingJsonReader.java

@@ -36,6 +36,7 @@ import org.bson.types.Decimal128;
 import org.bson.types.MaxKey;
 import org.bson.types.MinKey;
 import org.bson.types.ObjectId;
+
 import org.springframework.data.spel.EvaluationContextProvider;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
@@ -227,7 +228,7 @@ public class ParameterBindingJsonReader extends AbstractBsonReader {
 			case REGULAR_EXPRESSION:
 
 				setCurrentBsonType(BsonType.REGULAR_EXPRESSION);
-				currentValue = bindableValueFor(token).getValue().toString();
+				currentValue = bindableValueFor(token).getValue();
 				break;
 			case STRING:
 
@@ -363,8 +364,11 @@ public class ParameterBindingJsonReader extends AbstractBsonReader {
 			return null;
 		}
 
+		boolean isRegularExpression = token.getType().equals(JsonTokenType.REGULAR_EXPRESSION);
+
 		BindableValue bindableValue = new BindableValue();
-		String tokenValue = String.class.cast(token.getValue());
+		String tokenValue = isRegularExpression ? token.getValue(BsonRegularExpression.class).getPattern()
+				: String.class.cast(token.getValue());
 		Matcher matcher = PARAMETER_BINDING_PATTERN.matcher(tokenValue);
 
 		if (token.getType().equals(JsonTokenType.UNQUOTED_STRING)) {
@@ -404,8 +408,6 @@ public class ParameterBindingJsonReader extends AbstractBsonReader {
 
 		String computedValue = tokenValue;
 
-
-
 		Matcher regexMatcher = EXPRESSION_BINDING_PATTERN.matcher(computedValue);
 
 		while (regexMatcher.find()) {
@@ -435,9 +437,15 @@ public class ParameterBindingJsonReader extends AbstractBsonReader {
 			computedValue = computedValue.replace(group, nullSafeToString(getBindableValueForIndex(index)));
 		}
 
-		bindableValue.setValue(computedValue);
-		bindableValue.setType(BsonType.STRING);
+		if (isRegularExpression) {
 
+			bindableValue.setValue(new BsonRegularExpression(computedValue));
+			bindableValue.setType(BsonType.REGULAR_EXPRESSION);
+		} else {
+
+			bindableValue.setValue(computedValue);
+			bindableValue.setType(BsonType.STRING);
+		}
 		return bindableValue;
 	}
 

spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/query/KPropertyPath.kt

@@ -20,8 +20,10 @@ import kotlin.reflect.KProperty1
 
 /**
  * Abstraction of a property path consisting of [KProperty].
+ *
  * @author Tjeu Kayim
  * @author Mark Paluch
+ * @author Yoann de Martino
  * @since 2.2
  */
 class KPropertyPath<T, U>(
@@ -45,7 +47,7 @@ internal fun asString(property: KProperty<*>): String {
  * Builds [KPropertyPath] from Property References.
  * Refer to a field in an embedded/nested document.
  *
- * For example, referring to the field "book.author":
+ * For example, referring to the field "author.name":
  * ```
  * Book::author / Author::name isEqualTo "Herman Melville"
  * ```

spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/query/KPropertyPathExtensions.kt

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2020 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.core.query
+
+import kotlin.reflect.KProperty
+
+/**
+ * Extension for [KProperty] providing an `toPath` function to render a [KProperty] as property path.
+ *
+ * @author Mark Paluch
+ * @since 3.1
+ */
+fun KProperty<*>.toPath(): String = asString(this)

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/buildtimetypeinfo/Address.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2020 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.buildtimetypeinfo;
+
+import org.springframework.util.ObjectUtils;
+
+/**
+ * @author Christoph Strobl
+ * @since 2020/10
+ */
+public class Address {
+
+	String city;
+	String street;
+
+	public Address(String city, String street) {
+		this.city = city;
+		this.street = street;
+	}
+
+	public String getCity() {
+		return city;
+	}
+
+	public String getStreet() {
+		return street;
+	}
+
+	@Override
+	public String toString() {
+		return "Address{" + "city='" + city + '\'' + ", street='" + street + '\'' + '}';
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o)
+			return true;
+		if (o == null || getClass() != o.getClass())
+			return false;
+
+		Address address = (Address) o;
+
+		if (!ObjectUtils.nullSafeEquals(city, address.city)) {
+			return false;
+		}
+		return ObjectUtils.nullSafeEquals(street, address.street);
+	}
+
+	@Override
+	public int hashCode() {
+		int result = ObjectUtils.nullSafeHashCode(city);
+		result = 31 * result + ObjectUtils.nullSafeHashCode(street);
+		return result;
+	}
+}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/buildtimetypeinfo/AddressTypeInformation.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2020 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.buildtimetypeinfo;
+
+import org.springframework.data.mapping.model.DomainTypeConstructor;
+import org.springframework.data.mapping.model.DomainTypeInformation;
+import org.springframework.data.mapping.model.Field;
+
+/**
+ * @author Christoph Strobl
+ * @since 2020/10
+ */
+public class AddressTypeInformation extends DomainTypeInformation<Address> {
+
+	private static final AddressTypeInformation INSTANCE = new AddressTypeInformation();
+
+	private AddressTypeInformation() {
+
+		super(Address.class);
+
+		// CONSTRUCTOR
+		setConstructor(computePreferredConstructor());
+
+		// FIELDS
+		addField(Field.<Address> string("city").getter(Address::getCity));
+		addField(Field.<Address> string("street").getter(Address::getStreet));
+	}
+
+	public static AddressTypeInformation instance() {
+		return INSTANCE;
+	}
+
+	private DomainTypeConstructor<Address> computePreferredConstructor() {
+		return DomainTypeConstructor.<Address> builder().args("city", "street")
+				.newInstanceFunction(args -> new Address((String) args[0], (String) args[1]));
+	}
+}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/buildtimetypeinfo/Person.java

@@ -0,0 +1,139 @@
+/*
+ * Copyright 2020 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.buildtimetypeinfo;
+
+import java.util.List;
+
+import org.springframework.util.ObjectUtils;
+
+/**
+ * @author Christoph Strobl
+ * @since 2020/10
+ */
+public class Person {
+
+	private long id;
+	private String firstname, lastname; // TODO: we need a persistence constructor to resolve this here.
+	private int age;
+	private Address address;
+	private List<String> nicknames;
+
+	public Person(String firstname, String lastname) {
+		this.firstname = firstname;
+		this.lastname = lastname;
+	}
+
+	private Person(long id, String firstname, String lastname, int age, Address address, List<String> nicknames) {
+		this.id = id;
+		this.firstname = firstname;
+		this.lastname = lastname;
+		this.age = age;
+		this.address = address;
+		this.nicknames = nicknames;
+	}
+
+	public String getFirstname() {
+		return firstname;
+	}
+
+	public String getLastname() {
+		return lastname;
+	}
+
+	public int getAge() {
+		return age;
+	}
+
+	public void setAge(int age) {
+		this.age = age;
+	}
+
+	public long getId() {
+		return id;
+	}
+
+	public Person withId(long id) {
+
+		return new Person(id, firstname, lastname, age, address, nicknames);
+
+	}
+
+	public Address getAddress() {
+		return address;
+	}
+
+	public void setAddress(Address address) {
+		this.address = address;
+	}
+
+	public List<String> getNicknames() {
+		return nicknames;
+	}
+
+	public void setNicknames(List<String> nicknames) {
+		this.nicknames = nicknames;
+	}
+
+	public void setFirstname(String firstname) {
+		this.firstname = firstname;
+	}
+
+	public void setLastname(String lastname) {
+		this.lastname = lastname;
+	}
+
+	@Override
+	public String toString() {
+		return "Person{" + "id=" + id + ", firstname='" + firstname + '\'' + ", lastname='" + lastname + '\'' + ", age="
+				+ age + ", address=" + address + ", nicknames=" + nicknames + '}';
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o)
+			return true;
+		if (o == null || getClass() != o.getClass())
+			return false;
+
+		Person person = (Person) o;
+
+		if (id != person.id)
+			return false;
+		if (age != person.age)
+			return false;
+		if (!ObjectUtils.nullSafeEquals(firstname, person.firstname)) {
+			return false;
+		}
+		if (!ObjectUtils.nullSafeEquals(lastname, person.lastname)) {
+			return false;
+		}
+		if (!ObjectUtils.nullSafeEquals(address, person.address)) {
+			return false;
+		}
+		return ObjectUtils.nullSafeEquals(nicknames, person.nicknames);
+	}
+
+	@Override
+	public int hashCode() {
+		int result = (int) (id ^ (id >>> 32));
+		result = 31 * result + ObjectUtils.nullSafeHashCode(firstname);
+		result = 31 * result + ObjectUtils.nullSafeHashCode(lastname);
+		result = 31 * result + age;
+		result = 31 * result + ObjectUtils.nullSafeHashCode(address);
+		result = 31 * result + ObjectUtils.nullSafeHashCode(nicknames);
+		return result;
+	}
+}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/buildtimetypeinfo/PersonTypeInformation.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2020 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.buildtimetypeinfo;
+
+import java.lang.annotation.Annotation;
+import java.util.List;
+
+import org.springframework.data.mapping.model.DomainTypeConstructor;
+import org.springframework.data.mapping.model.DomainTypeInformation;
+import org.springframework.data.mapping.model.Field;
+import org.springframework.data.mapping.model.ListTypeInformation;
+import org.springframework.data.mapping.model.StringTypeInformation;
+import org.springframework.data.mongodb.core.mapping.Document;
+import org.springframework.data.mongodb.core.mapping.FieldType;
+
+/**
+ * @author Christoph Strobl
+ * @since 2020/10
+ */
+public class PersonTypeInformation extends DomainTypeInformation<Person> {
+
+	private static final PersonTypeInformation INSTANCE = new PersonTypeInformation();
+
+	private PersonTypeInformation() {
+
+		super(Person.class);
+
+		// CONSTRUCTOR
+		setConstructor(computePreferredConstructor());
+
+		// ANNOTATIONS
+		addAnnotation(computeAtDocumentAnnotation());
+
+		// FIELDS
+		addField(
+				Field.<Person> int64("id").annotatedWithAtId().getter(Person::getId).wither((bean, id) -> bean.withId(id)));
+		addField(Field.<Person> string("firstname").getter(Person::getFirstname).annotation(atFieldOnFirstname()));
+		addField(Field.<Person> string("lastname").getter(Person::getLastname));
+		addField(Field.<Person> int32("age").getter(Person::getAge).setter(Person::setAge));
+		addField(Field.<Person, Address> type("address", AddressTypeInformation.instance()).getter(Person::getAddress)
+				.setter(Person::setAddress));
+		addField(Field.<Person, List<String>> type("nicknames", new ListTypeInformation<>(StringTypeInformation.instance()))
+				.getter(Person::getNicknames).setter(Person::setNicknames));
+
+	}
+
+	public static PersonTypeInformation instance() {
+		return INSTANCE;
+	}
+
+	private DomainTypeConstructor<Person> computePreferredConstructor() {
+		return DomainTypeConstructor.<Person> builder().args("firstname", "lastname")
+				.newInstanceFunction((args) -> new Person((String) args[0], (String) args[1]));
+	}
+
+	private Document computeAtDocumentAnnotation() {
+		return new Document() {
+			@Override
+			public Class<? extends Annotation> annotationType() {
+				return Document.class;
+			}
+
+			@Override
+			public String value() {
+				return collection();
+			}
+
+			@Override
+			public String collection() {
+				return "star-wars";
+			}
+
+			@Override
+			public String language() {
+				return "";
+			}
+
+			@Override
+			public String collation() {
+				return "";
+			}
+		};
+	}
+
+	private Annotation atFieldOnFirstname() {
+
+		return new org.springframework.data.mongodb.core.mapping.Field() {
+
+			@Override
+			public Class<? extends Annotation> annotationType() {
+				return org.springframework.data.mongodb.core.mapping.Field.class;
+			}
+
+			@Override
+			public String value() {
+				return "first-name";
+			}
+
+			@Override
+			public String name() {
+				return value();
+			}
+
+			@Override
+			public int order() {
+				return 0;
+			}
+
+			@Override
+			public FieldType targetType() {
+				return FieldType.IMPLICIT;
+			}
+		};
+	}
+}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/config/AuditingViaJavaConfigRepositoriesTests.java

@@ -21,6 +21,7 @@ import static org.mockito.Mockito.*;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Function;
@@ -34,12 +35,16 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan.Filter;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.FilterType;
+import org.springframework.core.ResolvableType;
 import org.springframework.data.annotation.Version;
 import org.springframework.data.domain.AuditorAware;
+import org.springframework.data.mapping.callback.EntityCallback;
 import org.springframework.data.mongodb.core.AuditablePerson;
 import org.springframework.data.mongodb.core.MongoOperations;
 import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
+import org.springframework.data.mongodb.core.mapping.event.AuditingEntityCallback;
+import org.springframework.data.mongodb.core.mapping.event.ReactiveAuditingEntityCallback;
 import org.springframework.data.mongodb.repository.MongoRepository;
 import org.springframework.data.mongodb.repository.config.EnableMongoRepositories;
 import org.springframework.data.mongodb.test.util.Client;
@@ -48,6 +53,7 @@ import org.springframework.data.mongodb.test.util.MongoTestUtils;
 import org.springframework.stereotype.Repository;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
+import org.springframework.test.util.ReflectionTestUtils;
 
 import com.mongodb.client.MongoClient;
 
@@ -60,7 +66,7 @@ import com.mongodb.client.MongoClient;
  */
 @ExtendWith({ MongoClientExtension.class, SpringExtension.class })
 @ContextConfiguration
-public class AuditingViaJavaConfigRepositoriesTests {
+class AuditingViaJavaConfigRepositoriesTests {
 
 	static @Client MongoClient mongoClient;
 
@@ -79,6 +85,7 @@ public class AuditingViaJavaConfigRepositoriesTests {
 
 		@Override
 		protected String getDatabaseName() {
+
 			return "database";
 		}
 
@@ -101,13 +108,13 @@ public class AuditingViaJavaConfigRepositoriesTests {
 	}
 
 	@BeforeEach
-	public void setup() {
+	void setup() {
 		auditablePersonRepository.deleteAll();
 		this.auditor = auditablePersonRepository.save(new AuditablePerson("auditor"));
 	}
 
 	@Test // DATAMONGO-792, DATAMONGO-883
-	public void basicAuditing() {
+	void basicAuditing() {
 
 		doReturn(Optional.of(this.auditor)).when(this.auditorAware).getCurrentAuditor();
 
@@ -122,18 +129,18 @@ public class AuditingViaJavaConfigRepositoriesTests {
 
 	@Test // DATAMONGO-843
 	@SuppressWarnings("resource")
-	public void auditingUsesFallbackMappingContextIfNoneConfiguredWithRepositories() {
+	void auditingUsesFallbackMappingContextIfNoneConfiguredWithRepositories() {
 		new AnnotationConfigApplicationContext(SimpleConfigWithRepositories.class);
 	}
 
 	@Test // DATAMONGO-843
 	@SuppressWarnings("resource")
-	public void auditingUsesFallbackMappingContextIfNoneConfigured() {
+	void auditingUsesFallbackMappingContextIfNoneConfigured() {
 		new AnnotationConfigApplicationContext(SimpleConfig.class);
 	}
 
 	@Test // DATAMONGO-2139
-	public void auditingWorksForVersionedEntityWithWrapperVersion() {
+	void auditingWorksForVersionedEntityWithWrapperVersion() {
 
 		verifyAuditingViaVersionProperty(new VersionedAuditablePerson(), //
 				it -> it.version, //
@@ -143,7 +150,7 @@ public class AuditingViaJavaConfigRepositoriesTests {
 	}
 
 	@Test // DATAMONGO-2179
-	public void auditingWorksForVersionedEntityBatchWithWrapperVersion() {
+	void auditingWorksForVersionedEntityBatchWithWrapperVersion() {
 
 		verifyAuditingViaVersionProperty(new VersionedAuditablePerson(), //
 				it -> it.version, //
@@ -153,7 +160,7 @@ public class AuditingViaJavaConfigRepositoriesTests {
 	}
 
 	@Test // DATAMONGO-2139
-	public void auditingWorksForVersionedEntityWithSimpleVersion() {
+	void auditingWorksForVersionedEntityWithSimpleVersion() {
 
 		verifyAuditingViaVersionProperty(new SimpleVersionedAuditablePerson(), //
 				it -> it.version, //
@@ -163,7 +170,7 @@ public class AuditingViaJavaConfigRepositoriesTests {
 	}
 
 	@Test // DATAMONGO-2139
-	public void auditingWorksForVersionedEntityWithWrapperVersionOnTemplate() {
+	void auditingWorksForVersionedEntityWithWrapperVersionOnTemplate() {
 
 		verifyAuditingViaVersionProperty(new VersionedAuditablePerson(), //
 				it -> it.version, //
@@ -173,7 +180,7 @@ public class AuditingViaJavaConfigRepositoriesTests {
 	}
 
 	@Test // DATAMONGO-2139
-	public void auditingWorksForVersionedEntityWithSimpleVersionOnTemplate() {
+	void auditingWorksForVersionedEntityWithSimpleVersionOnTemplate() {
 
 		verifyAuditingViaVersionProperty(new SimpleVersionedAuditablePerson(), //
 				it -> it.version, //
@@ -182,6 +189,19 @@ public class AuditingViaJavaConfigRepositoriesTests {
 				0L, 1L, 2L);
 	}
 
+	@Test // DATAMONGO-2586
+	void auditingShouldOnlyRegisterImperativeAuditingCallback() {
+
+		Object callbacks = ReflectionTestUtils.getField(operations, "entityCallbacks");
+		Object callbackDiscoverer = ReflectionTestUtils.getField(callbacks, "callbackDiscoverer");
+		List<EntityCallback<?>> actualCallbacks = ReflectionTestUtils.invokeMethod(callbackDiscoverer, "getEntityCallbacks",
+				AuditablePerson.class, ResolvableType.forClass(EntityCallback.class));
+
+		assertThat(actualCallbacks) //
+				.hasAtLeastOneElementOfType(AuditingEntityCallback.class) //
+				.doesNotHaveAnyElementsOfTypes(ReactiveAuditingEntityCallback.class);
+	}
+
 	private <T extends AuditablePerson> void verifyAuditingViaVersionProperty(T instance,
 			Function<T, Object> versionExtractor, Function<T, Object> createdDateExtractor, Function<T, T> persister,
 			Object... expectedValues) {

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/config/ReactiveAuditingTests.java

@@ -16,27 +16,33 @@
 package org.springframework.data.mongodb.config;
 
 import static org.assertj.core.api.Assertions.*;
-import static org.mockito.Mockito.*;
 
+import org.springframework.core.ResolvableType;
+import org.springframework.data.mapping.callback.EntityCallback;
+import org.springframework.data.mongodb.core.mapping.event.AuditingEntityCallback;
+import org.springframework.data.mongodb.core.mapping.event.ReactiveAuditingEntityCallback;
+import org.springframework.test.util.ReflectionTestUtils;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan.Filter;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.FilterType;
 import org.springframework.data.annotation.Version;
-import org.springframework.data.domain.AuditorAware;
+import org.springframework.data.domain.ReactiveAuditorAware;
 import org.springframework.data.mongodb.core.AuditablePerson;
 import org.springframework.data.mongodb.core.ReactiveMongoOperations;
 import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
@@ -57,17 +63,16 @@ import com.mongodb.reactivestreams.client.MongoClient;
  */
 @ExtendWith({ MongoClientExtension.class, SpringExtension.class })
 @ContextConfiguration
-public class ReactiveAuditingTests {
+class ReactiveAuditingTests {
 
 	static @Client MongoClient mongoClient;
 
 	@Autowired ReactiveAuditablePersonRepository auditablePersonRepository;
-	@Autowired AuditorAware<AuditablePerson> auditorAware;
 	@Autowired MongoMappingContext context;
 	@Autowired ReactiveMongoOperations operations;
 
 	@Configuration
-	@EnableMongoAuditing(auditorAwareRef = "auditorProvider")
+	@EnableReactiveMongoAuditing
 	@EnableReactiveMongoRepositories(basePackageClasses = ReactiveAuditingTests.class, considerNestedRepositories = true,
 			includeFilters = @Filter(type = FilterType.ASSIGNABLE_TYPE, classes = ReactiveAuditablePersonRepository.class))
 	static class Config extends AbstractReactiveMongoConfiguration {
@@ -89,14 +94,17 @@ public class ReactiveAuditingTests {
 		}
 
 		@Bean
-		@SuppressWarnings("unchecked")
-		public AuditorAware<AuditablePerson> auditorProvider() {
-			return mock(AuditorAware.class);
+		public ReactiveAuditorAware<AuditablePerson> auditorProvider() {
+
+			AuditablePerson person = new AuditablePerson("some-person");
+			person.setId("foo");
+
+			return () -> Mono.just(person);
 		}
 	}
 
-	@Test // DATAMONGO-2139, DATAMONGO-2150
-	public void auditingWorksForVersionedEntityWithWrapperVersion() {
+	@Test // DATAMONGO-2139, DATAMONGO-2150, DATAMONGO-2586
+	void auditingWorksForVersionedEntityWithWrapperVersion() {
 
 		verifyAuditingViaVersionProperty(new VersionedAuditablePerson(), //
 				it -> it.version, //
@@ -106,7 +114,7 @@ public class ReactiveAuditingTests {
 	}
 
 	@Test // DATAMONGO-2179
-	public void auditingWorksForVersionedEntityBatchWithWrapperVersion() {
+	void auditingWorksForVersionedEntityBatchWithWrapperVersion() {
 
 		verifyAuditingViaVersionProperty(new VersionedAuditablePerson(), //
 				it -> it.version, //
@@ -115,8 +123,8 @@ public class ReactiveAuditingTests {
 				null, 0L, 1L);
 	}
 
-	@Test // DATAMONGO-2139, DATAMONGO-2150
-	public void auditingWorksForVersionedEntityWithSimpleVersion() {
+	@Test // DATAMONGO-2139, DATAMONGO-2150, DATAMONGO-2586
+	void auditingWorksForVersionedEntityWithSimpleVersion() {
 
 		verifyAuditingViaVersionProperty(new SimpleVersionedAuditablePerson(), //
 				it -> it.version, //
@@ -125,8 +133,8 @@ public class ReactiveAuditingTests {
 				0L, 1L, 2L);
 	}
 
-	@Test // DATAMONGO-2139, DATAMONGO-2150
-	public void auditingWorksForVersionedEntityWithWrapperVersionOnTemplate() {
+	@Test // DATAMONGO-2139, DATAMONGO-2150, DATAMONGO-2586
+	void auditingWorksForVersionedEntityWithWrapperVersionOnTemplate() {
 
 		verifyAuditingViaVersionProperty(new VersionedAuditablePerson(), //
 				it -> it.version, //
@@ -135,8 +143,8 @@ public class ReactiveAuditingTests {
 				null, 0L, 1L);
 	}
 
-	@Test // DATAMONGO-2139, DATAMONGO-2150
-	public void auditingWorksForVersionedEntityWithSimpleVersionOnTemplate() {
+	@Test // DATAMONGO-2139, DATAMONGO-2150, DATAMONGO-2586
+	void auditingWorksForVersionedEntityWithSimpleVersionOnTemplate() {
 		verifyAuditingViaVersionProperty(new SimpleVersionedAuditablePerson(), //
 				it -> it.version, //
 				AuditablePerson::getCreatedAt, //
@@ -144,6 +152,19 @@ public class ReactiveAuditingTests {
 				0L, 1L, 2L);
 	}
 
+	@Test // DATAMONGO-2586
+	void auditingShouldOnlyRegisterReactiveAuditingCallback() {
+
+		Object callbacks = ReflectionTestUtils.getField(operations, "entityCallbacks");
+		Object callbackDiscoverer = ReflectionTestUtils.getField(callbacks, "callbackDiscoverer");
+		List<EntityCallback<?>> actualCallbacks = ReflectionTestUtils.invokeMethod(callbackDiscoverer, "getEntityCallbacks",
+				AuditablePerson.class, ResolvableType.forClass(EntityCallback.class));
+
+		assertThat(actualCallbacks) //
+				.hasAtLeastOneElementOfType(ReactiveAuditingEntityCallback.class) //
+				.doesNotHaveAnyElementsOfTypes(AuditingEntityCallback.class);
+	}
+
 	private <T extends AuditablePerson> void verifyAuditingViaVersionProperty(T instance,
 			Function<T, Object> versionExtractor, Function<T, Object> createdDateExtractor, Function<T, Mono<T>> persister,
 			Object... expectedValues) {

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ClientSessionTests.java

@@ -28,6 +28,7 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
+import org.springframework.data.mongodb.core.mapping.DBRef;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.test.util.EnableIfMongoServerVersion;
 import org.springframework.data.mongodb.test.util.EnableIfReplicaSetAvailable;
@@ -41,23 +42,26 @@ import com.mongodb.client.ClientSession;
 import com.mongodb.client.MongoClient;
 
 /**
+ * Integration tests for {@link ClientSession} through {@link MongoTemplate#withSession(ClientSession)}.
+ *
  * @author Christoph Strobl
  * @author Mark Paluch
  */
 @ExtendWith({ MongoClientExtension.class })
 @EnableIfReplicaSetAvailable
 @EnableIfMongoServerVersion(isGreaterThanEqual = "4.0")
-public class ClientSessionTests {
+class ClientSessionTests {
 
 	private static final String DB_NAME = "client-session-tests";
 	private static final String COLLECTION_NAME = "test";
+	private static final String REF_COLLECTION_NAME = "test-with-ref";
 
-	static @ReplSetClient MongoClient mongoClient;
+	private static @ReplSetClient MongoClient mongoClient;
 
-	MongoTemplate template;
+	private MongoTemplate template;
 
 	@BeforeEach
-	public void setUp() {
+	void setUp() {
 
 		MongoTestUtils.createOrReplaceCollection(DB_NAME, COLLECTION_NAME, mongoClient);
 
@@ -66,7 +70,7 @@ public class ClientSessionTests {
 	}
 
 	@Test // DATAMONGO-1880
-	public void shouldApplyClientSession() {
+	void shouldApplyClientSession() {
 
 		ClientSession session = mongoClient.startSession(ClientSessionOptions.builder().causallyConsistent(true).build());
 
@@ -83,7 +87,7 @@ public class ClientSessionTests {
 	}
 
 	@Test // DATAMONGO-2241
-	public void shouldReuseConfiguredInfrastructure() {
+	void shouldReuseConfiguredInfrastructure() {
 
 		ClientSession session = mongoClient.startSession(ClientSessionOptions.builder().causallyConsistent(true).build());
 
@@ -99,7 +103,7 @@ public class ClientSessionTests {
 	}
 
 	@Test // DATAMONGO-1920
-	public void withCommittedTransaction() {
+	void withCommittedTransaction() {
 
 		ClientSession session = mongoClient.startSession(ClientSessionOptions.builder().causallyConsistent(true).build());
 
@@ -124,7 +128,7 @@ public class ClientSessionTests {
 	}
 
 	@Test // DATAMONGO-1920
-	public void withAbortedTransaction() {
+	void withAbortedTransaction() {
 
 		ClientSession session = mongoClient.startSession(ClientSessionOptions.builder().causallyConsistent(true).build());
 
@@ -148,6 +152,31 @@ public class ClientSessionTests {
 		assertThat(template.exists(query(where("id").is(saved.getId())), SomeDoc.class)).isFalse();
 	}
 
+	@Test // DATAMONGO-2490
+	void shouldBeAbleToReadDbRefDuringTransaction() {
+
+		SomeDoc ref = new SomeDoc("ref-1", "da value");
+		WithDbRef source = new WithDbRef("source-1", "da source", ref);
+
+		ClientSession session = mongoClient.startSession(ClientSessionOptions.builder().causallyConsistent(true).build());
+
+		assertThat(session.getOperationTime()).isNull();
+
+		session.startTransaction();
+
+		WithDbRef saved = template.withSession(() -> session).execute(action -> {
+
+			template.save(ref);
+			template.save(source);
+
+			return template.findOne(query(where("id").is(source.id)), WithDbRef.class);
+		});
+
+		assertThat(saved.getSomeDocRef()).isEqualTo(ref);
+
+		session.abortTransaction();
+	}
+
 	@Data
 	@AllArgsConstructor
 	@org.springframework.data.mongodb.core.mapping.Document(COLLECTION_NAME)
@@ -157,4 +186,14 @@ public class ClientSessionTests {
 		String value;
 	}
 
+	@Data
+	@AllArgsConstructor
+	@org.springframework.data.mongodb.core.mapping.Document(REF_COLLECTION_NAME)
+	static class WithDbRef {
+
+		@Id String id;
+		String value;
+		@DBRef SomeDoc someDocRef;
+	}
+
 }

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsIntegrationTests.java

@@ -27,6 +27,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.springframework.dao.DuplicateKeyException;
+import org.springframework.data.mongodb.BulkOperationException;
 import org.springframework.data.mongodb.core.BulkOperations.BulkMode;
 import org.springframework.data.mongodb.core.DefaultBulkOperations.BulkOperationContext;
 import org.springframework.data.mongodb.core.convert.QueryMapper;
@@ -91,13 +92,13 @@ public class DefaultBulkOperationsIntegrationTests {
 		assertThat(createBulkOps(BulkMode.ORDERED).insert(documents).execute().getInsertedCount()).isEqualTo(2);
 	}
 
-	@Test // DATAMONGO-934
+	@Test // DATAMONGO-934, DATAMONGO-2285
 	public void insertOrderedFails() {
 
 		List<BaseDoc> documents = Arrays.asList(newDoc("1"), newDoc("1"), newDoc("2"));
 
 		assertThatThrownBy(() -> createBulkOps(BulkMode.ORDERED).insert(documents).execute()) //
-				.isInstanceOf(DuplicateKeyException.class) //
+				.isInstanceOf(BulkOperationException.class) //
 				.hasCauseInstanceOf(MongoBulkWriteException.class) //
 				.extracting(Throwable::getCause) //
 				.satisfies(it -> {
@@ -117,13 +118,13 @@ public class DefaultBulkOperationsIntegrationTests {
 		assertThat(createBulkOps(BulkMode.UNORDERED).insert(documents).execute().getInsertedCount()).isEqualTo(2);
 	}
 
-	@Test // DATAMONGO-934
+	@Test // DATAMONGO-934, DATAMONGO-2285
 	public void insertUnOrderedContinuesOnError() {
 
 		List<BaseDoc> documents = Arrays.asList(newDoc("1"), newDoc("1"), newDoc("2"));
 
 		assertThatThrownBy(() -> createBulkOps(BulkMode.UNORDERED).insert(documents).execute()) //
-				.isInstanceOf(DuplicateKeyException.class) //
+				.isInstanceOf(BulkOperationException.class) //
 				.hasCauseInstanceOf(MongoBulkWriteException.class) //
 				.extracting(Throwable::getCause) //
 				.satisfies(it -> {

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsUnitTests.java

@@ -24,6 +24,7 @@ import static org.mockito.Mockito.eq;
 import static org.springframework.data.mongodb.core.query.Criteria.*;
 import static org.springframework.data.mongodb.core.query.Query.*;
 
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 
@@ -40,9 +41,11 @@ import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.dao.DataAccessException;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.mapping.callback.EntityCallbacks;
+import org.springframework.data.mongodb.BulkOperationException;
 import org.springframework.data.mongodb.MongoDatabaseFactory;
 import org.springframework.data.mongodb.core.BulkOperations.BulkMode;
 import org.springframework.data.mongodb.core.DefaultBulkOperations.BulkOperationContext;
@@ -64,9 +67,13 @@ import org.springframework.data.mongodb.core.query.Collation;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Update;
 
+import com.mongodb.MongoBulkWriteException;
 import com.mongodb.MongoWriteException;
+import com.mongodb.ServerAddress;
 import com.mongodb.WriteConcern;
 import com.mongodb.WriteError;
+import com.mongodb.bulk.BulkWriteError;
+import com.mongodb.bulk.WriteConcernError;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoDatabase;
 import com.mongodb.client.model.BulkWriteOptions;
@@ -85,6 +92,7 @@ import com.mongodb.client.model.WriteModel;
  * @author Minsu Kim
  * @author Jens Schauder
  * @author Roman Puchkovskiy
+ * @author Jacob Botuck
  */
 @ExtendWith(MockitoExtension.class)
 class DefaultBulkOperationsUnitTests {
@@ -367,6 +375,29 @@ class DefaultBulkOperationsUnitTests {
 				.isEqualTo(new Document("$set", new Document("items.$.documents.0.the_file_id", "file-id")));
 	}
 
+	@Test // DATAMONGO-2285
+	public void translateMongoBulkOperationExceptionWithWriteConcernError() {
+
+		when(collection.bulkWrite(anyList(), any(BulkWriteOptions.class))).thenThrow(new MongoBulkWriteException(null,
+				Collections.emptyList(),
+				new WriteConcernError(42, "codename", "writeconcern error happened", new BsonDocument()), new ServerAddress()));
+
+		assertThatExceptionOfType(DataIntegrityViolationException.class)
+				.isThrownBy(() -> ops.insert(new SomeDomainType()).execute());
+
+	}
+
+	@Test // DATAMONGO-2285
+	public void translateMongoBulkOperationExceptionWithoutWriteConcernError() {
+
+		when(collection.bulkWrite(anyList(), any(BulkWriteOptions.class))).thenThrow(new MongoBulkWriteException(null,
+				Collections.singletonList(new BulkWriteError(42, "a write error happened", new BsonDocument(), 49)), null,
+				new ServerAddress()));
+
+		assertThatExceptionOfType(BulkOperationException.class)
+				.isThrownBy(() -> ops.insert(new SomeDomainType()).execute());
+	}
+
 	static class OrderTest {
 
 		String id;