Add support for fluent Querydsl and Query by Example query definition.

We now support the functional fluent query definition API for imperative and reactive usage with Querydsl and Query by Example.

Page<PersonProjection> first = repository.findBy(Example.of(probe),
		it -> it.as(PersonProjection.class).project("firstname").page(PageRequest.of(0, 1, Sort.by("firstname"))));

Closes #3757
Original pull request: #3788.

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

@@ -1385,7 +1385,6 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 		return source.isVersionedEntity() //
 				? doSaveVersioned(source, collectionName) //
 				: (T) doSave(collectionName, objectToSave, this.mongoConverter);
-
 	}
 
 	@SuppressWarnings("unchecked")

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

@@ -164,9 +164,9 @@ public class BasicQuery extends Query {
 	 * @throws IllegalArgumentException when {@code fieldsObject} is {@literal null}.
 	 * @since 1.6
 	 */
-	protected void setFieldsObject(Document fieldsObject) {
+	public void setFieldsObject(Document fieldsObject) {
 
-		Assert.notNull(sortObject, "Field document must not be null");
+		Assert.notNull(fieldsObject, "Field document must not be null");
 
 		this.fieldsObject = fieldsObject;
 	}

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

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2021 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.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.springframework.data.domain.Sort;
+import org.springframework.data.repository.query.FluentQuery;
+import org.springframework.util.Assert;
+
+/**
+ * Support class for {@link org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery} implementations.
+ *
+ * @author Mark Paluch
+ * @since 3.3
+ */
+abstract class FetchableFluentQuerySupport<P, T> implements FluentQuery.FetchableFluentQuery<T> {
+
+	private final P predicate;
+	private final Sort sort;
+	private final Class<T> resultType;
+	private final List<String> fieldsToInclude;
+
+	FetchableFluentQuerySupport(P predicate, Sort sort, Class<T> resultType, List<String> fieldsToInclude) {
+		this.predicate = predicate;
+		this.sort = sort;
+		this.resultType = resultType;
+		this.fieldsToInclude = fieldsToInclude;
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#sortBy(org.springframework.data.domain.Sort)
+	 */
+	@Override
+	public FluentQuery.FetchableFluentQuery<T> sortBy(Sort sort) {
+
+		Assert.notNull(sort, "Sort must not be null!");
+
+		return create(predicate, sort, resultType, fieldsToInclude);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#as(java.lang.Class)
+	 */
+	@Override
+	public <R> FluentQuery.FetchableFluentQuery<R> as(Class<R> projection) {
+
+		Assert.notNull(projection, "Projection target type must not be null!");
+
+		return create(predicate, sort, projection, fieldsToInclude);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#project(java.util.Collection)
+	 */
+	@Override
+	public FluentQuery.FetchableFluentQuery<T> project(Collection<String> properties) {
+
+		Assert.notNull(properties, "Projection properties must not be null!");
+
+		return create(predicate, sort, resultType, new ArrayList<>(properties));
+	}
+
+	protected abstract <R> FetchableFluentQuerySupport<P, R> create(P predicate, Sort sort, Class<R> resultType,
+			List<String> fieldsToInclude);
+
+	P getPredicate() {
+		return predicate;
+	}
+
+	Sort getSort() {
+		return sort;
+	}
+
+	Class<T> getResultType() {
+		return resultType;
+	}
+
+	List<String> getFieldsToInclude() {
+		return fieldsToInclude;
+	}
+}

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

@@ -15,18 +15,24 @@
  */
 package org.springframework.data.mongodb.repository.support;
 
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Stream;
 
+import org.bson.Document;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mongodb.core.MongoOperations;
+import org.springframework.data.mongodb.core.query.BasicQuery;
 import org.springframework.data.mongodb.repository.query.MongoEntityInformation;
 import org.springframework.data.querydsl.EntityPathResolver;
 import org.springframework.data.querydsl.QuerydslPredicateExecutor;
 import org.springframework.data.querydsl.SimpleEntityPathResolver;
+import org.springframework.data.repository.query.FluentQuery;
 import org.springframework.data.support.PageableExecutionUtils;
 import org.springframework.util.Assert;
 
@@ -184,6 +190,21 @@ public class QuerydslMongoPredicateExecutor<T> extends QuerydslPredicateExecutor
 		return createQueryFor(predicate).fetchCount() > 0;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.querydsl.QuerydslPredicateExecutor#findBy(com.querydsl.core.types.Predicate, java.util.function.Function)
+	 */
+	@Override
+	@SuppressWarnings("unchecked")
+	public <S extends T, R> R findBy(Predicate predicate,
+			Function<FluentQuery.FetchableFluentQuery<S>, R> queryFunction) {
+
+		Assert.notNull(predicate, "Predicate must not be null!");
+		Assert.notNull(queryFunction, "Query function must not be null!");
+
+		return queryFunction.apply(new FluentQuerydsl<>(predicate, (Class<S>) typeInformation().getJavaType()));
+	}
+
 	/**
 	 * Creates a {@link SpringDataMongodbQuery} for the given {@link Predicate}.
 	 *
@@ -232,4 +253,113 @@ public class QuerydslMongoPredicateExecutor<T> extends QuerydslPredicateExecutor
 		toOrderSpecifiers(sort).forEach(query::orderBy);
 		return query;
 	}
+
+	/**
+	 * {@link org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery} using Querydsl
+	 * {@link Predicate}.
+	 *
+	 * @author Mark Paluch
+	 * @since 3.3
+	 */
+	class FluentQuerydsl<T> extends FetchableFluentQuerySupport<Predicate, T> {
+
+		FluentQuerydsl(Predicate predicate, Class<T> resultType) {
+			this(predicate, Sort.unsorted(), resultType, Collections.emptyList());
+		}
+
+		FluentQuerydsl(Predicate predicate, Sort sort, Class<T> resultType, List<String> fieldsToInclude) {
+			super(predicate, sort, resultType, fieldsToInclude);
+		}
+
+		@Override
+		protected <R> FluentQuerydsl<R> create(Predicate predicate, Sort sort, Class<R> resultType,
+				List<String> fieldsToInclude) {
+			return new FluentQuerydsl<>(predicate, sort, resultType, fieldsToInclude);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#oneValue()
+		 */
+		@Override
+		public T oneValue() {
+			return createQuery().fetchOne();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#firstValue()
+		 */
+		@Override
+		public T firstValue() {
+			return createQuery().fetchFirst();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#all()
+		 */
+		@Override
+		public List<T> all() {
+			return createQuery().fetch();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#page(org.springframework.data.domain.Pageable)
+		 */
+		@Override
+		public Page<T> page(Pageable pageable) {
+
+			Assert.notNull(pageable, "Pageable must not be null!");
+
+			return createQuery().fetchPage(pageable);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#stream()
+		 */
+		@Override
+		public Stream<T> stream() {
+			return createQuery().stream();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#count()
+		 */
+		@Override
+		public long count() {
+			return createQuery().fetchCount();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#exists()
+		 */
+		@Override
+		public boolean exists() {
+			return count() > 0;
+		}
+
+		private SpringDataMongodbQuery<T> createQuery() {
+			return new SpringDataMongodbQuery<>(mongoOperations, typeInformation().getJavaType(), getResultType(),
+					mongoOperations.getCollectionName(typeInformation().getJavaType()), this::customize).where(getPredicate());
+		}
+
+		private void customize(BasicQuery query) {
+
+			List<String> fieldsToInclude = getFieldsToInclude();
+			if (!fieldsToInclude.isEmpty()) {
+				Document fields = new Document();
+				fieldsToInclude.forEach(field -> fields.put(field, 1));
+				query.setFieldsObject(fields);
+			}
+
+			if (getSort().isSorted()) {
+				query.with(getSort());
+			}
+		}
+	}
 }

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

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2021 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.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.springframework.data.domain.Sort;
+import org.springframework.data.repository.query.FluentQuery;
+import org.springframework.util.Assert;
+
+/**
+ * Support class for {@link org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery} implementations.
+ *
+ * @author Mark Paluch
+ * @since 3.3
+ */
+abstract class ReactiveFluentQuerySupport<P, T> implements FluentQuery.ReactiveFluentQuery<T> {
+
+	private final P predicate;
+	private final Sort sort;
+	private final Class<T> resultType;
+	private final List<String> fieldsToInclude;
+
+	ReactiveFluentQuerySupport(P predicate, Sort sort, Class<T> resultType, List<String> fieldsToInclude) {
+		this.predicate = predicate;
+		this.sort = sort;
+		this.resultType = resultType;
+		this.fieldsToInclude = fieldsToInclude;
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#sortBy(org.springframework.data.domain.Sort)
+	 */
+	@Override
+	public ReactiveFluentQuery<T> sortBy(Sort sort) {
+
+		Assert.notNull(sort, "Sort must not be null!");
+
+		return create(predicate, sort, resultType, fieldsToInclude);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#as(java.lang.Class)
+	 */
+	@Override
+	public <R> ReactiveFluentQuery<R> as(Class<R> projection) {
+
+		Assert.notNull(projection, "Projection target type must not be null!");
+
+		return create(predicate, sort, projection, fieldsToInclude);
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#project(java.util.Collection)
+	 */
+	@Override
+	public ReactiveFluentQuery<T> project(Collection<String> properties) {
+
+		Assert.notNull(properties, "Projection properties must not be null!");
+
+		return create(predicate, sort, resultType, new ArrayList<>(properties));
+	}
+
+	protected abstract <R> ReactiveFluentQuerySupport<P, R> create(P predicate, Sort sort, Class<R> resultType,
+			List<String> fieldsToInclude);
+
+	P getPredicate() {
+		return predicate;
+	}
+
+	Sort getSort() {
+		return sort;
+	}
+
+	Class<T> getResultType() {
+		return resultType;
+	}
+
+	List<String> getFieldsToInclude() {
+		return fieldsToInclude;
+	}
+}

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

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2021 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 reactor.core.publisher.Mono;
+
+import java.util.List;
+
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageImpl;
+import org.springframework.data.domain.Pageable;
+import org.springframework.util.Assert;
+
+/**
+ * Support for query execution using {@link Pageable}. Using {@link ReactivePageableExecutionUtils} assumes that data
+ * queries are cheaper than {@code COUNT} queries and so some cases can take advantage of optimizations.
+ *
+ * @author Mark Paluch
+ * @since 3.3
+ */
+abstract class ReactivePageableExecutionUtils {
+
+	private ReactivePageableExecutionUtils() {}
+
+	/**
+	 * Constructs a {@link Page} based on the given {@code content}, {@link Pageable} and {@link Mono} applying
+	 * optimizations. The construction of {@link Page} omits a count query if the total can be determined based on the
+	 * result size and {@link Pageable}.
+	 *
+	 * @param content must not be {@literal null}.
+	 * @param pageable must not be {@literal null}.
+	 * @param totalSupplier must not be {@literal null}.
+	 * @return the {@link Page}.
+	 */
+	public static <T> Mono<Page<T>> getPage(List<T> content, Pageable pageable, Mono<Long> totalSupplier) {
+
+		Assert.notNull(content, "Content must not be null!");
+		Assert.notNull(pageable, "Pageable must not be null!");
+		Assert.notNull(totalSupplier, "TotalSupplier must not be null!");
+
+		if (pageable.isUnpaged() || pageable.getOffset() == 0) {
+
+			if (pageable.isUnpaged() || pageable.getPageSize() > content.size()) {
+				return Mono.just(new PageImpl<>(content, pageable, content.size()));
+			}
+
+			return totalSupplier.map(total -> new PageImpl<>(content, pageable, total));
+		}
+
+		if (content.size() != 0 && pageable.getPageSize() > content.size()) {
+			return Mono.just(new PageImpl<>(content, pageable, pageable.getOffset() + content.size()));
+		}
+
+		return totalSupplier.map(total -> new PageImpl<>(content, pageable, total));
+	}
+}

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

@@ -18,13 +18,23 @@ package org.springframework.data.mongodb.repository.support;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+
+import org.bson.Document;
+import org.reactivestreams.Publisher;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mongodb.core.ReactiveMongoOperations;
+import org.springframework.data.mongodb.core.query.BasicQuery;
 import org.springframework.data.mongodb.repository.query.MongoEntityInformation;
 import org.springframework.data.querydsl.EntityPathResolver;
 import org.springframework.data.querydsl.QuerydslPredicateExecutor;
 import org.springframework.data.querydsl.ReactiveQuerydslPredicateExecutor;
 import org.springframework.data.querydsl.SimpleEntityPathResolver;
+import org.springframework.data.repository.query.FluentQuery;
 import org.springframework.util.Assert;
 
 import com.querydsl.core.types.EntityPath;
@@ -159,6 +169,20 @@ public class ReactiveQuerydslMongoPredicateExecutor<T> extends QuerydslPredicate
 		return createQueryFor(predicate).fetchCount().map(it -> it != 0);
 	}
 
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.querydsl.ReactiveQuerydslPredicateExecutor#findBy(com.querydsl.core.types.Predicate, java.util.function.Function)
+	 */
+	@Override
+	public <S extends T, R, P extends Publisher<R>> P findBy(Predicate predicate,
+			Function<FluentQuery.ReactiveFluentQuery<S>, P> queryFunction) {
+
+		Assert.notNull(predicate, "Predicate must not be null!");
+		Assert.notNull(queryFunction, "Query function must not be null!");
+
+		return queryFunction.apply(new ReactiveFluentQuerydsl<S>(predicate, (Class<S>) typeInformation().getJavaType()));
+	}
+
 	/**
 	 * Creates a {@link ReactiveSpringDataMongodbQuery} for the given {@link Predicate}.
 	 *
@@ -177,8 +201,8 @@ public class ReactiveQuerydslMongoPredicateExecutor<T> extends QuerydslPredicate
 	private ReactiveSpringDataMongodbQuery<T> createQuery() {
 
 		Class<T> javaType = typeInformation().getJavaType();
-		return new ReactiveSpringDataMongodbQuery<>(mongodbSerializer(), mongoOperations, javaType,
-				mongoOperations.getCollectionName(javaType));
+		return new ReactiveSpringDataMongodbQuery<>(mongoOperations, javaType, javaType,
+				mongoOperations.getCollectionName(javaType), it -> {});
 	}
 
 	/**
@@ -194,4 +218,105 @@ public class ReactiveQuerydslMongoPredicateExecutor<T> extends QuerydslPredicate
 		return query;
 	}
 
+	/**
+	 * {@link org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery} using Querydsl {@link Predicate}.
+	 *
+	 * @since 3.3
+	 * @author Mark Paluch
+	 */
+	class ReactiveFluentQuerydsl<T> extends ReactiveFluentQuerySupport<Predicate, T> {
+
+		ReactiveFluentQuerydsl(Predicate predicate, Class<T> resultType) {
+			this(predicate, Sort.unsorted(), resultType, Collections.emptyList());
+		}
+
+		ReactiveFluentQuerydsl(Predicate predicate, Sort sort, Class<T> resultType, List<String> fieldsToInclude) {
+			super(predicate, sort, resultType, fieldsToInclude);
+		}
+
+		@Override
+		protected <R> ReactiveFluentQuerydsl<R> create(Predicate predicate, Sort sort, Class<R> resultType,
+				List<String> fieldsToInclude) {
+			return new ReactiveFluentQuerydsl<>(predicate, sort, resultType, fieldsToInclude);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#one()
+		 */
+		@Override
+		public Mono<T> one() {
+			return createQuery().fetchOne();
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#first()
+		 */
+		@Override
+		public Mono<T> first() {
+			return createQuery().fetchFirst();
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#all()
+		 */
+		@Override
+		public Flux<T> all() {
+			return createQuery().fetch();
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#page(org.springframework.data.domain.Pageable)
+		 */
+		@Override
+		public Mono<Page<T>> page(Pageable pageable) {
+
+			Assert.notNull(pageable, "Pageable must not be null!");
+
+			return createQuery().fetchPage(pageable);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#count()
+		 */
+		@Override
+		public Mono<Long> count() {
+			return createQuery().fetchCount();
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#exists()
+		 */
+		@Override
+		public Mono<Boolean> exists() {
+			return count().map(it -> it > 0).defaultIfEmpty(false);
+		}
+
+		private ReactiveSpringDataMongodbQuery<T> createQuery() {
+
+			return new ReactiveSpringDataMongodbQuery<>(mongoOperations, typeInformation().getJavaType(), getResultType(),
+					mongoOperations.getCollectionName(typeInformation().getJavaType()), this::customize).where(getPredicate());
+		}
+
+		private void customize(BasicQuery query) {
+
+			List<String> fieldsToInclude = getFieldsToInclude();
+
+			if (!fieldsToInclude.isEmpty()) {
+				Document fields = new Document();
+				fieldsToInclude.forEach(field -> fields.put(field, 1));
+				query.setFieldsObject(fields);
+			}
+
+			if (getSort().isSorted()) {
+				query.with(getSort());
+			}
+		}
+	}
+
 }

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

@@ -21,11 +21,14 @@ import reactor.core.publisher.Mono;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.function.Consumer;
 
 import org.bson.Document;
 
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
 import org.springframework.data.mongodb.core.MongoOperations;
-import org.springframework.data.mongodb.core.ReactiveFindOperation.FindWithProjection;
+import org.springframework.data.mongodb.core.ReactiveFindOperation;
 import org.springframework.data.mongodb.core.ReactiveMongoOperations;
 import org.springframework.data.mongodb.core.query.BasicQuery;
 import org.springframework.data.mongodb.core.query.Query;
@@ -44,7 +47,6 @@ import com.querydsl.core.types.OrderSpecifier;
 import com.querydsl.core.types.Path;
 import com.querydsl.core.types.Predicate;
 import com.querydsl.mongodb.MongodbOps;
-import com.querydsl.mongodb.document.MongodbDocumentSerializer;
 
 /**
  * MongoDB query with utilizing {@link ReactiveMongoOperations} for command execution.
@@ -59,21 +61,23 @@ import com.querydsl.mongodb.document.MongodbDocumentSerializer;
 class ReactiveSpringDataMongodbQuery<K> extends SpringDataMongodbQuerySupport<ReactiveSpringDataMongodbQuery<K>> {
 
 	private final ReactiveMongoOperations mongoOperations;
-	private final FindWithProjection<K> find;
+	private final Consumer<BasicQuery> queryCustomizer;
+	private final ReactiveFindOperation.FindWithQuery<K> find;
 
 	ReactiveSpringDataMongodbQuery(ReactiveMongoOperations mongoOperations, Class<? extends K> entityClass) {
-		this(new SpringDataMongodbSerializer(mongoOperations.getConverter()), mongoOperations, entityClass, null);
+		this(mongoOperations, entityClass, entityClass, null, it -> {});
 	}
 
 	@SuppressWarnings("unchecked")
-	ReactiveSpringDataMongodbQuery(MongodbDocumentSerializer serializer, ReactiveMongoOperations mongoOperations,
-			Class<? extends K> entityClass, @Nullable String collection) {
+	ReactiveSpringDataMongodbQuery(ReactiveMongoOperations mongoOperations, Class<?> domainType,
+			Class<? extends K> resultType, @Nullable String collection, Consumer<BasicQuery> queryCustomizer) {
 
-		super(serializer);
+		super(new SpringDataMongodbSerializer(mongoOperations.getConverter()));
 
 		this.mongoOperations = mongoOperations;
-		this.find = StringUtils.hasText(collection) ? mongoOperations.query((Class<K>) entityClass).inCollection(collection)
-				: mongoOperations.query((Class<K>) entityClass);
+		this.queryCustomizer = queryCustomizer;
+		this.find = (StringUtils.hasText(collection) ? mongoOperations.query(domainType).inCollection(collection)
+				: mongoOperations.query(domainType)).as((Class<K>) resultType);
 	}
 
 	/**
@@ -86,7 +90,19 @@ class ReactiveSpringDataMongodbQuery<K> extends SpringDataMongodbQuerySupport<Re
 	}
 
 	/**
-	 * Fetch the first matching query result.
+	 * Fetch all matching query results as page.
+	 *
+	 * @return {@link Mono} emitting the requested page.
+	 */
+	Mono<Page<K>> fetchPage(Pageable pageable) {
+
+		Mono<List<K>> content = createQuery().flatMapMany(it -> find.matching(it).all()).collectList();
+
+		return content.flatMap(it -> ReactivePageableExecutionUtils.getPage(it, pageable, fetchCount()));
+	}
+
+	/**
+	 * Fetch the one matching query result.
 	 *
 	 * @return {@link Mono} emitting the first query result or {@link Mono#empty()} if there are none.
 	 * @throws org.springframework.dao.IncorrectResultSizeDataAccessException if more than one match found.
@@ -95,6 +111,16 @@ class ReactiveSpringDataMongodbQuery<K> extends SpringDataMongodbQuerySupport<Re
 		return createQuery().flatMap(it -> find.matching(it).one());
 	}
 
+	/**
+	 * Fetch the first matching query result. @return {@link Mono} emitting the first query result or {@link Mono#empty()}
+	 * if there are none.
+	 *
+	 * @since 3.3
+	 */
+	Mono<K> fetchFirst() {
+		return createQuery().flatMap(it -> find.matching(it).first());
+	}
+
 	/**
 	 * Fetch the count of matching query results.
 	 *
@@ -144,6 +170,8 @@ class ReactiveSpringDataMongodbQuery<K> extends SpringDataMongodbQuerySupport<Re
 						basicQuery.setSortObject(createSort(orderBy));
 					}
 
+					queryCustomizer.accept(basicQuery);
+
 					return basicQuery;
 				});
 	}

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

@@ -22,7 +22,10 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Function;
+import java.util.function.UnaryOperator;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.data.domain.Example;
@@ -30,6 +33,7 @@ import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageImpl;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.mongodb.core.ExecutableFindOperation;
 import org.springframework.data.mongodb.core.MongoOperations;
 import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.data.mongodb.core.query.Criteria;
@@ -211,8 +215,7 @@ public class SimpleMongoRepository<T, ID> implements MongoRepository<T, ID> {
 
 		Assert.notNull(ids, "The given Iterable of ids must not be null!");
 
-		mongoOperations.remove(getIdQuery(ids), entityInformation.getJavaType(),
-				entityInformation.getCollectionName());
+		mongoOperations.remove(getIdQuery(ids), entityInformation.getJavaType(), entityInformation.getCollectionName());
 	}
 
 	/*
@@ -362,8 +365,8 @@ public class SimpleMongoRepository<T, ID> implements MongoRepository<T, ID> {
 
 		List<S> list = mongoOperations.find(query, example.getProbeType(), entityInformation.getCollectionName());
 
-		return PageableExecutionUtils.getPage(list, pageable,
-				() -> mongoOperations.count(Query.of(query).limit(-1).skip(-1), example.getProbeType(), entityInformation.getCollectionName()));
+		return PageableExecutionUtils.getPage(list, pageable, () -> mongoOperations
+				.count(Query.of(query).limit(-1).skip(-1), example.getProbeType(), entityInformation.getCollectionName()));
 	}
 
 	/*
@@ -396,6 +399,20 @@ public class SimpleMongoRepository<T, ID> implements MongoRepository<T, ID> {
 		return mongoOperations.exists(query, example.getProbeType(), entityInformation.getCollectionName());
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.QueryByExampleExecutor#findBy(org.springframework.data.domain.Example, java.util.function.Function)
+	 */
+	@Override
+	public <S extends T, R> R findBy(Example<S> example,
+			Function<org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery<S>, R> queryFunction) {
+
+		Assert.notNull(example, "Sample must not be null!");
+		Assert.notNull(queryFunction, "Query function must not be null!");
+
+		return queryFunction.apply(new FluentQueryByExample<>(example, example.getProbeType()));
+	}
+
 	// -------------------------------------------------------------------------
 	// Utility methods
 	// -------------------------------------------------------------------------
@@ -410,8 +427,7 @@ public class SimpleMongoRepository<T, ID> implements MongoRepository<T, ID> {
 
 	private Query getIdQuery(Iterable<? extends ID> ids) {
 
-		return new Query(new Criteria(entityInformation.getIdAttribute())
-				.in(toCollection(ids)));
+		return new Query(new Criteria(entityInformation.getIdAttribute()).in(toCollection(ids)));
 	}
 
 	private static <E> Collection<E> toCollection(Iterable<E> ids) {
@@ -428,4 +444,119 @@ public class SimpleMongoRepository<T, ID> implements MongoRepository<T, ID> {
 		return mongoOperations.find(query, entityInformation.getJavaType(), entityInformation.getCollectionName());
 	}
 
+	/**
+	 * {@link org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery} using {@link Example}.
+	 *
+	 * @author Mark Paluch
+	 * @since 3.3
+	 */
+	class FluentQueryByExample<S, T> extends FetchableFluentQuerySupport<Example<S>, T> {
+
+		FluentQueryByExample(Example<S> example, Class<T> resultType) {
+			this(example, Sort.unsorted(), resultType, Collections.emptyList());
+		}
+
+		FluentQueryByExample(Example<S> example, Sort sort, Class<T> resultType, List<String> fieldsToInclude) {
+			super(example, sort, resultType, fieldsToInclude);
+		}
+
+		@Override
+		protected <R> FluentQueryByExample<S, R> create(Example<S> predicate, Sort sort, Class<R> resultType,
+				List<String> fieldsToInclude) {
+			return new FluentQueryByExample<>(predicate, sort, resultType, fieldsToInclude);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#oneValue()
+		 */
+		@Override
+		public T oneValue() {
+			return createQuery().oneValue();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#first()
+		 */
+		@Override
+		public T firstValue() {
+			return createQuery().firstValue();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#all()
+		 */
+		@Override
+		public List<T> all() {
+			return createQuery().all();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#page(org.springframework.data.domain.Pageable)
+		 */
+		@Override
+		public Page<T> page(Pageable pageable) {
+
+			Assert.notNull(pageable, "Pageable must not be null!");
+
+			List<T> list = createQuery(q -> q.with(pageable)).all();
+
+			return PageableExecutionUtils.getPage(list, pageable, this::count);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#stream()
+		 */
+		@Override
+		public Stream<T> stream() {
+			return createQuery().stream();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#count()
+		 */
+		@Override
+		public long count() {
+			return createQuery().count();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery#exists()
+		 */
+		@Override
+		public boolean exists() {
+			return createQuery().exists();
+		}
+
+		private ExecutableFindOperation.TerminatingFind<T> createQuery() {
+			return createQuery(UnaryOperator.identity());
+		}
+
+		private ExecutableFindOperation.TerminatingFind<T> createQuery(UnaryOperator<Query> queryCustomizer) {
+
+			Query query = new Query(new Criteria().alike(getPredicate())) //
+					.collation(entityInformation.getCollation());
+
+			if (getSort().isSorted()) {
+				query.with(getSort());
+			}
+
+			if (!getFieldsToInclude().isEmpty()) {
+				query.fields().include(getFieldsToInclude().toArray(new String[0]));
+			}
+
+			query = queryCustomizer.apply(query);
+
+			return mongoOperations.query(getPredicate().getProbeType()).inCollection(entityInformation.getCollectionName())
+					.as(getResultType()).matching(query);
+		}
+
+	}
+
 }

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

@@ -22,19 +22,26 @@ import reactor.core.publisher.Mono;
 
 import java.io.Serializable;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+import java.util.function.UnaryOperator;
 import java.util.stream.Collectors;
 
 import org.reactivestreams.Publisher;
-
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.data.domain.Example;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.mongodb.core.ReactiveFindOperation;
 import org.springframework.data.mongodb.core.ReactiveMongoOperations;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
 import org.springframework.data.mongodb.repository.query.MongoEntityInformation;
+import org.springframework.data.repository.query.FluentQuery;
 import org.springframework.data.util.StreamUtils;
 import org.springframework.data.util.Streamable;
 import org.springframework.util.Assert;
@@ -210,7 +217,6 @@ public class SimpleReactiveMongoRepository<T, ID extends Serializable> implement
 		return mongoOperations.count(new Query(), entityInformation.getCollectionName());
 	}
 
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#deleteById(java.lang.Object)
@@ -466,6 +472,20 @@ public class SimpleReactiveMongoRepository<T, ID extends Serializable> implement
 		return mongoOperations.exists(query, example.getProbeType(), entityInformation.getCollectionName());
 	}
 
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.ReactiveQueryByExampleExecutor#findBy(org.springframework.data.domain.Example, java.util.function.Function)
+	 */
+	@Override
+	public <S extends T, R, P extends Publisher<R>> P findBy(Example<S> example,
+			Function<FluentQuery.ReactiveFluentQuery<S>, P> queryFunction) {
+
+		Assert.notNull(example, "Sample must not be null!");
+		Assert.notNull(queryFunction, "Query function must not be null!");
+
+		return queryFunction.apply(new ReactiveFluentQueryByExample<>(example, example.getProbeType()));
+	}
+
 	private Query getIdQuery(Object id) {
 		return new Query(getIdCriteria(id));
 	}
@@ -486,4 +506,110 @@ public class SimpleReactiveMongoRepository<T, ID extends Serializable> implement
 	private Flux<T> findAll(Query query) {
 		return mongoOperations.find(query, entityInformation.getJavaType(), entityInformation.getCollectionName());
 	}
+
+	/**
+	 * {@link org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery} using {@link Example}.
+	 *
+	 * @author Mark Paluch
+	 * @since 3.3
+	 */
+	class ReactiveFluentQueryByExample<S, T> extends ReactiveFluentQuerySupport<Example<S>, T> {
+
+		ReactiveFluentQueryByExample(Example<S> example, Class<T> resultType) {
+			this(example, Sort.unsorted(), resultType, Collections.emptyList());
+		}
+
+		ReactiveFluentQueryByExample(Example<S> example, Sort sort, Class<T> resultType, List<String> fieldsToInclude) {
+			super(example, sort, resultType, fieldsToInclude);
+		}
+
+		@Override
+		protected <R> ReactiveFluentQueryByExample<S, R> create(Example<S> predicate, Sort sort, Class<R> resultType,
+				List<String> fieldsToInclude) {
+			return new ReactiveFluentQueryByExample<>(predicate, sort, resultType, fieldsToInclude);
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#one()
+		 */
+		@Override
+		public Mono<T> one() {
+			return createQuery().one();
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#first()
+		 */
+		@Override
+		public Mono<T> first() {
+			return createQuery().first();
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#all()
+		 */
+		@Override
+		public Flux<T> all() {
+			return createQuery().all();
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#page(org.springframework.data.domain.Pageable)
+		 */
+		@Override
+		public Mono<Page<T>> page(Pageable pageable) {
+
+			Assert.notNull(pageable, "Pageable must not be null!");
+
+			Mono<List<T>> items = createQuery(q -> q.with(pageable)).all().collectList();
+
+			return items.flatMap(content -> ReactivePageableExecutionUtils.getPage(content, pageable, this.count()));
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#count()
+		 */
+		@Override
+		public Mono<Long> count() {
+			return createQuery().count();
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.FluentQuery.ReactiveFluentQuery#exists()
+		 */
+		@Override
+		public Mono<Boolean> exists() {
+			return createQuery().exists();
+		}
+
+		private ReactiveFindOperation.TerminatingFind<T> createQuery() {
+			return createQuery(UnaryOperator.identity());
+		}
+
+		private ReactiveFindOperation.TerminatingFind<T> createQuery(UnaryOperator<Query> queryCustomizer) {
+
+			Query query = new Query(new Criteria().alike(getPredicate())) //
+					.collation(entityInformation.getCollation());
+
+			if (getSort().isSorted()) {
+				query.with(getSort());
+			}
+
+			if (!getFieldsToInclude().isEmpty()) {
+				query.fields().include(getFieldsToInclude().toArray(new String[0]));
+			}
+
+			query = queryCustomizer.apply(query);
+
+			return mongoOperations.query(getPredicate().getProbeType()).inCollection(entityInformation.getCollectionName())
+					.as(getResultType()).matching(query);
+		}
+
+	}
 }

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

@@ -16,14 +16,21 @@
 package org.springframework.data.mongodb.repository.support;
 
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
 
 import org.bson.Document;
 
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageImpl;
+import org.springframework.data.domain.Pageable;
 import org.springframework.data.mongodb.core.ExecutableFindOperation;
 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.support.PageableExecutionUtils;
 import org.springframework.lang.Nullable;
 
 import com.mysema.commons.lang.CloseableIterator;
@@ -35,7 +42,6 @@ import com.querydsl.core.QueryResults;
 import com.querydsl.core.types.Expression;
 import com.querydsl.core.types.OrderSpecifier;
 import com.querydsl.core.types.Predicate;
-import com.querydsl.mongodb.document.MongodbDocumentSerializer;
 
 /**
  * Spring Data specific simple {@link com.querydsl.core.Fetchable} {@link com.querydsl.core.SimpleQuery Query}
@@ -48,10 +54,9 @@ import com.querydsl.mongodb.document.MongodbDocumentSerializer;
 public class SpringDataMongodbQuery<T> extends SpringDataMongodbQuerySupport<SpringDataMongodbQuery<T>>
 		implements Fetchable<T> {
 
-	private final Class<T> entityClass;
-	private final String collection;
 	private final MongoOperations mongoOperations;
-	private final ExecutableFindOperation.FindWithProjection<T> find;
+	private final Consumer<BasicQuery> queryCustomizer;
+	private final ExecutableFindOperation.FindWithQuery<T> find;
 
 	/**
 	 * Creates a new {@link SpringDataMongodbQuery}.
@@ -72,18 +77,26 @@ public class SpringDataMongodbQuery<T> extends SpringDataMongodbQuerySupport<Spr
 	 */
 	public SpringDataMongodbQuery(MongoOperations operations, Class<? extends T> type,
 			String collectionName) {
-		this(new SpringDataMongodbSerializer(operations.getConverter()), operations, type, collectionName);
+		this(operations, type, type, collectionName, it -> {});
 	}
 
-	private SpringDataMongodbQuery(MongodbDocumentSerializer serializer, MongoOperations operations,
-			Class<? extends T> type, String collectionName) {
+	/**
+	 * Creates a new {@link SpringDataMongodbQuery}.
+	 *
+	 * @param operations must not be {@literal null}.
+	 * @param domainType must not be {@literal null}.
+	 * @param resultType must not be {@literal null}.
+	 * @param collectionName must not be {@literal null} or empty.
+	 * @since 3.3
+	 */
+	SpringDataMongodbQuery(MongoOperations operations, Class<?> domainType, Class<? extends T> resultType,
+			String collectionName, Consumer<BasicQuery> queryCustomizer) {
+		super(new SpringDataMongodbSerializer(operations.getConverter()));
 
-		super(serializer);
-
-		this.entityClass = (Class<T>) type;
-		this.collection = collectionName;
+		Class<T> resultType1 = (Class<T>) resultType;
 		this.mongoOperations = operations;
-		this.find = mongoOperations.query(this.entityClass).inCollection(collection);
+		this.queryCustomizer = queryCustomizer;
+		this.find = mongoOperations.query(domainType).inCollection(collectionName).as(resultType1);
 	}
 
 	/*
@@ -94,19 +107,19 @@ public class SpringDataMongodbQuery<T> extends SpringDataMongodbQuerySupport<Spr
 	public CloseableIterator<T> iterate() {
 
 		try {
-			org.springframework.data.util.CloseableIterator<? extends T> stream = mongoOperations.stream(createQuery(),
-					entityClass, collection);
+			Stream<T> stream = stream();
+			Iterator<T> iterator = stream.iterator();
 
 			return new CloseableIterator<T>() {
 
 				@Override
 				public boolean hasNext() {
-					return stream.hasNext();
+					return iterator.hasNext();
 				}
 
 				@Override
 				public T next() {
-					return stream.next();
+					return iterator.next();
 				}
 
 				@Override
@@ -124,6 +137,20 @@ public class SpringDataMongodbQuery<T> extends SpringDataMongodbQuerySupport<Spr
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see com.querydsl.core.Fetchable#iterable()
+	 */
+	@Override
+	public Stream<T> stream() {
+
+		try {
+			return find.matching(createQuery()).stream();
+		} catch (RuntimeException e) {
+			return handleException(e, Stream.empty());
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see com.querydsl.core.Fetchable#fetch()
@@ -137,6 +164,24 @@ public class SpringDataMongodbQuery<T> extends SpringDataMongodbQuerySupport<Spr
 		}
 	}
 
+	/**
+	 * Fetch a {@link Page}.
+	 *
+	 * @param pageable
+	 * @return
+	 */
+	public Page<T> fetchPage(Pageable pageable) {
+
+		try {
+
+			List<T> content = find.matching(createQuery().with(pageable)).all();
+
+			return PageableExecutionUtils.getPage(content, pageable, this::fetchCount);
+		} catch (RuntimeException e) {
+			return handleException(e, new PageImpl<>(Collections.emptyList(), pageable, 0));
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see com.querydsl.core.Fetchable#fetchFirst()
@@ -215,6 +260,8 @@ public class SpringDataMongodbQuery<T> extends SpringDataMongodbQuerySupport<Spr
 			basicQuery.setSortObject(createSort(orderBy));
 		}
 
+		queryCustomizer.accept(basicQuery);
+
 		return basicQuery;
 	}
 

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/SimpleReactiveMongoRepositoryTests.java

@@ -40,12 +40,14 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.domain.Example;
+import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.domain.Sort.Order;
 import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
 import org.springframework.data.mongodb.repository.support.ReactiveMongoRepositoryFactory;
 import org.springframework.data.mongodb.repository.support.SimpleReactiveMongoRepository;
+import org.springframework.data.repository.query.FluentQuery;
 import org.springframework.data.repository.query.ReactiveQueryMethodEvaluationContextProvider;
 import org.springframework.lang.Nullable;
 import org.springframework.test.context.ContextConfiguration;
@@ -475,6 +477,150 @@ public class SimpleReactiveMongoRepositoryTests implements BeanClassLoaderAware,
 		repository.findOne(example).as(StepVerifier::create).verifyComplete();
 	}
 
+	@Test // GH-3757
+	void findByShouldReturnFirstResult() {
+
+		ReactivePerson probe = new ReactivePerson();
+		probe.setFirstname(oliver.getFirstname());
+
+		repository.findBy(Example.of(probe, matching().withIgnorePaths("age")), FluentQuery.ReactiveFluentQuery::first) //
+				.as(StepVerifier::create) //
+				.expectNext(oliver) //
+				.verifyComplete();
+	}
+
+	@Test // GH-3757
+	void findByShouldReturnOneResult() {
+
+		ReactivePerson probe = new ReactivePerson();
+		probe.setFirstname(oliver.getFirstname());
+
+		repository.findBy(Example.of(probe, matching().withIgnorePaths("age")), FluentQuery.ReactiveFluentQuery::one) //
+				.as(StepVerifier::create) //
+				.expectNext(oliver) //
+				.verifyComplete();
+
+		probe = new ReactivePerson();
+		probe.setLastname(oliver.getLastname());
+
+		repository.findBy(Example.of(probe, matching().withIgnorePaths("age")), FluentQuery.ReactiveFluentQuery::one) //
+				.as(StepVerifier::create) //
+				.verifyError(IncorrectResultSizeDataAccessException.class);
+	}
+
+	@Test // GH-3757
+	void findByShouldReturnAll() {
+
+		ReactivePerson probe = new ReactivePerson();
+		probe.setLastname(oliver.getLastname());
+
+		repository.findBy(Example.of(probe, matching().withIgnorePaths("age")), FluentQuery.ReactiveFluentQuery::all) //
+				.as(StepVerifier::create) //
+				.expectNextCount(2) //
+				.verifyComplete();
+	}
+
+	@Test // GH-3757
+	void findByShouldApplySortAll() {
+
+		ReactivePerson probe = new ReactivePerson();
+		probe.setLastname(oliver.getLastname());
+
+		repository.findBy(Example.of(probe, matching().withIgnorePaths("age")), it -> it.sortBy(Sort.by("firstname")).all()) //
+				.as(StepVerifier::create) //
+				.expectNext(dave, oliver) //
+				.verifyComplete();
+
+		repository
+				.findBy(Example.of(probe, matching().withIgnorePaths("age")),
+						it -> it.sortBy(Sort.by(Direction.DESC, "firstname")).all()) //
+				.as(StepVerifier::create) //
+				.expectNext(oliver, dave) //
+				.verifyComplete();
+	}
+
+	@Test // GH-3757
+	void findByShouldApplyProjection() {
+
+		ReactivePerson probe = new ReactivePerson();
+		probe.setLastname(oliver.getLastname());
+
+		repository.findBy(Example.of(probe, matching().withIgnorePaths("age")), it -> it.project("firstname").first()) //
+				.as(StepVerifier::create) //
+				.assertNext(it -> {
+
+					assertThat(it.getFirstname()).isNotNull();
+					assertThat(it.getLastname()).isNull();
+				}).verifyComplete();
+	}
+
+	@Test // GH-3757
+	void findByShouldApplyPagination() {
+
+		ReactivePerson probe = new ReactivePerson();
+		probe.setLastname(oliver.getLastname());
+
+		repository
+				.findBy(Example.of(probe, matching().withIgnorePaths("age")),
+						it -> it.page(PageRequest.of(0, 1, Sort.by("firstname")))) //
+				.as(StepVerifier::create) //
+				.assertNext(it -> {
+
+					assertThat(it.getTotalElements()).isEqualTo(2);
+					assertThat(it.getContent()).contains(dave);
+				}).verifyComplete();
+
+		repository
+				.findBy(Example.of(probe, matching().withIgnorePaths("age")),
+						it -> it.page(PageRequest.of(1, 1, Sort.by("firstname")))) //
+				.as(StepVerifier::create) //
+				.assertNext(it -> {
+
+					assertThat(it.getTotalElements()).isEqualTo(2);
+					assertThat(it.getContent()).contains(oliver);
+				}).verifyComplete();
+	}
+
+	@Test // GH-3757
+	void findByShouldCount() {
+
+		ReactivePerson probe = new ReactivePerson();
+		probe.setLastname(oliver.getLastname());
+
+		repository.findBy(Example.of(probe, matching().withIgnorePaths("age")), FluentQuery.ReactiveFluentQuery::count) //
+				.as(StepVerifier::create) //
+				.expectNext(2L) //
+				.verifyComplete();
+
+		probe = new ReactivePerson();
+		probe.setLastname("foo");
+
+		repository.findBy(Example.of(probe, matching().withIgnorePaths("age")), FluentQuery.ReactiveFluentQuery::count) //
+				.as(StepVerifier::create) //
+				.expectNext(0L) //
+				.verifyComplete();
+	}
+
+	@Test // GH-3757
+	void findByShouldReportExists() {
+
+		ReactivePerson probe = new ReactivePerson();
+		probe.setLastname(oliver.getLastname());
+
+		repository.findBy(Example.of(probe, matching().withIgnorePaths("age")), FluentQuery.ReactiveFluentQuery::exists) //
+				.as(StepVerifier::create) //
+				.expectNext(true) //
+				.verifyComplete();
+
+		probe = new ReactivePerson();
+		probe.setLastname("foo");
+
+		repository.findBy(Example.of(probe, matching().withIgnorePaths("age")), FluentQuery.ReactiveFluentQuery::exists) //
+				.as(StepVerifier::create) //
+				.expectNext(false) //
+				.verifyComplete();
+	}
+
 	interface ReactivePersonRepository extends ReactiveMongoRepository<ReactivePerson, String> {
 
 		Flux<ReactivePerson> findByLastname(String lastname);

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/support/QuerydslMongoPredicateExecutorIntegrationTests.java

@@ -24,9 +24,11 @@ import java.util.List;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.dao.PermissionDeniedDataAccessException;
+import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
@@ -41,6 +43,7 @@ import org.springframework.data.mongodb.repository.QPerson;
 import org.springframework.data.mongodb.repository.QUser;
 import org.springframework.data.mongodb.repository.User;
 import org.springframework.data.mongodb.repository.query.MongoEntityInformation;
+import org.springframework.data.repository.query.FluentQuery;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringRunner;
 
@@ -240,4 +243,104 @@ public class QuerydslMongoPredicateExecutorIntegrationTests {
 
 		repository.findOne(person.firstname.contains("batman"));
 	}
+
+	@Test // GH-3757
+	public void findByShouldReturnFirstResult() {
+
+		Person result = repository.findBy(person.firstname.eq(oliver.getFirstname()),
+				FluentQuery.FetchableFluentQuery::oneValue);
+
+		assertThat(result).isEqualTo(oliver);
+	}
+
+	@Test // GH-3757
+	public void findByShouldReturnOneResult() {
+
+		Person result = repository.findBy(person.firstname.eq(oliver.getFirstname()),
+				FluentQuery.FetchableFluentQuery::oneValue);
+
+		assertThat(result).isEqualTo(oliver);
+
+		assertThatExceptionOfType(IncorrectResultSizeDataAccessException.class).isThrownBy(
+				() -> repository.findBy(person.lastname.eq(oliver.getLastname()), FluentQuery.FetchableFluentQuery::one));
+	}
+
+	@Test // GH-3757
+	public void findByShouldReturnAll() {
+
+		List<Person> result = repository.findBy(person.lastname.eq(oliver.getLastname()),
+				FluentQuery.FetchableFluentQuery::all);
+
+		assertThat(result).hasSize(2);
+	}
+
+	@Test // GH-3757
+	public void findByShouldApplySortAll() {
+
+		Person probe = new Person();
+		probe.setLastname(oliver.getLastname());
+
+		List<Person> result = repository.findBy(person.lastname.eq(oliver.getLastname()),
+				it -> it.sortBy(Sort.by("firstname")).all());
+		assertThat(result).containsSequence(dave, oliver);
+
+		result = repository.findBy(person.lastname.eq(oliver.getLastname()),
+				it -> it.sortBy(Sort.by(Sort.Direction.DESC, "firstname")).all());
+		assertThat(result).containsSequence(oliver, dave);
+	}
+
+	@Test // GH-3757
+	public void findByShouldApplyProjection() {
+
+		Person probe = new Person();
+		probe.setLastname(oliver.getLastname());
+
+		Person result = repository.findBy(person.lastname.eq(oliver.getLastname()),
+				it -> it.project("firstname").firstValue());
+
+		assertThat(result.getFirstname()).isNotNull();
+		assertThat(result.getLastname()).isNull();
+	}
+
+	@Test // GH-3757
+	public void findByShouldApplyPagination() {
+
+		Page<Person> first = repository.findBy(person.lastname.eq(oliver.getLastname()),
+				it -> it.page(PageRequest.of(0, 1, Sort.by("firstname"))));
+		assertThat(first.getTotalElements()).isEqualTo(2);
+		assertThat(first.getContent()).contains(dave);
+
+		Page<Person> next = repository.findBy(person.lastname.eq(oliver.getLastname()),
+				it -> it.page(PageRequest.of(1, 1, Sort.by("firstname"))));
+
+		assertThat(next.getTotalElements()).isEqualTo(2);
+		assertThat(next.getContent()).contains(oliver);
+	}
+
+	@Test // GH-3757
+	public void findByShouldCount() {
+
+		long count = repository.findBy(person.lastname.eq(oliver.getLastname()), FluentQuery.FetchableFluentQuery::count);
+		assertThat(count).isEqualTo(2L);
+
+		count = repository.findBy(person.lastname.eq("foo"), FluentQuery.FetchableFluentQuery::count);
+		assertThat(count).isEqualTo(0L);
+	}
+
+	@Test // GH-3757
+	public void findByShouldReportExists() {
+
+		Person probe = new Person();
+		probe.setLastname(oliver.getLastname());
+
+		boolean exists = repository.findBy(person.lastname.eq(oliver.getLastname()),
+				FluentQuery.FetchableFluentQuery::exists);
+		assertThat(exists).isTrue();
+
+		probe = new Person();
+		probe.setLastname("foo");
+
+		exists = repository.findBy(person.lastname.eq("foo"), FluentQuery.FetchableFluentQuery::exists);
+		assertThat(exists).isFalse();
+	}
 }

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/support/ReactiveQuerydslMongoPredicateExecutorTests.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.mongodb.repository.support;
 
+import static org.assertj.core.api.Assertions.*;
+
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
@@ -34,6 +36,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.dao.PermissionDeniedDataAccessException;
+import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
@@ -49,6 +52,7 @@ import org.springframework.data.mongodb.repository.QUser;
 import org.springframework.data.mongodb.repository.User;
 import org.springframework.data.mongodb.repository.query.MongoEntityInformation;
 import org.springframework.data.mongodb.test.util.MongoTestUtils;
+import org.springframework.data.repository.query.FluentQuery;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringRunner;
 
@@ -269,12 +273,11 @@ public class ReactiveQuerydslMongoPredicateExecutorTests {
 				.as(StepVerifier::create) //
 				.expectNextCount(1) //
 				.verifyComplete();
-		;
+
 		operations.save(person2) //
 				.as(StepVerifier::create) //
 				.expectNextCount(1) //
 				.verifyComplete();
-		;
 
 		Flux<Person> result = new ReactiveSpringDataMongodbQuery<>(operations, Person.class).where()
 				.join(person.coworker, QUser.user).on(QUser.user.username.eq("does-not-exist")).fetch();
@@ -330,4 +333,112 @@ public class ReactiveQuerydslMongoPredicateExecutorTests {
 				.expectError(PermissionDeniedDataAccessException.class) //
 				.verify();
 	}
+
+	@Test // GH-3757
+	public void findByShouldReturnFirstResult() {
+
+		repository.findBy(person.firstname.eq(oliver.getFirstname()), FluentQuery.ReactiveFluentQuery::first) //
+				.as(StepVerifier::create) //
+				.expectNext(oliver) //
+				.verifyComplete();
+	}
+
+	@Test // GH-3757
+	public void findByShouldReturnOneResult() {
+
+		repository.findBy(person.firstname.eq(oliver.getFirstname()), FluentQuery.ReactiveFluentQuery::one) //
+				.as(StepVerifier::create) //
+				.expectNext(oliver) //
+				.verifyComplete();
+
+		repository.findBy(person.lastname.eq(oliver.getLastname()), FluentQuery.ReactiveFluentQuery::one) //
+				.as(StepVerifier::create) //
+				.verifyError(IncorrectResultSizeDataAccessException.class);
+	}
+
+	@Test // GH-3757
+	public void findByShouldReturnAll() {
+
+		repository.findBy(person.lastname.eq(oliver.getLastname()), FluentQuery.ReactiveFluentQuery::all) //
+				.as(StepVerifier::create) //
+				.expectNextCount(2) //
+				.verifyComplete();
+	}
+
+	@Test // GH-3757
+	public void findByShouldApplySortAll() {
+
+		repository.findBy(person.lastname.eq(oliver.getLastname()), it -> it.sortBy(Sort.by("firstname")).all()) //
+				.as(StepVerifier::create) //
+				.expectNext(dave, oliver) //
+				.verifyComplete();
+
+		repository
+				.findBy(person.lastname.eq(oliver.getLastname()), it -> it.sortBy(Sort.by(Direction.DESC, "firstname")).all()) //
+				.as(StepVerifier::create) //
+				.expectNext(oliver, dave) //
+				.verifyComplete();
+	}
+
+	@Test // GH-3757
+	public void findByShouldApplyProjection() {
+
+		repository.findBy(person.lastname.eq(oliver.getLastname()), it -> it.project("firstname").first()) //
+				.as(StepVerifier::create) //
+				.assertNext(it -> {
+
+					assertThat(it.getFirstname()).isNotNull();
+					assertThat(it.getLastname()).isNull();
+				}).verifyComplete();
+	}
+
+	@Test // GH-3757
+	public void findByShouldApplyPagination() {
+
+		repository
+				.findBy(person.lastname.eq(oliver.getLastname()), it -> it.page(PageRequest.of(0, 1, Sort.by("firstname")))) //
+				.as(StepVerifier::create) //
+				.assertNext(it -> {
+
+					assertThat(it.getTotalElements()).isEqualTo(2);
+					assertThat(it.getContent()).contains(dave);
+				}).verifyComplete();
+
+		repository
+				.findBy(person.lastname.eq(oliver.getLastname()), it -> it.page(PageRequest.of(1, 1, Sort.by("firstname")))) //
+				.as(StepVerifier::create) //
+				.assertNext(it -> {
+
+					assertThat(it.getTotalElements()).isEqualTo(2);
+					assertThat(it.getContent()).contains(oliver);
+				}).verifyComplete();
+	}
+
+	@Test // GH-3757
+	public void findByShouldCount() {
+
+		repository.findBy(person.lastname.eq(oliver.getLastname()), FluentQuery.ReactiveFluentQuery::count) //
+				.as(StepVerifier::create) //
+				.expectNext(2L) //
+				.verifyComplete();
+
+		repository.findBy(person.lastname.eq("foo"), FluentQuery.ReactiveFluentQuery::count) //
+				.as(StepVerifier::create) //
+				.expectNext(0L) //
+				.verifyComplete();
+	}
+
+	@Test // GH-3757
+	public void findByShouldReportExists() {
+
+		repository.findBy(person.lastname.eq(oliver.getLastname()), FluentQuery.ReactiveFluentQuery::exists) //
+				.as(StepVerifier::create) //
+				.expectNext(true) //
+				.verifyComplete();
+
+		repository.findBy(person.lastname.eq("foo"), FluentQuery.ReactiveFluentQuery::exists) //
+				.as(StepVerifier::create) //
+				.expectNext(false) //
+				.verifyComplete();
+	}
 }

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/support/SimpleMongoRepositoryTests.java

@@ -31,10 +31,13 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
+import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.domain.Example;
+import org.springframework.data.domain.ExampleMatcher;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.geo.Point;
 import org.springframework.data.mongodb.MongoTransactionManager;
 import org.springframework.data.mongodb.core.geo.GeoJsonPoint;
@@ -51,6 +54,7 @@ import org.springframework.data.mongodb.test.util.MongoServerCondition;
 import org.springframework.data.mongodb.test.util.MongoTemplateExtension;
 import org.springframework.data.mongodb.test.util.MongoTestTemplate;
 import org.springframework.data.mongodb.test.util.Template;
+import org.springframework.data.repository.query.FluentQuery;
 import org.springframework.test.util.ReflectionTestUtils;
 import org.springframework.transaction.support.TransactionTemplate;
 
@@ -459,6 +463,126 @@ class SimpleMongoRepositoryTests {
 				.hasSize(all.size() - 2).doesNotContain(dave, carter);
 	}
 
+	@Test // GH-3757
+	void findByShouldReturnFirstResult() {
+
+		Person probe = new Person();
+		probe.setFirstname(oliver.getFirstname());
+
+		Person result = repository.findBy(Example.of(probe, getMatcher()), FluentQuery.FetchableFluentQuery::firstValue);
+
+		assertThat(result).isEqualTo(oliver);
+	}
+
+	@Test // GH-3757
+	void findByShouldReturnOneResult() {
+
+		Person probe = new Person();
+		probe.setFirstname(oliver.getFirstname());
+
+		Person result = repository.findBy(Example.of(probe, getMatcher()), FluentQuery.FetchableFluentQuery::oneValue);
+
+		assertThat(result).isEqualTo(oliver);
+
+		Person probeByLastname = new Person();
+		probeByLastname.setLastname(oliver.getLastname());
+
+		assertThatExceptionOfType(IncorrectResultSizeDataAccessException.class).isThrownBy(
+				() -> repository.findBy(Example.of(probeByLastname, getMatcher()), FluentQuery.FetchableFluentQuery::one));
+	}
+
+	@Test // GH-3757
+	void findByShouldReturnAll() {
+
+		Person probe = new Person();
+		probe.setLastname(oliver.getLastname());
+
+		List<Person> result = repository.findBy(Example.of(probe, getMatcher()), FluentQuery.FetchableFluentQuery::all);
+
+		assertThat(result).hasSize(2);
+	}
+
+	@Test // GH-3757
+	void findByShouldApplySortAll() {
+
+		Person probe = new Person();
+		probe.setLastname(oliver.getLastname());
+
+		List<Person> result = repository.findBy(Example.of(probe, getMatcher()),
+				it -> it.sortBy(Sort.by("firstname")).all());
+		assertThat(result).containsSequence(dave, oliver);
+
+		result = repository.findBy(Example.of(probe, getMatcher()),
+				it -> it.sortBy(Sort.by(Sort.Direction.DESC, "firstname")).all());
+		assertThat(result).containsSequence(oliver, dave);
+	}
+
+	@Test // GH-3757
+	void findByShouldApplyProjection() {
+
+		Person probe = new Person();
+		probe.setLastname(oliver.getLastname());
+
+		Person result = repository.findBy(Example.of(probe, getMatcher()), it -> it.project("firstname").firstValue());
+
+		assertThat(result.getFirstname()).isNotNull();
+		assertThat(result.getLastname()).isNull();
+	}
+
+	@Test // GH-3757
+	void findByShouldApplyPagination() {
+
+		Person probe = new Person();
+		probe.setLastname(oliver.getLastname());
+
+		Page<Person> first = repository.findBy(Example.of(probe, getMatcher()),
+				it -> it.page(PageRequest.of(0, 1, Sort.by("firstname"))));
+		assertThat(first.getTotalElements()).isEqualTo(2);
+		assertThat(first.getContent()).contains(dave);
+
+		Page<Person> next = repository.findBy(Example.of(probe, getMatcher()),
+				it -> it.page(PageRequest.of(1, 1, Sort.by("firstname"))));
+
+		assertThat(next.getTotalElements()).isEqualTo(2);
+		assertThat(next.getContent()).contains(oliver);
+	}
+
+	@Test // GH-3757
+	void findByShouldCount() {
+
+		Person probe = new Person();
+		probe.setLastname(oliver.getLastname());
+
+		long count = repository.findBy(Example.of(probe, getMatcher()), FluentQuery.FetchableFluentQuery::count);
+		assertThat(count).isEqualTo(2L);
+
+		probe = new Person();
+		probe.setLastname("foo");
+
+		count = repository.findBy(Example.of(probe, getMatcher()), FluentQuery.FetchableFluentQuery::count);
+		assertThat(count).isEqualTo(0L);
+	}
+
+	@Test // GH-3757
+	void findByShouldReportExists() {
+
+		Person probe = new Person();
+		probe.setLastname(oliver.getLastname());
+
+		boolean exists = repository.findBy(Example.of(probe, getMatcher()), FluentQuery.FetchableFluentQuery::exists);
+		assertThat(exists).isTrue();
+
+		probe = new Person();
+		probe.setLastname("foo");
+
+		exists = repository.findBy(Example.of(probe, getMatcher()), FluentQuery.FetchableFluentQuery::exists);
+		assertThat(exists).isFalse();
+	}
+
+	private ExampleMatcher getMatcher() {
+		return matching().withIgnorePaths("age", "createdAt", "sex", "email", "id");
+	}
+
 	private void assertThatAllReferencePersonsWereStoredCorrectly(Map<String, Person> references, List<Person> saved) {
 
 		for (Person person : saved) {