Add support for fluent QueryResultConverter.

Closes: #4949

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

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2025 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;
+
+import org.bson.Document;
+
+enum EntityResultConverter implements QueryResultConverter<Object, Object> {
+
+	INSTANCE;
+
+	@Override
+	public Object mapDocument(Document document, ConversionResultSupplier<Object> reader) {
+		return reader.get();
+	}
+
+	@Override
+	public <V> QueryResultConverter<Object, V> andThen(QueryResultConverter<? super Object, ? extends V> after) {
+		return (QueryResultConverter) after;
+	}
+}

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

@@ -19,6 +19,7 @@ import java.util.stream.Stream;
 
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.AggregationResults;
+import org.springframework.lang.Contract;
 
 /**
  * {@link ExecutableAggregationOperation} allows creation and execution of MongoDB aggregation operations in a fluent
@@ -45,7 +46,7 @@ public interface ExecutableAggregationOperation {
 	/**
 	 * Start creating an aggregation operation that returns results mapped to the given domain type. <br />
 	 * Use {@link org.springframework.data.mongodb.core.aggregation.TypedAggregation} to specify a potentially different
-	 * input type for he aggregation.
+	 * input type for the aggregation.
 	 *
 	 * @param domainType must not be {@literal null}.
 	 * @return new instance of {@link ExecutableAggregation}.
@@ -76,10 +77,23 @@ public interface ExecutableAggregationOperation {
 	 * Trigger execution by calling one of the terminating methods.
 	 *
 	 * @author Christoph Strobl
+	 * @author Mark Paluch
 	 * @since 2.0
 	 */
 	interface TerminatingAggregation<T> {
 
+		/**
+		 * Map the query result to a different type using {@link QueryResultConverter}.
+		 *
+		 * @param <R> {@link Class type} of the result.
+		 * @param converter the converter, must not be {@literal null}.
+		 * @return new instance of {@link TerminatingAggregation}.
+		 * @throws IllegalArgumentException if {@link QueryResultConverter converter} is {@literal null}.
+		 * @since x.y
+		 */
+		@Contract("_ -> new")
+		<R> TerminatingAggregation<R> map(QueryResultConverter<? super T, ? extends R> converter);
+
 		/**
 		 * Apply pipeline operations as specified and get all matching elements.
 		 *

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

@@ -44,25 +44,28 @@ class ExecutableAggregationOperationSupport implements ExecutableAggregationOper
 
 		Assert.notNull(domainType, "DomainType must not be null");
 
-		return new ExecutableAggregationSupport<>(template, domainType, null, null);
+		return new ExecutableAggregationSupport<>(template, domainType, QueryResultConverter.entity(), null, null);
 	}
 
 	/**
 	 * @author Christoph Strobl
 	 * @since 2.0
 	 */
-	static class ExecutableAggregationSupport<T>
+	static class ExecutableAggregationSupport<S, T>
 			implements AggregationWithAggregation<T>, ExecutableAggregation<T>, TerminatingAggregation<T> {
 
 		private final MongoTemplate template;
-		private final Class<T> domainType;
+		private final Class<S> domainType;
+		private final QueryResultConverter<? super S, ? extends T> resultConverter;
 		private final @Nullable Aggregation aggregation;
 		private final @Nullable String collection;
 
-		public ExecutableAggregationSupport(MongoTemplate template, Class<T> domainType, @Nullable Aggregation aggregation,
+		public ExecutableAggregationSupport(MongoTemplate template, Class<S> domainType,
+				QueryResultConverter<? super S, ? extends T> resultConverter, @Nullable Aggregation aggregation,
 				@Nullable String collection) {
 			this.template = template;
 			this.domainType = domainType;
+			this.resultConverter = resultConverter;
 			this.aggregation = aggregation;
 			this.collection = collection;
 		}
@@ -72,7 +75,7 @@ class ExecutableAggregationOperationSupport implements ExecutableAggregationOper
 
 			Assert.hasText(collection, "Collection must not be null nor empty");
 
-			return new ExecutableAggregationSupport<>(template, domainType, aggregation, collection);
+			return new ExecutableAggregationSupport<>(template, domainType, resultConverter, aggregation, collection);
 		}
 
 		@Override
@@ -80,21 +83,30 @@ class ExecutableAggregationOperationSupport implements ExecutableAggregationOper
 
 			Assert.notNull(aggregation, "Aggregation must not be null");
 
-			return new ExecutableAggregationSupport<>(template, domainType, aggregation, collection);
+			return new ExecutableAggregationSupport<>(template, domainType, resultConverter, aggregation, collection);
+		}
+
+		@Override
+		public <R> TerminatingAggregation<R> map(QueryResultConverter<? super T, ? extends R> converter) {
+
+			Assert.notNull(converter, "QueryResultConverter must not be null");
+
+			return new ExecutableAggregationSupport<>(template, domainType, this.resultConverter.andThen(converter),
+					aggregation, collection);
 		}
 
 		@Override
 		public AggregationResults<T> all() {
 
 			Assert.notNull(aggregation, "Aggregation must be set first");
-			return template.aggregate(aggregation, getCollectionName(aggregation), domainType);
+			return template.doAggregate(aggregation, getCollectionName(aggregation), domainType, resultConverter);
 		}
 
 		@Override
 		public Stream<T> stream() {
 
 			Assert.notNull(aggregation, "Aggregation must be set first");
-			return template.aggregateStream(aggregation, getCollectionName(aggregation), domainType);
+			return template.doAggregateStream(aggregation, getCollectionName(aggregation), domainType, resultConverter, null);
 		}
 
 		private String getCollectionName(@Nullable Aggregation aggregation) {

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

@@ -28,6 +28,7 @@ import org.springframework.data.geo.GeoResults;
 import org.springframework.data.mongodb.core.query.CriteriaDefinition;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.lang.Contract;
 
 import com.mongodb.client.MongoCollection;
 
@@ -71,9 +72,33 @@ public interface ExecutableFindOperation {
 	 * Trigger find execution by calling one of the terminating methods.
 	 *
 	 * @author Christoph Strobl
+	 * @author Mark Paluch
 	 * @since 2.0
 	 */
-	interface TerminatingFind<T> {
+	interface TerminatingFind<T> extends TerminatingResults<T>, TerminatingProjection {
+
+	}
+
+	/**
+	 * Trigger find execution by calling one of the terminating methods.
+	 *
+	 * @author Christoph Strobl
+	 * @author Mark Paluch
+	 * @since x.y
+	 */
+	interface TerminatingResults<T> {
+
+		/**
+		 * Map the query result to a different type using {@link QueryResultConverter}.
+		 *
+		 * @param <R> {@link Class type} of the result.
+		 * @param converter the converter, must not be {@literal null}.
+		 * @return new instance of {@link TerminatingResults}.
+		 * @throws IllegalArgumentException if {@link QueryResultConverter converter} is {@literal null}.
+		 * @since x.y
+		 */
+		@Contract("_ -> new")
+		<R> TerminatingResults<R> map(QueryResultConverter<? super T, ? extends R> converter);
 
 		/**
 		 * Get exactly zero or one result.
@@ -142,6 +167,16 @@ public interface ExecutableFindOperation {
 		 */
 		Window<T> scroll(ScrollPosition scrollPosition);
 
+	}
+
+	/**
+	 * Trigger find execution by calling one of the terminating methods.
+	 *
+	 * @author Christoph Strobl
+	 * @since x.y
+	 */
+	interface TerminatingProjection {
+
 		/**
 		 * Get the number of matching elements. <br />
 		 * This method uses an
@@ -160,16 +195,30 @@ public interface ExecutableFindOperation {
 		 * @return {@literal true} if at least one matching element exists.
 		 */
 		boolean exists();
+
 	}
 
 	/**
-	 * Trigger geonear execution by calling one of the terminating methods.
+	 * Trigger {@code geoNear} execution by calling one of the terminating methods.
 	 *
 	 * @author Christoph Strobl
+	 * @author Mark Paluch
 	 * @since 2.0
 	 */
 	interface TerminatingFindNear<T> {
 
+		/**
+		 * Map the query result to a different type using {@link QueryResultConverter}.
+		 *
+		 * @param <R> {@link Class type} of the result.
+		 * @param converter the converter, must not be {@literal null}.
+		 * @return new instance of {@link TerminatingFindNear}.
+		 * @throws IllegalArgumentException if {@link QueryResultConverter converter} is {@literal null}.
+		 * @since x.y
+		 */
+		@Contract("_ -> new")
+		<R> TerminatingFindNear<R> map(QueryResultConverter<? super T, ? extends R> converter);
+
 		/**
 		 * Find all matching elements and return them as {@link org.springframework.data.geo.GeoResult}.
 		 *

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

@@ -23,6 +23,7 @@ import org.jspecify.annotations.Nullable;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Window;
+import org.springframework.data.geo.GeoResults;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.SerializationUtils;
@@ -57,7 +58,8 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 
 		Assert.notNull(domainType, "DomainType must not be null");
 
-		return new ExecutableFindSupport<>(template, domainType, domainType, null, ALL_QUERY);
+		return new ExecutableFindSupport<>(template, domainType, domainType, QueryResultConverter.entity(), null,
+				ALL_QUERY);
 	}
 
 	/**
@@ -65,19 +67,22 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 	 * @author Christoph Strobl
 	 * @since 2.0
 	 */
-	static class ExecutableFindSupport<T>
+	static class ExecutableFindSupport<S, T>
 			implements ExecutableFind<T>, FindWithCollection<T>, FindWithProjection<T>, FindWithQuery<T> {
 
 		private final MongoTemplate template;
 		private final Class<?> domainType;
-		private final Class<T> returnType;
+		private final Class<S> returnType;
+		private final QueryResultConverter<? super S, ? extends T> resultConverter;
 		private final @Nullable String collection;
 		private final Query query;
 
-		ExecutableFindSupport(MongoTemplate template, Class<?> domainType, Class<T> returnType, @Nullable String collection,
+		ExecutableFindSupport(MongoTemplate template, Class<?> domainType, Class<S> returnType,
+				QueryResultConverter<? super S, ? extends T> resultConverter, @Nullable String collection,
 				Query query) {
 			this.template = template;
 			this.domainType = domainType;
+			this.resultConverter = resultConverter;
 			this.returnType = returnType;
 			this.collection = collection;
 			this.query = query;
@@ -89,7 +94,7 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 
 			Assert.hasText(collection, "Collection name must not be null nor empty");
 
-			return new ExecutableFindSupport<>(template, domainType, returnType, collection, query);
+			return new ExecutableFindSupport<>(template, domainType, returnType, resultConverter, collection, query);
 		}
 
 		@Override
@@ -98,7 +103,8 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 
 			Assert.notNull(returnType, "ReturnType must not be null");
 
-			return new ExecutableFindSupport<>(template, domainType, returnType, collection, query);
+			return new ExecutableFindSupport<>(template, domainType, returnType, QueryResultConverter.entity(), collection,
+					query);
 		}
 
 		@Override
@@ -107,7 +113,16 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 
 			Assert.notNull(query, "Query must not be null");
 
-			return new ExecutableFindSupport<>(template, domainType, returnType, collection, query);
+			return new ExecutableFindSupport<>(template, domainType, returnType, resultConverter, collection, query);
+		}
+
+		@Override
+		public <R> TerminatingResults<R> map(QueryResultConverter<? super T, ? extends R> converter) {
+
+			Assert.notNull(converter, "QueryResultConverter must not be null");
+
+			return new ExecutableFindSupport<>(template, domainType, returnType, this.resultConverter.andThen(converter),
+					collection, query);
 		}
 
 		@Override
@@ -146,12 +161,13 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 
 		@Override
 		public Window<T> scroll(ScrollPosition scrollPosition) {
-			return template.doScroll(query.with(scrollPosition), domainType, returnType, getCollectionName());
+			return template.doScroll(query.with(scrollPosition), domainType, returnType, resultConverter,
+					getCollectionName());
 		}
 
 		@Override
 		public TerminatingFindNear<T> near(NearQuery nearQuery) {
-			return () -> template.geoNear(nearQuery, domainType, getCollectionName(), returnType);
+			return new TerminatingFindNearSupport<>(nearQuery, this.resultConverter);
 		}
 
 		@Override
@@ -179,17 +195,17 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 			Document fieldsObject = query.getFieldsObject();
 
 			return template.doFind(template.createDelegate(query), getCollectionName(), queryObject, fieldsObject, domainType,
-					returnType, getCursorPreparer(query, preparer));
+					returnType, resultConverter, getCursorPreparer(query, preparer));
 		}
 
 		private List<T> doFindDistinct(String field) {
 
 			return template.findDistinct(query, field, getCollectionName(), domainType,
-					returnType == domainType ? (Class<T>) Object.class : returnType);
+					returnType == domainType ? (Class) Object.class : returnType);
 		}
 
 		private Stream<T> doStream() {
-			return template.doStream(query, domainType, getCollectionName(), returnType);
+			return template.doStream(query, domainType, getCollectionName(), returnType, resultConverter);
 		}
 
 		private CursorPreparer getCursorPreparer(Query query, @Nullable CursorPreparer preparer) {
@@ -203,6 +219,31 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 		private String asString() {
 			return SerializationUtils.serializeToJsonSafely(query);
 		}
+
+		class TerminatingFindNearSupport<G> implements TerminatingFindNear<G> {
+
+			private final NearQuery nearQuery;
+			private final QueryResultConverter<? super S, ? extends G> resultConverter;
+
+			public TerminatingFindNearSupport(NearQuery nearQuery,
+					QueryResultConverter<? super S, ? extends G> resultConverter) {
+				this.nearQuery = nearQuery;
+				this.resultConverter = resultConverter;
+			}
+
+			@Override
+			public <R> TerminatingFindNear<R> map(QueryResultConverter<? super G, ? extends R> converter) {
+
+				Assert.notNull(converter, "QueryResultConverter must not be null");
+
+				return new TerminatingFindNearSupport<>(nearQuery, this.resultConverter.andThen(converter));
+			}
+
+			@Override
+			public GeoResults<G> all() {
+				return template.doGeoNear(nearQuery, domainType, getCollectionName(), returnType, resultConverter);
+			}
+		}
 	}
 
 	/**
@@ -245,19 +286,19 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 	 * @author Christoph Strobl
 	 * @since 2.1
 	 */
-	static class DistinctOperationSupport<T> implements TerminatingDistinct<T> {
+	static class DistinctOperationSupport<S, T> implements TerminatingDistinct<T> {
 
 		private final String field;
-		private final ExecutableFindSupport<T> delegate;
+		private final ExecutableFindSupport<S, T> delegate;
 
-		public DistinctOperationSupport(ExecutableFindSupport<T> delegate, String field) {
+		public DistinctOperationSupport(ExecutableFindSupport<S, T> delegate, String field) {
 
 			this.delegate = delegate;
 			this.field = field;
 		}
 
 		@Override
-		@SuppressWarnings("unchecked")
+		@SuppressWarnings({ "unchecked", "rawtypes" })
 		@Contract("_ -> new")
 		public <R> TerminatingDistinct<R> as(Class<R> resultType) {
 
@@ -272,12 +313,13 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 
 			Assert.notNull(query, "Query must not be null");
 
-			return new DistinctOperationSupport<>((ExecutableFindSupport<T>) delegate.matching(query), field);
+			return new DistinctOperationSupport<>((ExecutableFindSupport<S, T>) delegate.matching(query), field);
 		}
 
 		@Override
 		public List<T> all() {
 			return delegate.doFindDistinct(field);
 		}
+
 	}
 }

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

@@ -131,6 +131,7 @@ import org.springframework.data.mongodb.core.timeseries.Granularity;
 import org.springframework.data.mongodb.core.validation.Validator;
 import org.springframework.data.projection.EntityProjection;
 import org.springframework.data.util.CloseableIterator;
+import org.springframework.data.util.Lazy;
 import org.springframework.data.util.Optionals;
 import org.springframework.lang.Contract;
 import org.springframework.util.Assert;
@@ -516,13 +517,19 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 
 	@SuppressWarnings({ "ConstantConditions", "NullAway" })
 	protected <T> Stream<T> doStream(Query query, Class<?> entityType, String collectionName, Class<T> returnType) {
+		return doStream(query, entityType, collectionName, returnType, QueryResultConverter.entity());
+	}
+
+	@SuppressWarnings("ConstantConditions")
+	<T, R> Stream<R> doStream(Query query, Class<?> entityType, String collectionName, Class<T> returnType,
+			QueryResultConverter<? super T, ? extends R> resultConverter) {
 
 		Assert.notNull(query, "Query must not be null");
 		Assert.notNull(entityType, "Entity type must not be null");
 		Assert.hasText(collectionName, "Collection name must not be null or empty");
 		Assert.notNull(returnType, "ReturnType must not be null");
 
-		return execute(collectionName, (CollectionCallback<Stream<T>>) collection -> {
+		return execute(collectionName, (CollectionCallback<Stream<R>>) collection -> {
 
 			MongoPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(entityType);
 
@@ -536,8 +543,10 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 			FindIterable<Document> cursor = new QueryCursorPreparer(query, entityType).initiateFind(collection,
 					col -> readPreference.prepare(col).find(mappedQuery, Document.class).projection(mappedFields));
 
+			DocumentCallback<R> resultReader = getResultReader(projection, collectionName, resultConverter);
+
 			return new CloseableIterableCursorAdapter<>(cursor, exceptionTranslator,
-					new ProjectingReadCallback<>(mongoConverter, projection, collectionName)).stream();
+					resultReader).stream();
 		});
 	}
 
@@ -936,10 +945,11 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 
 	@Override
 	public <T> Window<T> scroll(Query query, Class<T> entityType, String collectionName) {
-		return doScroll(query, entityType, entityType, collectionName);
+		return doScroll(query, entityType, entityType, QueryResultConverter.entity(), collectionName);
 	}
 
-	<T> Window<T> doScroll(Query query, Class<?> sourceClass, Class<T> targetClass, String collectionName) {
+	<T, R> Window<R> doScroll(Query query, Class<?> sourceClass, Class<T> targetClass,
+			QueryResultConverter<? super T, ? extends R> resultConverter, String collectionName) {
 
 		Assert.notNull(query, "Query must not be null");
 		Assert.notNull(collectionName, "CollectionName must not be null");
@@ -947,7 +957,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 		Assert.notNull(targetClass, "Target type must not be null");
 
 		EntityProjection<T, ?> projection = operations.introspectProjection(targetClass, sourceClass);
-		ProjectingReadCallback<?, T> callback = new ProjectingReadCallback<>(mongoConverter, projection, collectionName);
+		DocumentCallback<R> callback = getResultReader(projection, collectionName, resultConverter);
 		int limit = query.isLimited() ? query.getLimit() + 1 : Integer.MAX_VALUE;
 
 		if (query.hasKeyset()) {
@@ -955,14 +965,14 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 			KeysetScrollQuery keysetPaginationQuery = ScrollUtils.createKeysetPaginationQuery(query,
 					operations.getIdPropertyName(sourceClass));
 
-			List<T> result = doFind(collectionName, createDelegate(query), keysetPaginationQuery.query(),
+			List<R> result = doFind(collectionName, createDelegate(query), keysetPaginationQuery.query(),
 					keysetPaginationQuery.fields(), sourceClass,
 					new QueryCursorPreparer(query, keysetPaginationQuery.sort(), limit, 0, sourceClass), callback);
 
 			return ScrollUtils.createWindow(query, result, sourceClass, operations);
 		}
 
-		List<T> result = doFind(collectionName, createDelegate(query), query.getQueryObject(), query.getFieldsObject(),
+		List<R> result = doFind(collectionName, createDelegate(query), query.getQueryObject(), query.getFieldsObject(),
 				sourceClass, new QueryCursorPreparer(query, query.getSortObject(), limit, query.getSkip(), sourceClass),
 				callback);
 
@@ -1054,6 +1064,11 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 	}
 
 	public <T> GeoResults<T> geoNear(NearQuery near, Class<?> domainType, String collectionName, Class<T> returnType) {
+		return doGeoNear(near, domainType, collectionName, returnType, QueryResultConverter.entity());
+	}
+
+	<T, R> GeoResults<R> doGeoNear(NearQuery near, Class<?> domainType, String collectionName, Class<T> returnType,
+			QueryResultConverter<? super T, ? extends R> resultConverter) {
 
 		if (near == null) {
 			throw new InvalidDataAccessApiUsageException("NearQuery must not be null");
@@ -1085,15 +1100,15 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 		AggregationResults<Document> results = aggregate($geoNear, collection, Document.class);
 		EntityProjection<T, ?> projection = operations.introspectProjection(returnType, domainType);
 
-		DocumentCallback<GeoResult<T>> callback = new GeoNearResultDocumentCallback<>(distanceField,
-				new ProjectingReadCallback<>(mongoConverter, projection, collection), near.getMetric());
+		DocumentCallback<GeoResult<R>> callback = new GeoNearResultDocumentCallback<>(distanceField,
+				getResultReader(projection, collectionName, resultConverter), near.getMetric());
 
-		List<GeoResult<T>> result = new ArrayList<>(results.getMappedResults().size());
+		List<GeoResult<R>> result = new ArrayList<>(results.getMappedResults().size());
 
 		BigDecimal aggregate = BigDecimal.ZERO;
 		for (Document element : results) {
 
-			GeoResult<T> geoResult = callback.doWith(element);
+			GeoResult<R> geoResult = callback.doWith(element);
 			aggregate = aggregate.add(BigDecimal.valueOf(geoResult.getDistance().getValue()));
 			result.add(geoResult);
 		}
@@ -2060,7 +2075,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 	@Override
 	public <O> AggregationResults<O> aggregate(TypedAggregation<?> aggregation, String inputCollectionName,
 			Class<O> outputType) {
-		return aggregate(aggregation, inputCollectionName, outputType, null);
+		return aggregate(aggregation, inputCollectionName, outputType, (AggregationOperationContext) null);
 	}
 
 	@Override
@@ -2073,7 +2088,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 
 	@Override
 	public <O> AggregationResults<O> aggregate(Aggregation aggregation, String collectionName, Class<O> outputType) {
-		return aggregate(aggregation, collectionName, outputType, null);
+		return doAggregate(aggregation, collectionName, outputType, QueryResultConverter.entity());
 	}
 
 	@Override
@@ -2204,11 +2219,25 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 		return doAggregate(aggregation, collectionName, outputType, context.getAggregationOperationContext());
 	}
 
+	<T, O> AggregationResults<O> doAggregate(Aggregation aggregation, String collectionName, Class<T> outputType,
+			QueryResultConverter<? super T, ? extends O> resultConverter) {
+
+		return doAggregate(aggregation, collectionName, outputType, resultConverter, queryOperations
+				.createAggregation(aggregation, (AggregationOperationContext) null).getAggregationOperationContext());
+	}
+
 	@SuppressWarnings({ "ConstantConditions", "NullAway" })
 	protected <O> AggregationResults<O> doAggregate(Aggregation aggregation, String collectionName, Class<O> outputType,
 			AggregationOperationContext context) {
+		return doAggregate(aggregation, collectionName, outputType, QueryResultConverter.entity(), context);
+	}
+
+	@SuppressWarnings("ConstantConditions")
+	<T, O> AggregationResults<O> doAggregate(Aggregation aggregation, String collectionName, Class<T> outputType,
+			QueryResultConverter<? super T, ? extends O> resultConverter, AggregationOperationContext context) {
 
-		ReadDocumentCallback<O> callback = new ReadDocumentCallback<>(mongoConverter, outputType, collectionName);
+		DocumentCallback<O> callback = new QueryResultConverterCallback<>(resultConverter,
+				new ReadDocumentCallback<>(mongoConverter, outputType, collectionName));
 
 		AggregationOptions options = aggregation.getOptions();
 		AggregationUtil aggregationUtil = new AggregationUtil(queryMapper, mappingContext);
@@ -2287,9 +2316,15 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 		});
 	}
 
-	@SuppressWarnings({ "ConstantConditions", "NullAway" })
 	protected <O> Stream<O> aggregateStream(Aggregation aggregation, String collectionName, Class<O> outputType,
 			@Nullable AggregationOperationContext context) {
+		return doAggregateStream(aggregation, collectionName, outputType, QueryResultConverter.entity(), context);
+	}
+
+	@SuppressWarnings({ "ConstantConditions", "NullAway" })
+	protected <T, O> Stream<O> doAggregateStream(Aggregation aggregation, String collectionName, Class<T> outputType,
+			QueryResultConverter<? super T, ? extends O> resultConverter,
+			@Nullable AggregationOperationContext context) {
 
 		Assert.notNull(aggregation, "Aggregation pipeline must not be null");
 		Assert.hasText(collectionName, "Collection name must not be null or empty");
@@ -2306,7 +2341,8 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 					String.format("Streaming aggregation: %s in collection %s", serializeToJsonSafely(pipeline), collectionName));
 		}
 
-		ReadDocumentCallback<O> readCallback = new ReadDocumentCallback<>(mongoConverter, outputType, collectionName);
+		DocumentCallback<O> readCallback = new QueryResultConverterCallback<>(resultConverter,
+				new ReadDocumentCallback<>(mongoConverter, outputType, collectionName));
 
 		return execute(collectionName, (CollectionCallback<Stream<O>>) collection -> {
 
@@ -2670,11 +2706,12 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 	 *
 	 * @since 2.0
 	 */
-	<S, T> List<T> doFind(CollectionPreparer<MongoCollection<Document>> collectionPreparer, String collectionName,
-			Document query, Document fields, Class<S> sourceClass, Class<T> targetClass, CursorPreparer preparer) {
+	<T, R> List<R> doFind(CollectionPreparer<MongoCollection<Document>> collectionPreparer, String collectionName,
+			Document query, Document fields, Class<?> sourceClass, Class<T> targetClass,
+			QueryResultConverter<? super T, ? extends R> resultConverter, CursorPreparer preparer) {
 
 		MongoPersistentEntity<?> entity = mappingContext.getPersistentEntity(sourceClass);
-		EntityProjection<T, S> projection = operations.introspectProjection(targetClass, sourceClass);
+		EntityProjection<T, ?> projection = operations.introspectProjection(targetClass, sourceClass);
 
 		QueryContext queryContext = queryOperations.createQueryContext(new BasicQuery(query, fields));
 		Document mappedFields = queryContext.getMappedFields(entity, projection);
@@ -2690,8 +2727,9 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 					collectionName));
 		}
 
+		DocumentCallback<R> callback = getResultReader(projection, collectionName, resultConverter);
 		return executeFindMultiInternal(new FindCallback(collectionPreparer, mappedQuery, mappedFields, null), preparer,
-				new ProjectingReadCallback<>(mongoConverter, projection, collectionName), collectionName);
+				callback, collectionName);
 	}
 
 	/**
@@ -3014,6 +3052,16 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 		}
 	}
 
+	@SuppressWarnings("unchecked")
+	private <T, R> DocumentCallback<R> getResultReader(EntityProjection<T, ?> projection, String collectionName,
+			QueryResultConverter<? super T, ? extends R> resultConverter) {
+
+		DocumentCallback<T> readCallback = new ProjectingReadCallback<>(mongoConverter, projection, collectionName);
+
+		return resultConverter == QueryResultConverter.entity() ? (DocumentCallback<R>) readCallback
+				: new QueryResultConverterCallback<T, R>(resultConverter, readCallback);
+	}
+
 	public PersistenceExceptionTranslator getExceptionTranslator() {
 		return exceptionTranslator;
 	}
@@ -3373,6 +3421,24 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 		}
 	}
 
+	static final class QueryResultConverterCallback<T, R> implements DocumentCallback<R> {
+
+		private final QueryResultConverter<? super T, ? extends R> converter;
+		private final DocumentCallback<T> delegate;
+
+		QueryResultConverterCallback(QueryResultConverter<? super T, ? extends R> converter, DocumentCallback<T> delegate) {
+			this.converter = converter;
+			this.delegate = delegate;
+		}
+
+		@Override
+		public R doWith(Document object) {
+
+			Lazy<T> lazy = Lazy.of(() -> delegate.doWith(object));
+			return converter.mapDocument(object, lazy::get);
+		}
+	}
+
 	/**
 	 * {@link DocumentCallback} transforming {@link Document} into the given {@code targetType} or decorating the
 	 * {@code sourceType} with a {@literal projection} in case the {@code targetType} is an {@literal interface}.

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

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2025 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;
+
+import org.bson.Document;
+
+/**
+ * Converter for MongoDB query results.
+ * <p>
+ * This is a functional interface that allows for mapping a {@link Document} to a result type.
+ * {@link #mapDocument(Document, ConversionResultSupplier) row mapping} can obtain upstream a
+ * {@link ConversionResultSupplier upstream converter} to enrich the final result object. This is useful when e.g.
+ * wrapping result objects where the wrapper needs to obtain information from the actual {@link Document}.
+ *
+ * @param <T> object type accepted by this converter.
+ * @param <R> the returned result type.
+ * @author Mark Paluch
+ * @since x.x
+ */
+@FunctionalInterface
+public interface QueryResultConverter<T, R> {
+
+	/**
+	 * Returns a function that returns the materialized entity.
+	 *
+	 * @param <T> the type of the input and output entity to the function.
+	 * @return a function that returns the materialized entity.
+	 */
+	@SuppressWarnings("unchecked")
+	static <T> QueryResultConverter<T, T> entity() {
+		return (QueryResultConverter<T, T>) EntityResultConverter.INSTANCE;
+	}
+
+	/**
+	 * Map a {@link Document} that is read from the MongoDB query/aggregation operation to a query result.
+	 *
+	 * @param document the raw document from the MongoDB query/aggregation result.
+	 * @param reader reader object that supplies an upstream result from an earlier converter.
+	 * @return the mapped result.
+	 */
+	R mapDocument(Document document, ConversionResultSupplier<T> reader);
+
+	/**
+	 * Returns a composed function that first applies this function to its input, and then applies the {@code after}
+	 * function to the result. If evaluation of either function throws an exception, it is relayed to the caller of the
+	 * composed function.
+	 *
+	 * @param <V> the type of output of the {@code after} function, and of the composed function.
+	 * @param after the function to apply after this function is applied.
+	 * @return a composed function that first applies this function and then applies the {@code after} function.
+	 */
+	default <V> QueryResultConverter<T, V> andThen(QueryResultConverter<? super R, ? extends V> after) {
+		return (row, reader) -> after.mapDocument(row, () -> mapDocument(row, reader));
+	}
+
+	/**
+	 * A supplier that converts a {@link Document} into {@code T}. Allows for lazy reading of query results.
+	 *
+	 * @param <T> type of the returned result.
+	 */
+	interface ConversionResultSupplier<T> {
+
+		/**
+		 * Obtain the upstream conversion result.
+		 *
+		 * @return the upstream conversion result.
+		 */
+		T get();
+
+	}
+
+}

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

@@ -18,6 +18,7 @@ package org.springframework.data.mongodb.core;
 import reactor.core.publisher.Flux;
 
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.lang.Contract;
 
 /**
  * {@link ReactiveAggregationOperation} allows creation and execution of reactive MongoDB aggregation operations in a
@@ -44,7 +45,7 @@ public interface ReactiveAggregationOperation {
 	/**
 	 * Start creating an aggregation operation that returns results mapped to the given domain type. <br />
 	 * Use {@link org.springframework.data.mongodb.core.aggregation.TypedAggregation} to specify a potentially different
-	 * input type for he aggregation.
+	 * input type for the aggregation.
 	 *
 	 * @param domainType must not be {@literal null}.
 	 * @return new instance of {@link ReactiveAggregation}. Never {@literal null}.
@@ -73,6 +74,18 @@ public interface ReactiveAggregationOperation {
 	 */
 	interface TerminatingAggregationOperation<T> {
 
+		/**
+		 * Map the query result to a different type using {@link QueryResultConverter}.
+		 *
+		 * @param <R> {@link Class type} of the result.
+		 * @param converter the converter, must not be {@literal null}.
+		 * @return new instance of {@link ExecutableFindOperation.TerminatingFindNear}.
+		 * @throws IllegalArgumentException if {@link QueryResultConverter converter} is {@literal null}.
+		 * @since x.y
+		 */
+		@Contract("_ -> new")
+		<R> TerminatingAggregationOperation<R> map(QueryResultConverter<? super T, ? extends R> converter);
+
 		/**
 		 * Apply pipeline operations as specified and stream all matching elements. <br />
 		 *

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

@@ -52,22 +52,25 @@ class ReactiveAggregationOperationSupport implements ReactiveAggregationOperatio
 
 		Assert.notNull(domainType, "DomainType must not be null");
 
-		return new ReactiveAggregationSupport<>(template, domainType, null, null);
+		return new ReactiveAggregationSupport<>(template, domainType, QueryResultConverter.entity(), null, null);
 	}
 
-	static class ReactiveAggregationSupport<T>
+	static class ReactiveAggregationSupport<S, T>
 			implements AggregationOperationWithAggregation<T>, ReactiveAggregation<T>, TerminatingAggregationOperation<T> {
 
 		private final ReactiveMongoTemplate template;
-		private final Class<T> domainType;
+		private final Class<S> domainType;
+		private final QueryResultConverter<? super S, ? extends T> resultConverter;
 		private final @Nullable Aggregation aggregation;
 		private final @Nullable String collection;
 
-		ReactiveAggregationSupport(ReactiveMongoTemplate template, Class<T> domainType, @Nullable Aggregation aggregation,
+		ReactiveAggregationSupport(ReactiveMongoTemplate template, Class<S> domainType,
+				QueryResultConverter<? super S, ? extends T> resultConverter, @Nullable Aggregation aggregation,
 				@Nullable String collection) {
 
 			this.template = template;
 			this.domainType = domainType;
+			this.resultConverter = resultConverter;
 			this.aggregation = aggregation;
 			this.collection = collection;
 		}
@@ -77,7 +80,7 @@ class ReactiveAggregationOperationSupport implements ReactiveAggregationOperatio
 
 			Assert.hasText(collection, "Collection must not be null nor empty");
 
-			return new ReactiveAggregationSupport<>(template, domainType, aggregation, collection);
+			return new ReactiveAggregationSupport<>(template, domainType, resultConverter, aggregation, collection);
 		}
 
 		@Override
@@ -85,7 +88,16 @@ class ReactiveAggregationOperationSupport implements ReactiveAggregationOperatio
 
 			Assert.notNull(aggregation, "Aggregation must not be null");
 
-			return new ReactiveAggregationSupport<>(template, domainType, aggregation, collection);
+			return new ReactiveAggregationSupport<>(template, domainType, resultConverter, aggregation, collection);
+		}
+
+		@Override
+		public <R> TerminatingAggregationOperation<R> map(QueryResultConverter<? super T, ? extends R> converter) {
+
+			Assert.notNull(converter, "QueryResultConverter must not be null");
+
+			return new ReactiveAggregationSupport<>(template, domainType, resultConverter.andThen(converter), aggregation,
+					collection);
 		}
 
 		@Override
@@ -93,7 +105,7 @@ class ReactiveAggregationOperationSupport implements ReactiveAggregationOperatio
 
 			Assert.notNull(aggregation, "Aggregation must be set first");
 
-			return template.aggregate(aggregation, getCollectionName(aggregation), domainType);
+			return template.doAggregate(aggregation, getCollectionName(aggregation), domainType, domainType, resultConverter);
 		}
 
 		private String getCollectionName(Aggregation aggregation) {

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

@@ -25,6 +25,7 @@ import org.springframework.data.geo.GeoResult;
 import org.springframework.data.mongodb.core.query.CriteriaDefinition;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.lang.Contract;
 
 /**
  * {@link ReactiveFindOperation} allows creation and execution of reactive MongoDB find operations in a fluent API
@@ -66,7 +67,28 @@ public interface ReactiveFindOperation {
 	/**
 	 * Compose find execution by calling one of the terminating methods.
 	 */
-	interface TerminatingFind<T> {
+	interface TerminatingFind<T> extends TerminatingResults<T>, TerminatingProjection {
+
+	}
+
+	/**
+	 * Compose find execution by calling one of the terminating methods.
+	 *
+	 * @since x.y
+	 */
+	interface TerminatingResults<T> {
+
+		/**
+		 * Map the query result to a different type using {@link QueryResultConverter}.
+		 *
+		 * @param <R> {@link Class type} of the result.
+		 * @param converter the converter, must not be {@literal null}.
+		 * @return new instance of {@link TerminatingResults}.
+		 * @throws IllegalArgumentException if {@link QueryResultConverter converter} is {@literal null}.
+		 * @since x.y
+		 */
+		@Contract("_ -> new")
+		<R> TerminatingResults<R> map(QueryResultConverter<? super T, ? extends R> converter);
 
 		/**
 		 * Get exactly zero or one result.
@@ -120,6 +142,15 @@ public interface ReactiveFindOperation {
 		 */
 		Flux<T> tail();
 
+	}
+
+	/**
+	 * Compose find execution by calling one of the terminating methods.
+	 *
+	 * @since x.y
+	 */
+	interface TerminatingProjection {
+
 		/**
 		 * Get the number of matching elements. <br />
 		 * This method uses an
@@ -145,6 +176,18 @@ public interface ReactiveFindOperation {
 	 */
 	interface TerminatingFindNear<T> {
 
+		/**
+		 * Map the query result to a different type using {@link QueryResultConverter}.
+		 *
+		 * @param <R> {@link Class type} of the result.
+		 * @param converter the converter, must not be {@literal null}.
+		 * @return new instance of {@link ExecutableFindOperation.TerminatingFindNear}.
+		 * @throws IllegalArgumentException if {@link QueryResultConverter converter} is {@literal null}.
+		 * @since x.y
+		 */
+		@Contract("_ -> new")
+		<R> TerminatingFindNear<R> map(QueryResultConverter<? super T, ? extends R> converter);
+
 		/**
 		 * Find all matching elements and return them as {@link org.springframework.data.geo.GeoResult}.
 		 *

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

@@ -21,8 +21,9 @@ import reactor.core.publisher.Mono;
 import org.bson.Document;
 import org.jspecify.annotations.Nullable;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
-import org.springframework.data.domain.Window;
 import org.springframework.data.domain.ScrollPosition;
+import org.springframework.data.domain.Window;
+import org.springframework.data.geo.GeoResult;
 import org.springframework.data.mongodb.core.CollectionPreparerSupport.ReactiveCollectionPreparerDelegate;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
@@ -52,7 +53,7 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 
 		Assert.notNull(domainType, "DomainType must not be null");
 
-		return new ReactiveFindSupport<>(template, domainType, domainType, null, ALL_QUERY);
+		return new ReactiveFindSupport<>(template, domainType, domainType, QueryResultConverter.entity(), null, ALL_QUERY);
 	}
 
 	/**
@@ -61,21 +62,24 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 	 * @author Christoph Strobl
 	 * @since 2.0
 	 */
-	static class ReactiveFindSupport<T>
+	static class ReactiveFindSupport<S, T>
 			implements ReactiveFind<T>, FindWithCollection<T>, FindWithProjection<T>, FindWithQuery<T> {
 
 		private final ReactiveMongoTemplate template;
 		private final Class<?> domainType;
-		private final Class<T> returnType;
+		private final Class<S> returnType;
+		private final QueryResultConverter<? super S, ? extends T> resultConverter;
 		private final @Nullable String collection;
 		private final Query query;
 
-		ReactiveFindSupport(ReactiveMongoTemplate template, Class<?> domainType, Class<T> returnType, @Nullable String collection,
+		ReactiveFindSupport(ReactiveMongoTemplate template, Class<?> domainType, Class<S> returnType,
+				QueryResultConverter<? super S, ? extends T> resultConverter, @Nullable String collection,
 				Query query) {
 
 			this.template = template;
 			this.domainType = domainType;
 			this.returnType = returnType;
+			this.resultConverter = resultConverter;
 			this.collection = collection;
 			this.query = query;
 		}
@@ -85,7 +89,7 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 
 			Assert.hasText(collection, "Collection name must not be null nor empty");
 
-			return new ReactiveFindSupport<>(template, domainType, returnType, collection, query);
+			return new ReactiveFindSupport<>(template, domainType, returnType, resultConverter, collection, query);
 		}
 
 		@Override
@@ -93,7 +97,8 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 
 			Assert.notNull(returnType, "ReturnType must not be null");
 
-			return new ReactiveFindSupport<>(template, domainType, returnType, collection, query);
+			return new ReactiveFindSupport<>(template, domainType, returnType, QueryResultConverter.entity(), collection,
+					query);
 		}
 
 		@Override
@@ -101,7 +106,16 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 
 			Assert.notNull(query, "Query must not be null");
 
-			return new ReactiveFindSupport<>(template, domainType, returnType, collection, query);
+			return new ReactiveFindSupport<>(template, domainType, returnType, resultConverter, collection, query);
+		}
+
+		@Override
+		public <R> TerminatingResults<R> map(QueryResultConverter<? super T, ? extends R> converter) {
+
+			Assert.notNull(converter, "QueryResultConverter must not be null");
+
+			return new ReactiveFindSupport<>(template, domainType, returnType, this.resultConverter.andThen(converter),
+					collection, query);
 		}
 
 		@Override
@@ -141,7 +155,8 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 
 		@Override
 		public Mono<Window<T>> scroll(ScrollPosition scrollPosition) {
-			return template.doScroll(query.with(scrollPosition), domainType, returnType, getCollectionName());
+			return template.doScroll(query.with(scrollPosition), domainType, returnType, resultConverter,
+					getCollectionName());
 		}
 
 		@Override
@@ -151,7 +166,7 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 
 		@Override
 		public TerminatingFindNear<T> near(NearQuery nearQuery) {
-			return () -> template.geoNear(nearQuery, domainType, getCollectionName(), returnType);
+			return new TerminatingFindNearSupport<>(nearQuery, resultConverter);
 		}
 
 		@Override
@@ -178,14 +193,15 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 			Document fieldsObject = query.getFieldsObject();
 
 			return template.doFind(getCollectionName(), ReactiveCollectionPreparerDelegate.of(query), queryObject,
-					fieldsObject, domainType, returnType, preparer != null ? preparer : getCursorPreparer(query));
+					fieldsObject, domainType, returnType, resultConverter,
+					preparer != null ? preparer : getCursorPreparer(query));
 		}
 
-		@SuppressWarnings("unchecked")
+		@SuppressWarnings({ "unchecked", "rawtypes" })
 		private Flux<T> doFindDistinct(String field) {
 
 			return template.findDistinct(query, field, getCollectionName(), domainType,
-					returnType == domainType ? (Class<T>) Object.class : returnType);
+					returnType == domainType ? (Class) Object.class : returnType);
 		}
 
 		private FindPublisherPreparer getCursorPreparer(Query query) {
@@ -200,10 +216,36 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 			return SerializationUtils.serializeToJsonSafely(query);
 		}
 
+		class TerminatingFindNearSupport<G> implements TerminatingFindNear<G> {
+
+			private final NearQuery nearQuery;
+			private final QueryResultConverter<? super S, ? extends G> resultConverter;
+
+			public TerminatingFindNearSupport(NearQuery nearQuery,
+					QueryResultConverter<? super S, ? extends G> resultConverter) {
+				this.nearQuery = nearQuery;
+				this.resultConverter = resultConverter;
+			}
+
+			@Override
+			public <R> TerminatingFindNear<R> map(QueryResultConverter<? super G, ? extends R> converter) {
+
+				Assert.notNull(converter, "QueryResultConverter must not be null");
+
+				return new TerminatingFindNearSupport<>(nearQuery, this.resultConverter.andThen(converter));
+			}
+
+			@Override
+			public Flux<GeoResult<G>> all() {
+				return template.doGeoNear(nearQuery, domainType, getCollectionName(), returnType, resultConverter);
+			}
+		}
+
 		/**
 		 * @author Christoph Strobl
 		 * @since 2.1
 		 */
+		@SuppressWarnings({ "unchecked", "rawtypes" })
 		static class DistinctOperationSupport<T> implements TerminatingDistinct<T> {
 
 			private final String field;
@@ -224,12 +266,11 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 			}
 
 			@Override
-			@SuppressWarnings("unchecked")
 			public TerminatingDistinct<T> matching(Query query) {
 
 				Assert.notNull(query, "Query must not be null");
 
-				return new DistinctOperationSupport<>((ReactiveFindSupport<T>) delegate.matching(query), field);
+				return new DistinctOperationSupport<>((ReactiveFindSupport) delegate.matching(query), field);
 			}
 
 			@Override

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

@@ -889,10 +889,11 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 
 	@Override
 	public <T> Mono<Window<T>> scroll(Query query, Class<T> entityType, String collectionName) {
-		return doScroll(query, entityType, entityType, collectionName);
+		return doScroll(query, entityType, entityType, QueryResultConverter.entity(), collectionName);
 	}
 
-	<T> Mono<Window<T>> doScroll(Query query, Class<?> sourceClass, Class<T> targetClass, String collectionName) {
+	<T, R> Mono<Window<R>> doScroll(Query query, Class<?> sourceClass, Class<T> targetClass,
+			QueryResultConverter<? super T, ? extends R> resultConverter, String collectionName) {
 
 		Assert.notNull(query, "Query must not be null");
 		Assert.notNull(collectionName, "CollectionName must not be null");
@@ -900,7 +901,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		Assert.notNull(targetClass, "Target type must not be null");
 
 		EntityProjection<T, ?> projection = operations.introspectProjection(targetClass, sourceClass);
-		ProjectingReadCallback<?, T> callback = new ProjectingReadCallback<>(mongoConverter, projection, collectionName);
+		DocumentCallback<R> callback = getResultReader(projection, collectionName, resultConverter);
 		int limit = query.isLimited() ? query.getLimit() + 1 : Integer.MAX_VALUE;
 
 		if (query.hasKeyset()) {
@@ -908,7 +909,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 			KeysetScrollQuery keysetPaginationQuery = ScrollUtils.createKeysetPaginationQuery(query,
 					operations.getIdPropertyName(sourceClass));
 
-			Mono<List<T>> result = doFind(collectionName, ReactiveCollectionPreparerDelegate.of(query),
+			Mono<List<R>> result = doFind(collectionName, ReactiveCollectionPreparerDelegate.of(query),
 					keysetPaginationQuery.query(), keysetPaginationQuery.fields(), sourceClass,
 					new QueryFindPublisherPreparer(query, keysetPaginationQuery.sort(), limit, 0, sourceClass), callback)
 					.collectList();
@@ -916,7 +917,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 			return result.map(it -> ScrollUtils.createWindow(query, it, sourceClass, operations));
 		}
 
-		Mono<List<T>> result = doFind(collectionName, ReactiveCollectionPreparerDelegate.of(query), query.getQueryObject(),
+		Mono<List<R>> result = doFind(collectionName, ReactiveCollectionPreparerDelegate.of(query), query.getQueryObject(),
 				query.getFieldsObject(), sourceClass,
 				new QueryFindPublisherPreparer(query, query.getSortObject(), limit, query.getSkip(), sourceClass), callback)
 				.collectList();
@@ -1015,6 +1016,11 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 
 	protected <O> Flux<O> doAggregate(Aggregation aggregation, String collectionName, @Nullable Class<?> inputType,
 			Class<O> outputType) {
+		return doAggregate(aggregation, collectionName, inputType, outputType, QueryResultConverter.entity());
+	}
+
+	<T, O> Flux<O> doAggregate(Aggregation aggregation, String collectionName, @Nullable Class<?> inputType,
+			Class<T> outputType, QueryResultConverter<? super T, ? extends O> resultConverter) {
 
 		Assert.notNull(aggregation, "Aggregation pipeline must not be null");
 		Assert.hasText(collectionName, "Collection name must not be null or empty");
@@ -1030,13 +1036,14 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 					serializeToJsonSafely(ctx.getAggregationPipeline()), collectionName));
 		}
 
-		ReadDocumentCallback<O> readCallback = new ReadDocumentCallback<>(mongoConverter, outputType, collectionName);
+		DocumentCallback<O> readCallback = new QueryResultConverterCallback<>(resultConverter,
+				new ReadDocumentCallback<>(mongoConverter, outputType, collectionName));
 		return execute(collectionName, collection -> aggregateAndMap(collection, ctx.getAggregationPipeline(),
 				ctx.isOutOrMerge(), options, readCallback, ctx.getInputType()));
 	}
 
 	private <O> Flux<O> aggregateAndMap(MongoCollection<Document> collection, List<Document> pipeline,
-			boolean isOutOrMerge, AggregationOptions options, ReadDocumentCallback<O> readCallback,
+			boolean isOutOrMerge, AggregationOptions options, DocumentCallback<O> readCallback,
 			@Nullable Class<?> inputType) {
 
 		ReactiveCollectionPreparerDelegate collectionPreparer = ReactiveCollectionPreparerDelegate.of(options);
@@ -1082,9 +1089,14 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		return geoNear(near, entityClass, collectionName, entityClass);
 	}
 
-	@SuppressWarnings("unchecked")
 	protected <T> Flux<GeoResult<T>> geoNear(NearQuery near, Class<?> entityClass, String collectionName,
 			Class<T> returnType) {
+		return doGeoNear(near, entityClass, collectionName, returnType, QueryResultConverter.entity());
+	}
+
+	@SuppressWarnings("unchecked")
+	<T, R> Flux<GeoResult<R>> doGeoNear(NearQuery near, Class<?> entityClass, String collectionName, Class<T> returnType,
+			QueryResultConverter<? super T, ? extends R> resultConverter) {
 
 		if (near == null) {
 			throw new InvalidDataAccessApiUsageException("NearQuery must not be null");
@@ -1098,8 +1110,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		String distanceField = operations.nearQueryDistanceFieldName(entityClass);
 		EntityProjection<T, ?> projection = operations.introspectProjection(returnType, entityClass);
 
-		GeoNearResultDocumentCallback<T> callback = new GeoNearResultDocumentCallback<>(distanceField,
-				new ProjectingReadCallback<>(mongoConverter, projection, collection), near.getMetric());
+		GeoNearResultDocumentCallback<R> callback = new GeoNearResultDocumentCallback<>(distanceField,
+				getResultReader(projection, collectionName, resultConverter), near.getMetric());
 
 		Builder optionsBuilder = AggregationOptions.builder();
 		if (near.hasReadPreference()) {
@@ -2428,11 +2440,12 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 	 *
 	 * @since 2.0
 	 */
-	<S, T> Flux<T> doFind(String collectionName, CollectionPreparer<MongoCollection<Document>> collectionPreparer,
-			Document query, Document fields, Class<S> sourceClass, Class<T> targetClass, FindPublisherPreparer preparer) {
+	<T, R> Flux<R> doFind(String collectionName, CollectionPreparer<MongoCollection<Document>> collectionPreparer,
+			Document query, Document fields, Class<?> sourceClass, Class<T> targetClass,
+			QueryResultConverter<? super T, ? extends R> resultConverter, FindPublisherPreparer preparer) {
 
 		MongoPersistentEntity<?> entity = mappingContext.getPersistentEntity(sourceClass);
-		EntityProjection<T, S> projection = operations.introspectProjection(targetClass, sourceClass);
+		EntityProjection<T, ?> projection = operations.introspectProjection(targetClass, sourceClass);
 
 		QueryContext queryContext = queryOperations.createQueryContext(new BasicQuery(query, fields));
 		Document mappedFields = queryContext.getMappedFields(entity, projection);
@@ -2444,7 +2457,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		}
 
 		return executeFindMultiInternal(new FindCallback(collectionPreparer, mappedQuery, mappedFields), preparer,
-				new ProjectingReadCallback<>(mongoConverter, projection, collectionName), collectionName);
+				getResultReader(projection, collectionName, resultConverter), collectionName);
 	}
 
 	protected CreateCollectionOptions convertToCreateCollectionOptions(@Nullable CollectionOptions collectionOptions) {
@@ -2752,6 +2765,16 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		});
 	}
 
+	@SuppressWarnings("unchecked")
+	private <T, R> DocumentCallback<R> getResultReader(EntityProjection<T, ?> projection, String collectionName,
+			QueryResultConverter<? super T, ? extends R> resultConverter) {
+
+		DocumentCallback<T> readCallback = new ProjectingReadCallback<>(mongoConverter, projection, collectionName);
+
+		return resultConverter == QueryResultConverter.entity() ? (DocumentCallback<R>) readCallback
+				: new QueryResultConverterCallback<T, R>(resultConverter, readCallback);
+	}
+
 	/**
 	 * Exception translation {@link Function} intended for {@link Flux#onErrorMap(Function)} usage.
 	 *
@@ -3111,6 +3134,22 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		FindPublisher<T> doInCollection(MongoCollection<Document> collection) throws MongoException, DataAccessException;
 	}
 
+	static final class QueryResultConverterCallback<T, R> implements DocumentCallback<R> {
+
+		private final QueryResultConverter<? super T, ? extends R> converter;
+		private final DocumentCallback<T> delegate;
+
+		QueryResultConverterCallback(QueryResultConverter<? super T, ? extends R> converter, DocumentCallback<T> delegate) {
+			this.converter = converter;
+			this.delegate = delegate;
+		}
+
+		@Override
+		public Mono<R> doWith(Document object) {
+			return delegate.doWith(object).map(it -> converter.mapDocument(object, () -> it));
+		}
+	}
+
 	/**
 	 * Simple {@link DocumentCallback} that will transform {@link Document} into the given target type using the given
 	 * {@link EntityReader}.

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

@@ -33,6 +33,7 @@ import org.springframework.data.mongodb.core.aggregation.Aggregation;
  * Unit tests for {@link ExecutableAggregationOperationSupport}.
  *
  * @author Christoph Strobl
+ * @author Mark Paluch
  */
 @ExtendWith(MockitoExtension.class)
 public class ExecutableAggregationOperationSupportUnitTests {
@@ -72,7 +73,8 @@ public class ExecutableAggregationOperationSupportUnitTests {
 		opSupport.aggregateAndReturn(Person.class).inCollection("star-wars").by(newAggregation(project("foo"))).all();
 
 		ArgumentCaptor<Class> captor = ArgumentCaptor.forClass(Class.class);
-		verify(template).aggregate(any(Aggregation.class), eq("star-wars"), captor.capture());
+		verify(template).doAggregate(any(Aggregation.class), eq("star-wars"), captor.capture(),
+				eq(QueryResultConverter.entity()));
 		assertThat(captor.getValue()).isEqualTo(Person.class);
 	}
 
@@ -86,7 +88,8 @@ public class ExecutableAggregationOperationSupportUnitTests {
 		ArgumentCaptor<Class> captor = ArgumentCaptor.forClass(Class.class);
 
 		verify(template).getCollectionName(captor.capture());
-		verify(template).aggregate(any(Aggregation.class), eq("person"), captor.capture());
+		verify(template).doAggregate(any(Aggregation.class), eq("person"), captor.capture(),
+				eq(QueryResultConverter.entity()));
 
 		assertThat(captor.getAllValues()).containsExactly(Person.class, Person.class);
 	}
@@ -101,7 +104,8 @@ public class ExecutableAggregationOperationSupportUnitTests {
 		ArgumentCaptor<Class> captor = ArgumentCaptor.forClass(Class.class);
 
 		verify(template).getCollectionName(captor.capture());
-		verify(template).aggregate(any(Aggregation.class), eq("person"), captor.capture());
+		verify(template).doAggregate(any(Aggregation.class), eq("person"), captor.capture(),
+				eq(QueryResultConverter.entity()));
 
 		assertThat(captor.getAllValues()).containsExactly(Person.class, Jedi.class);
 	}
@@ -112,7 +116,8 @@ public class ExecutableAggregationOperationSupportUnitTests {
 		opSupport.aggregateAndReturn(Person.class).inCollection("star-wars").by(newAggregation(project("foo"))).stream();
 
 		ArgumentCaptor<Class> captor = ArgumentCaptor.forClass(Class.class);
-		verify(template).aggregateStream(any(Aggregation.class), eq("star-wars"), captor.capture());
+		verify(template).doAggregateStream(any(Aggregation.class), eq("star-wars"), captor.capture(),
+				eq(QueryResultConverter.entity()), any());
 		assertThat(captor.getValue()).isEqualTo(Person.class);
 	}
 
@@ -126,7 +131,8 @@ public class ExecutableAggregationOperationSupportUnitTests {
 		ArgumentCaptor<Class> captor = ArgumentCaptor.forClass(Class.class);
 
 		verify(template).getCollectionName(captor.capture());
-		verify(template).aggregateStream(any(Aggregation.class), eq("person"), captor.capture());
+		verify(template).doAggregateStream(any(Aggregation.class), eq("person"), captor.capture(),
+				eq(QueryResultConverter.entity()), any());
 
 		assertThat(captor.getAllValues()).containsExactly(Person.class, Person.class);
 	}
@@ -141,7 +147,8 @@ public class ExecutableAggregationOperationSupportUnitTests {
 		ArgumentCaptor<Class> captor = ArgumentCaptor.forClass(Class.class);
 
 		verify(template).getCollectionName(captor.capture());
-		verify(template).aggregateStream(any(Aggregation.class), eq("person"), captor.capture());
+		verify(template).doAggregateStream(any(Aggregation.class), eq("person"), captor.capture(),
+				eq(QueryResultConverter.entity()), any());
 
 		assertThat(captor.getAllValues()).containsExactly(Person.class, Jedi.class);
 	}

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

@@ -21,7 +21,9 @@ import static org.springframework.data.mongodb.core.query.Query.*;
 import static org.springframework.data.mongodb.test.util.DirtiesStateExtension.*;
 
 import java.util.Date;
+import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.stream.Stream;
 
 import org.bson.BsonString;
@@ -170,6 +172,16 @@ class ExecutableFindOperationSupportTests implements StateFunctions {
 				.hasOnlyElementsOfType(Jedi.class).hasSize(1);
 	}
 
+	@Test // GH-
+	void findAllByWithConverter() {
+
+		List<Optional<Jedi>> result = template.query(Person.class).as(Jedi.class)
+				.matching(query(where("firstname").is("luke"))).map((document, reader) -> Optional.of(reader.get())).all();
+
+		assertThat(result).hasOnlyElementsOfType(Optional.class).hasSize(1);
+		assertThat(result).extracting(Optional::get).hasOnlyElementsOfType(Jedi.class).hasSize(1);
+	}
+
 	@Test // DATAMONGO-1563
 	void findBy() {
 		assertThat(template.query(Person.class).matching(query(where("firstname").is("luke"))).one()).contains(luke);
@@ -260,6 +272,15 @@ class ExecutableFindOperationSupportTests implements StateFunctions {
 		}
 	}
 
+	@Test // GH-
+	void streamAllWithConverter() {
+
+		try (Stream<Optional<Jedi>> stream = template.query(Person.class).as(Jedi.class)
+				.map((document, reader) -> Optional.of(reader.get())).stream()) {
+			assertThat(stream).extracting(Optional::get).hasOnlyElementsOfType(Jedi.class).hasSize(2);
+		}
+	}
+
 	@Test // DATAMONGO-1733
 	void streamAllReturningResultsAsClosedInterfaceProjection() {
 
@@ -315,6 +336,20 @@ class ExecutableFindOperationSupportTests implements StateFunctions {
 		assertThat(results.getContent().get(0).getContent().getId()).isEqualTo("alderan");
 	}
 
+	@Test // GH-
+	void findAllNearByWithConverter() {
+
+		GeoResults<Optional<Human>> results = template.query(Object.class).inCollection(STAR_WARS_PLANETS).as(Human.class)
+				.near(NearQuery.near(-73.9667, 40.78).spherical(true)).map((document, reader) -> Optional.of(reader.get()))
+				.all();
+
+		assertThat(results.getContent()).hasSize(2);
+		assertThat(results.getContent().get(0).getDistance()).isNotNull();
+		assertThat(results.getContent().get(0).getContent()).isInstanceOf(Optional.class);
+		assertThat(results.getContent().get(0).getContent().get()).isInstanceOf(Human.class);
+		assertThat(results.getContent().get(0).getContent().get().getId()).isEqualTo("alderan");
+	}
+
 	@Test // DATAMONGO-1733
 	void findAllNearByReturningGeoResultContentAsClosedInterfaceProjection() {
 

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

@@ -1156,7 +1156,7 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests {
 	void appliesFieldsWhenInterfaceProjectionIsClosedAndQueryDoesNotDefineFields() {
 
 		template.doFind(CollectionPreparer.identity(), "star-wars", new Document(), new Document(), Person.class,
-				PersonProjection.class, CursorPreparer.NO_OP_PREPARER);
+				PersonProjection.class, QueryResultConverter.entity(), CursorPreparer.NO_OP_PREPARER);
 
 		verify(findIterable).projection(eq(new Document("firstname", 1)));
 	}
@@ -1165,7 +1165,7 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests {
 	void doesNotApplyFieldsWhenInterfaceProjectionIsClosedAndQueryDefinesFields() {
 
 		template.doFind(CollectionPreparer.identity(), "star-wars", new Document(), new Document("bar", 1), Person.class,
-				PersonProjection.class, CursorPreparer.NO_OP_PREPARER);
+				PersonProjection.class, QueryResultConverter.entity(), CursorPreparer.NO_OP_PREPARER);
 
 		verify(findIterable).projection(eq(new Document("bar", 1)));
 	}
@@ -1174,7 +1174,7 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests {
 	void doesNotApplyFieldsWhenInterfaceProjectionIsOpen() {
 
 		template.doFind(CollectionPreparer.identity(), "star-wars", new Document(), new Document(), Person.class,
-				PersonSpELProjection.class, CursorPreparer.NO_OP_PREPARER);
+				PersonSpELProjection.class, QueryResultConverter.entity(), CursorPreparer.NO_OP_PREPARER);
 
 		verify(findIterable).projection(eq(BsonUtils.EMPTY_DOCUMENT));
 	}
@@ -1183,7 +1183,7 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests {
 	void appliesFieldsToDtoProjection() {
 
 		template.doFind(CollectionPreparer.identity(), "star-wars", new Document(), new Document(), Person.class,
-				Jedi.class, CursorPreparer.NO_OP_PREPARER);
+				Jedi.class, QueryResultConverter.entity(), CursorPreparer.NO_OP_PREPARER);
 
 		verify(findIterable).projection(eq(new Document("firstname", 1)));
 	}
@@ -1192,7 +1192,7 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests {
 	void doesNotApplyFieldsToDtoProjectionWhenQueryDefinesFields() {
 
 		template.doFind(CollectionPreparer.identity(), "star-wars", new Document(), new Document("bar", 1), Person.class,
-				Jedi.class, CursorPreparer.NO_OP_PREPARER);
+				Jedi.class, QueryResultConverter.entity(), CursorPreparer.NO_OP_PREPARER);
 
 		verify(findIterable).projection(eq(new Document("bar", 1)));
 	}
@@ -1201,7 +1201,7 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests {
 	void doesNotApplyFieldsWhenTargetIsNotAProjection() {
 
 		template.doFind(CollectionPreparer.identity(), "star-wars", new Document(), new Document(), Person.class,
-				Person.class, CursorPreparer.NO_OP_PREPARER);
+				Person.class, QueryResultConverter.entity(), CursorPreparer.NO_OP_PREPARER);
 
 		verify(findIterable).projection(eq(BsonUtils.EMPTY_DOCUMENT));
 	}
@@ -1210,7 +1210,7 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests {
 	void doesNotApplyFieldsWhenTargetExtendsDomainType() {
 
 		template.doFind(CollectionPreparer.identity(), "star-wars", new Document(), new Document(), Person.class,
-				PersonExtended.class, CursorPreparer.NO_OP_PREPARER);
+				PersonExtended.class, QueryResultConverter.entity(), CursorPreparer.NO_OP_PREPARER);
 
 		verify(findIterable).projection(eq(BsonUtils.EMPTY_DOCUMENT));
 	}

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

@@ -72,7 +72,8 @@ public class ReactiveAggregationOperationSupportUnitTests {
 		opSupport.aggregateAndReturn(Person.class).inCollection("star-wars").by(newAggregation(project("foo"))).all();
 
 		ArgumentCaptor<Class> captor = ArgumentCaptor.forClass(Class.class);
-		verify(template).aggregate(any(Aggregation.class), eq("star-wars"), captor.capture());
+		verify(template).doAggregate(any(Aggregation.class), eq("star-wars"), captor.capture(), any(Class.class),
+				eq(QueryResultConverter.entity()));
 		assertThat(captor.getValue()).isEqualTo(Person.class);
 	}
 
@@ -86,7 +87,8 @@ public class ReactiveAggregationOperationSupportUnitTests {
 		ArgumentCaptor<Class> captor = ArgumentCaptor.forClass(Class.class);
 
 		verify(template).getCollectionName(captor.capture());
-		verify(template).aggregate(any(Aggregation.class), eq("person"), captor.capture());
+		verify(template).doAggregate(any(Aggregation.class), eq("person"), captor.capture(), any(Class.class),
+				eq(QueryResultConverter.entity()));
 
 		assertThat(captor.getAllValues()).containsExactly(Person.class, Person.class);
 	}
@@ -101,7 +103,8 @@ public class ReactiveAggregationOperationSupportUnitTests {
 		ArgumentCaptor<Class> captor = ArgumentCaptor.forClass(Class.class);
 
 		verify(template).getCollectionName(captor.capture());
-		verify(template).aggregate(any(Aggregation.class), eq("person"), captor.capture());
+		verify(template).doAggregate(any(Aggregation.class), eq("person"), captor.capture(), any(Class.class),
+				eq(QueryResultConverter.entity()));
 
 		assertThat(captor.getAllValues()).containsExactly(Person.class, Jedi.class);
 	}

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

@@ -26,6 +26,7 @@ import reactor.test.StepVerifier;
 
 import java.util.Date;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -167,6 +168,17 @@ class ReactiveFindOperationSupportTests implements StateFunctions {
 				.verifyComplete();
 	}
 
+	@Test // GH-…
+	void findAllWithConverter() {
+
+		template.query(Person.class).as(Jedi.class).map((document, reader) -> Optional.of(reader.get())).all()
+				.map(Optional::get) //
+				.map(it -> it.getClass().getName()) //
+				.as(StepVerifier::create) //
+				.expectNext(Jedi.class.getName(), Jedi.class.getName()) //
+				.verifyComplete();
+	}
+
 	@Test // DATAMONGO-1719
 	void findAllBy() {
 
@@ -299,6 +311,32 @@ class ReactiveFindOperationSupportTests implements StateFunctions {
 				.verifyComplete();
 	}
 
+	@Test // GH-…
+	@DirtiesState
+	void findAllNearByWithConverter() {
+
+		blocking.indexOps(Planet.class).ensureIndex(
+				new GeospatialIndex("coordinates").typed(GeoSpatialIndexType.GEO_2DSPHERE).named("planet-coordinate-idx"));
+
+		Planet alderan = new Planet("alderan", new Point(-73.9836, 40.7538));
+		Planet dantooine = new Planet("dantooine", new Point(-73.9928, 40.7193));
+
+		blocking.save(alderan);
+		blocking.save(dantooine);
+
+		template.query(Object.class).inCollection(STAR_WARS).as(Human.class)
+				.near(NearQuery.near(-73.9667, 40.78).spherical(true)).map((document, reader) -> Optional.of(reader.get())) //
+				.all() //
+				.as(StepVerifier::create).consumeNextWith(actual -> {
+					assertThat(actual.getDistance()).isNotNull();
+					assertThat(actual.getContent()).isInstanceOf(Optional.class);
+					assertThat(actual.getContent().get()).isInstanceOf(Human.class);
+					assertThat(actual.getContent().get().getId()).isEqualTo("alderan");
+				}) //
+				.expectNextCount(1) //
+				.verifyComplete();
+	}
+
 	@Test // DATAMONGO-1719
 	@DirtiesState
 	void findAllNearByReturningGeoResultContentAsClosedInterfaceProjection() {

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

@@ -18,6 +18,7 @@ package org.springframework.data.mongodb.core;
 import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
 import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;
+import static org.springframework.data.mongodb.test.util.Assertions.*;
 import static org.springframework.data.mongodb.test.util.Assertions.assertThat;
 
 import reactor.core.publisher.Flux;
@@ -54,6 +55,7 @@ import org.mockito.junit.jupiter.MockitoSettings;
 import org.mockito.quality.Strictness;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
+
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationListener;
@@ -437,7 +439,7 @@ public class ReactiveMongoTemplateUnitTests {
 	void appliesFieldsWhenInterfaceProjectionIsClosedAndQueryDoesNotDefineFields() {
 
 		template.doFind("star-wars", CollectionPreparer.identity(), new Document(), new Document(), Person.class,
-				PersonProjection.class, FindPublisherPreparer.NO_OP_PREPARER).subscribe();
+				PersonProjection.class, QueryResultConverter.entity(), FindPublisherPreparer.NO_OP_PREPARER).subscribe();
 
 		verify(findPublisher).projection(eq(new Document("firstname", 1)));
 	}
@@ -446,7 +448,7 @@ public class ReactiveMongoTemplateUnitTests {
 	void doesNotApplyFieldsWhenInterfaceProjectionIsClosedAndQueryDefinesFields() {
 
 		template.doFind("star-wars", CollectionPreparer.identity(), new Document(), new Document("bar", 1), Person.class,
-				PersonProjection.class, FindPublisherPreparer.NO_OP_PREPARER).subscribe();
+				PersonProjection.class, QueryResultConverter.entity(), FindPublisherPreparer.NO_OP_PREPARER).subscribe();
 
 		verify(findPublisher).projection(eq(new Document("bar", 1)));
 	}
@@ -455,7 +457,7 @@ public class ReactiveMongoTemplateUnitTests {
 	void doesNotApplyFieldsWhenInterfaceProjectionIsOpen() {
 
 		template.doFind("star-wars", CollectionPreparer.identity(), new Document(), new Document(), Person.class,
-				PersonSpELProjection.class, FindPublisherPreparer.NO_OP_PREPARER).subscribe();
+				PersonSpELProjection.class, QueryResultConverter.entity(), FindPublisherPreparer.NO_OP_PREPARER).subscribe();
 
 		verify(findPublisher, never()).projection(any());
 	}
@@ -464,7 +466,7 @@ public class ReactiveMongoTemplateUnitTests {
 	void appliesFieldsToDtoProjection() {
 
 		template.doFind("star-wars", CollectionPreparer.identity(), new Document(), new Document(), Person.class,
-				Jedi.class, FindPublisherPreparer.NO_OP_PREPARER).subscribe();
+				Jedi.class, QueryResultConverter.entity(), FindPublisherPreparer.NO_OP_PREPARER).subscribe();
 
 		verify(findPublisher).projection(eq(new Document("firstname", 1)));
 	}
@@ -473,7 +475,7 @@ public class ReactiveMongoTemplateUnitTests {
 	void doesNotApplyFieldsToDtoProjectionWhenQueryDefinesFields() {
 
 		template.doFind("star-wars", CollectionPreparer.identity(), new Document(), new Document("bar", 1), Person.class,
-				Jedi.class, FindPublisherPreparer.NO_OP_PREPARER).subscribe();
+				Jedi.class, QueryResultConverter.entity(), FindPublisherPreparer.NO_OP_PREPARER).subscribe();
 
 		verify(findPublisher).projection(eq(new Document("bar", 1)));
 	}
@@ -482,7 +484,7 @@ public class ReactiveMongoTemplateUnitTests {
 	void doesNotApplyFieldsWhenTargetIsNotAProjection() {
 
 		template.doFind("star-wars", CollectionPreparer.identity(), new Document(), new Document(), Person.class,
-				Person.class, FindPublisherPreparer.NO_OP_PREPARER).subscribe();
+				Person.class, QueryResultConverter.entity(), FindPublisherPreparer.NO_OP_PREPARER).subscribe();
 
 		verify(findPublisher, never()).projection(any());
 	}
@@ -491,7 +493,7 @@ public class ReactiveMongoTemplateUnitTests {
 	void doesNotApplyFieldsWhenTargetExtendsDomainType() {
 
 		template.doFind("star-wars", CollectionPreparer.identity(), new Document(), new Document(), Person.class,
-				PersonExtended.class, FindPublisherPreparer.NO_OP_PREPARER).subscribe();
+				PersonExtended.class, QueryResultConverter.entity(), FindPublisherPreparer.NO_OP_PREPARER).subscribe();
 
 		verify(findPublisher, never()).projection(any());
 	}

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

@@ -37,6 +37,7 @@ import java.util.Collections;
 import java.util.Date;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Scanner;
 import java.util.stream.Stream;
 
@@ -287,6 +288,60 @@ public class AggregationTests {
 		}
 	}
 
+	@Test // GH-
+	void shouldAggregateAsStreamWithConverter() {
+
+		MongoCollection<Document> coll = mongoTemplate.getCollection(INPUT_COLLECTION);
+
+		coll.insertOne(createDocument("Doc1", "spring", "mongodb", "nosql"));
+		coll.insertOne(createDocument("Doc2"));
+
+		Aggregation aggregation = newAggregation(//
+				project("tags"), //
+				unwind("tags"), //
+				group("tags") //
+						.count().as("n"), //
+				project("n") //
+						.and("tag").previousOperation(), //
+				sort(DESC, "n") //
+		);
+
+		try (Stream<Optional<TagCount>> stream = mongoTemplate.aggregateAndReturn(TagCount.class)
+				.inCollection(INPUT_COLLECTION).by(aggregation).map((document, reader) -> Optional.of(reader.get())).stream()) {
+
+			List<TagCount> tagCount = stream.flatMap(Optional::stream).toList();
+
+			assertThat(tagCount).hasSize(3);
+		}
+	}
+
+	@Test // GH-
+	void shouldAggregateWithConverter() {
+
+		MongoCollection<Document> coll = mongoTemplate.getCollection(INPUT_COLLECTION);
+
+		coll.insertOne(createDocument("Doc1", "spring", "mongodb", "nosql"));
+		coll.insertOne(createDocument("Doc2"));
+
+		Aggregation aggregation = newAggregation(//
+				project("tags"), //
+				unwind("tags"), //
+				group("tags") //
+						.count().as("n"), //
+				project("n") //
+						.and("tag").previousOperation(), //
+				sort(DESC, "n") //
+		);
+
+		AggregationResults<Optional<TagCount>> results = mongoTemplate.aggregateAndReturn(TagCount.class)
+				.inCollection(INPUT_COLLECTION) //
+				.by(aggregation) //
+				.map((document, reader) -> Optional.of(reader.get())) //
+				.all();
+
+		assertThat(results.getMappedResults()).extracting(Optional::get).hasOnlyElementsOfType(TagCount.class).hasSize(3);
+	}
+
 	@Test // DATAMONGO-1391
 	void shouldUnwindWithIndex() {
 
@@ -501,7 +556,7 @@ public class AggregationTests {
 		/*
 		 //complex mongodb aggregation framework example from
 		 https://docs.mongodb.org/manual/tutorial/aggregation-examples/#largest-and-smallest-cities-by-state
-		
+
 		 db.zipcodes.aggregate(
 			 	{
 				   $group: {

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

@@ -22,6 +22,7 @@ import static org.springframework.data.mongodb.core.query.Criteria.*;
 import reactor.test.StepVerifier;
 
 import java.util.Arrays;
+import java.util.Optional;
 
 import org.bson.Document;
 import org.junit.After;
@@ -115,6 +116,29 @@ public class ReactiveAggregationTests {
 				}).verifyComplete();
 	}
 
+	@Test // GH-…
+	public void shouldProjectAndConvertMultipleDocuments() {
+
+		City dresden = new City("Dresden", 100);
+		City linz = new City("Linz", 101);
+		City braunschweig = new City("Braunschweig", 102);
+		City weinheim = new City("Weinheim", 103);
+
+		reactiveMongoTemplate.insertAll(Arrays.asList(dresden, linz, braunschweig, weinheim)).as(StepVerifier::create)
+				.expectNextCount(4).verifyComplete();
+
+		Aggregation agg = newAggregation( //
+				match(where("population").lt(103)));
+
+		reactiveMongoTemplate.aggregateAndReturn(City.class).inCollection("city").by(agg)
+				.map((document, reader) -> Optional.of(reader.get())) //
+				.all() //
+				.collectList() //
+				.as(StepVerifier::create).consumeNextWith(actual -> {
+					assertThat(actual).hasSize(3).extracting(Optional::get).contains(dresden, linz, braunschweig);
+				}).verifyComplete();
+	}
+
 	@Test // DATAMONGO-1646
 	public void shouldAggregateToOutCollection() {