diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/CodecRegistryProvider.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/CodecRegistryProvider.java
new file mode 100644
index 000000000..f470dd8a1
--- /dev/null
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/CodecRegistryProvider.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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;
+
+import java.util.Optional;
+
+import org.bson.codecs.Codec;
+import org.bson.codecs.configuration.CodecConfigurationException;
+import org.bson.codecs.configuration.CodecRegistry;
+import org.springframework.util.Assert;
+
+/**
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+public interface CodecRegistryProvider {
+
+	/**
+	 * Get the underlying {@link CodecRegistry} used by the MongoDB Java driver.
+	 *
+	 * @return never {@literal null}.
+	 * @throws IllegalStateException if {@link CodecRegistry} cannot be obtained.
+	 */
+	CodecRegistry getCodecRegistry();
+
+	/**
+	 * Checks if a {@link Codec} is registered for a given type.
+	 *
+	 * @param type must not be {@literal null}.
+	 * @return true if {@link #getCodecRegistry()} holds a {@link Codec} for given type.
+	 * @throws IllegalStateException if {@link CodecRegistry} cannot be obtained.
+	 */
+	default boolean hasCodecFor(Class<?> type) {
+		return getCodecFor(type).isPresent();
+	}
+
+	/**
+	 * Get the {@link Codec} registered for the given {@literal type} or an {@link Optional#empty() empty Optional}
+	 * instead.
+	 * 
+	 * @param type must not be {@literal null}.
+	 * @param <T>
+	 * @return never {@literal null}.
+	 * @throws IllegalArgumentException if {@literal type} is {@literal null}.
+	 */
+	default <T> Optional<Codec<T>> getCodecFor(Class<T> type) {
+
+		Assert.notNull(type, "Type must not be null!");
+
+		try {
+			return Optional.of(getCodecRegistry().get(type));
+		} catch (CodecConfigurationException e) {
+			// ignore
+		}
+		return Optional.empty();
+	}
+}
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/MongoDbFactory.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/MongoDbFactory.java
index 55d06aec3..3604fe6ac 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/MongoDbFactory.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/MongoDbFactory.java
@@ -15,6 +15,7 @@
  */
 package org.springframework.data.mongodb;
 
+import org.bson.codecs.configuration.CodecRegistry;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.mongodb.core.MongoExceptionTranslator;
@@ -28,7 +29,7 @@ import com.mongodb.client.MongoDatabase;
  * @author Mark Pollack
  * @author Thomas Darimont
  */
-public interface MongoDbFactory {
+public interface MongoDbFactory extends CodecRegistryProvider {
 
 	/**
 	 * Creates a default {@link DB} instance.
@@ -55,4 +56,9 @@ public interface MongoDbFactory {
 	PersistenceExceptionTranslator getExceptionTranslator();
 
 	DB getLegacyDb();
+
+	@Override
+	default CodecRegistry getCodecRegistry() {
+		return getDb().getCodecRegistry();
+	}
 }
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseFactory.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseFactory.java
index e67f46084..46a945056 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseFactory.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/ReactiveMongoDatabaseFactory.java
@@ -16,6 +16,7 @@
 
 package org.springframework.data.mongodb;
 
+import org.bson.codecs.configuration.CodecRegistry;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.mongodb.core.MongoExceptionTranslator;
@@ -26,9 +27,10 @@ import com.mongodb.reactivestreams.client.MongoDatabase;
  * Interface for factories creating reactive {@link MongoDatabase} instances.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.0
  */
-public interface ReactiveMongoDatabaseFactory {
+public interface ReactiveMongoDatabaseFactory extends CodecRegistryProvider {
 
 	/**
 	 * Creates a default {@link MongoDatabase} instance.
@@ -53,4 +55,9 @@ public interface ReactiveMongoDatabaseFactory {
 	 * @return will never be {@literal null}.
 	 */
 	PersistenceExceptionTranslator getExceptionTranslator();
+
+	@Override
+	default CodecRegistry getCodecRegistry() {
+		return getMongoDatabase().getCodecRegistry();
+	}
 }
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperation.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperation.java
index a383f4470..92bdb5e6e 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperation.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperation.java
@@ -19,11 +19,14 @@ import java.util.List;
 import java.util.Optional;
 import java.util.stream.Stream;
 
+import org.springframework.dao.DataAccessException;
 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.lang.Nullable;
 
+import com.mongodb.client.MongoCollection;
+
 /**
  * {@link ExecutableFindOperation} allows creation and execution of MongoDB find operations in a fluent API style.
  * <br />
@@ -202,7 +205,7 @@ public interface ExecutableFindOperation {
 	 * @author Christoph Strobl
 	 * @since 2.0
 	 */
-	interface FindWithProjection<T> extends FindWithQuery<T> {
+	interface FindWithProjection<T> extends FindWithQuery<T>, FindDistinct {
 
 		/**
 		 * Define the target type fields should be mapped to. <br />
@@ -214,6 +217,101 @@ public interface ExecutableFindOperation {
 		 * @throws IllegalArgumentException if resultType is {@literal null}.
 		 */
 		<R> FindWithQuery<R> as(Class<R> resultType);
+
+	}
+
+	/**
+	 * Distinct Find support.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface FindDistinct {
+
+		/**
+		 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view.
+		 *
+		 * @param field name of the field. Must not be {@literal null}.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if field is {@literal null}.
+		 */
+		TerminatingDistinct<Object> distinct(String field);
+	}
+
+	/**
+	 * Result type override. Optional.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface DistinctWithProjection {
+
+		/**
+		 * Define the target type the result should be mapped to. <br />
+		 * Skip this step if you are anyway fine with the default conversion.
+		 * <dl>
+		 * <dt>{@link Object} (the default)</dt>
+		 * <dd>Result is mapped according to the {@link org.bson.BsonType} converting eg. {@link org.bson.BsonString} into
+		 * plain {@link String}, {@link org.bson.BsonInt64} to {@link Long}, etc. always picking the most concrete type with
+		 * respect to the domain types property.<br />
+		 * Any {@link org.bson.BsonType#DOCUMENT} is run through the {@link org.springframework.data.convert.EntityReader}
+		 * to obtain the domain type. <br />
+		 * Using {@link Object} also works for non strictly typed fields. Eg. a mixture different types like fields using
+		 * {@link String} in one {@link org.bson.Document} while {@link Long} in another.</dd>
+		 * <dt>Any Simple type like {@link String}, {@link Long}, ...</dt>
+		 * <dd>The result is mapped directly by the MongoDB Java driver and the {@link org.bson.codecs.CodeCodec Codecs} in
+		 * place. This works only for results where all documents considered for the operation use the very same type for
+		 * the field.</dd>
+		 * <dt>Any Domain type</dt>
+		 * <dd>Domain types can only be mapped if the if the result of the actual {@code distinct()} operation returns
+		 * {@link org.bson.BsonType#DOCUMENT}.</dd>
+		 * <dt>{@link org.bson.BsonValue}</dt>
+		 * <dd>Using {@link org.bson.BsonValue} allows retrieval of the raw driver specific format, which returns eg.
+		 * {@link org.bson.BsonString}.</dd>
+		 * </dl>
+		 *
+		 * @param resultType must not be {@literal null}.
+		 * @param <R> result type.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if resultType is {@literal null}.
+		 */
+		<R> TerminatingDistinct<R> as(Class<R> resultType);
+	}
+
+	/**
+	 * Result restrictions. Optional.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface DistinctWithQuery<T> extends DistinctWithProjection {
+
+		/**
+		 * Set the filter query to be used.
+		 *
+		 * @param query must not be {@literal null}.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if resultType is {@literal null}.
+		 */
+		TerminatingDistinct<T> matching(Query query);
+	}
+
+	/**
+	 * Terminating distinct find operations.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface TerminatingDistinct<T> extends DistinctWithQuery<T> {
+
+		/**
+		 * Get all matching distinct field values.
+		 *
+		 * @return empty {@link List} if not match found. Never {@literal null}.
+		 * @throws DataAccessException if eg. result cannot be converted correctly which may happen if the document contains
+		 *           {@link String} whereas the result type is specified as {@link Long}.
+		 */
+		List<T> all();
 	}
 
 	/**
@@ -222,5 +320,5 @@ public interface ExecutableFindOperation {
 	 * @author Christoph Strobl
 	 * @since 2.0
 	 */
-	interface ExecutableFind<T> extends FindWithCollection<T>, FindWithProjection<T> {}
+	interface ExecutableFind<T> extends FindWithCollection<T>, FindWithProjection<T>, FindDistinct {}
 }
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupport.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupport.java
index b0002344c..8f15265d3 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupport.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupport.java
@@ -205,6 +205,18 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 			return template.exists(query, domainType, getCollectionName());
 		}
 
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ExecutableFindOperation.FindDistinct#distinct(java.lang.String)
+		 */
+		@Override
+		public TerminatingDistinct<Object> distinct(String field) {
+
+			Assert.notNull(field, "Field must not be null!");
+
+			return new DistinctOperationSupport<>(this, field);
+		}
+
 		private List<T> doFind(@Nullable CursorPreparer preparer) {
 
 			Document queryObject = query.getQueryObject();
@@ -214,6 +226,12 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 					getCursorPreparer(query, preparer));
 		}
 
+		private List<T> doFindDistinct(String field) {
+
+			return template.findDistinct(query, field, getCollectionName(), domainType,
+					returnType == domainType ? (Class<T>) Object.class : returnType);
+		}
+
 		private CloseableIterator<T> doStream() {
 			return template.doStream(query, domainType, getCollectionName(), returnType);
 		}
@@ -261,4 +279,53 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation {
 			return this;
 		}
 	}
+
+	/**
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	static class DistinctOperationSupport<T> implements TerminatingDistinct<T> {
+
+		private final String field;
+		private final ExecutableFindSupport delegate;
+
+		public DistinctOperationSupport(ExecutableFindSupport delegate, String field) {
+
+			this.delegate = delegate;
+			this.field = field;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ExecutableFindOperation.DistinctWithProjection#as(java.lang.Class)
+		 */
+		@Override
+		public <R> TerminatingDistinct<R> as(Class<R> resultType) {
+
+			Assert.notNull(resultType, "ResultType must not be null!");
+
+			return new DistinctOperationSupport((ExecutableFindSupport) delegate.as(resultType), field);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ExecutableFindOperation.DistinctWithQuery#matching(org.springframework.data.mongodb.core.query.Query)
+		 */
+		@Override
+		public TerminatingDistinct<T> matching(Query query) {
+
+			Assert.notNull(query, "Query must not be null!");
+
+			return new DistinctOperationSupport((ExecutableFindSupport) delegate.matching(query), field);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ExecutableFindOperation.TerminatingDistinct#all()
+		 */
+		@Override
+		public List<T> all() {
+			return delegate.doFindDistinct(field);
+		}
+	}
 }
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java
index 0c87ffab8..d3f457dff 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java
@@ -709,6 +709,66 @@ public interface MongoOperations extends FluentMongoOperations {
 	@Nullable
 	<T> T findById(Object id, Class<T> entityClass, String collectionName);
 
+	/**
+	 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view and
+	 * returns the results in a {@link List}.
+	 *
+	 * @param field the name of the field to inspect for distinct values. Must not be {@literal null}.
+	 * @param entityClass the domain type used for determining the actual {@link MongoCollection}. Must not be
+	 *          {@literal null}.
+	 * @param resultClass the result type. Must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	default <T> List<T> findDistinct(String field, Class<?> entityClass, Class<T> resultClass) {
+		return findDistinct(new Query(), field, entityClass, resultClass);
+	}
+
+	/**
+	 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view and
+	 * returns the results in a {@link List}.
+	 *
+	 * @param query filter {@link Query} to restrict search. Must not be {@literal null}.
+	 * @param field the name of the field to inspect for distinct values. Must not be {@literal null}.
+	 * @param entityClass the domain type used for determining the actual {@link MongoCollection} and mapping the
+	 *          {@link Query} to the domain type fields. Must not be {@literal null}.
+	 * @param resultClass the result type. Must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	<T> List<T> findDistinct(Query query, String field, Class<?> entityClass, Class<T> resultClass);
+
+	/**
+	 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view and
+	 * returns the results in a {@link List}.
+	 *
+	 * @param query filter {@link Query} to restrict search. Must not be {@literal null}.
+	 * @param field the name of the field to inspect for distinct values. Must not be {@literal null}.
+	 * @param collectionName the explicit name of the actual {@link MongoCollection}. Must not be {@literal null}.
+	 * @param entityClass the domain type used for mapping the {@link Query} to the domain type fields.
+	 * @param resultClass the result type. Must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	<T> List<T> findDistinct(Query query, String field, String collectionName, Class<?> entityClass,
+			Class<T> resultClass);
+
+	/**
+	 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view and
+	 * returns the results in a {@link List}.
+	 *
+	 * @param query filter {@link Query} to restrict search. Must not be {@literal null}.
+	 * @param field the name of the field to inspect for distinct values. Must not be {@literal null}.
+	 * @param collection the explicit name of the actual {@link MongoCollection}. Must not be {@literal null}.
+	 * @param resultClass the result type. Must not be {@literal null}.
+	 * @param <T>
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	default <T> List<T> findDistinct(Query query, String field, String collection, Class<T> resultClass) {
+		return findDistinct(query, field, collection, Object.class, resultClass);
+	}
+
 	/**
 	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify<a/>
 	 * to apply provided {@link Update} on documents matching {@link Criteria} of given {@link Query}.
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java
index b0adb6795..91bbf1533 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java
@@ -29,7 +29,9 @@ import java.util.Map.Entry;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
+import org.bson.BsonValue;
 import org.bson.Document;
+import org.bson.codecs.Codec;
 import org.bson.conversions.Bson;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -55,6 +57,8 @@ import org.springframework.data.geo.GeoResults;
 import org.springframework.data.geo.Metric;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.mapping.PropertyPath;
+import org.springframework.data.mapping.PropertyReferenceException;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
 import org.springframework.data.mongodb.MongoDbFactory;
@@ -117,6 +121,7 @@ import org.springframework.util.ObjectUtils;
 import org.springframework.util.ResourceUtils;
 import org.springframework.util.StringUtils;
 
+import com.mongodb.Function;
 import com.mongodb.MongoClient;
 import com.mongodb.MongoException;
 import com.mongodb.ReadPreference;
@@ -797,28 +802,91 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 		return doFindOne(collectionName, new Document(idKey, id), new Document(), entityClass);
 	}
 
-	public <T, Z> List<T> distinct(String field, Class<Z> entityClass, Class<T> resultClass) {
-		return distinct(new Query(), field, determineCollectionName(entityClass), resultClass);
+	/*
+	 * (non-Javadoc) 
+	 * @see org.springframework.data.mongodb.core.MongoOperations#findDistinct(org.springframework.data.mongodb.core.query.Query, java.lang.String, java.lang.Class, java.lang.Class)
+	 */
+	@Override
+	public <T> List<T> findDistinct(Query query, String field, Class<?> entityClass, Class<T> resultClass) {
+		return findDistinct(query, field, determineCollectionName(entityClass), entityClass, resultClass);
 	}
 
-	public <T, Z> List<T> distinct(Query query, String field, Class<Z> entityClass, Class<T> resultClass) {
-		return distinct(query, field, determineCollectionName(entityClass), resultClass);
-	}
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.MongoOperations#findDistinct(org.springframework.data.mongodb.core.query.Query, java.lang.String, java.lang.String, java.lang.Class, java.lang.Class)
+	 */
+	@Override
+	public <T> List<T> findDistinct(Query query, String field, String collectionName, Class<?> entityClass,
+			Class<T> resultClass) {
 
-	public <T> List<T> distinct(Query query, String field, String collectionName, Class<T> resultClass) {
-		MongoCollection<Document> collection = this.getCollection(collectionName);
-		DistinctIterable<T> iterable = collection.distinct(field, query.getQueryObject(), resultClass);
+		Assert.notNull(query, "Query must not be null!");
+		Assert.notNull(field, "Field must not be null!");
+		Assert.notNull(collectionName, "CollectionName must not be null!");
+		Assert.notNull(entityClass, "EntityClass must not be null!");
+		Assert.notNull(resultClass, "ResultClass must not be null!");
 
-		MongoCursor<T> cursor = iterable.iterator();
+		MongoPersistentEntity<?> entity = entityClass != Object.class ? getPersistentEntity(entityClass) : null;
 
-		List<T> result = new ArrayList<T>();
+		Document mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), entity);
+		String mappedFieldName = queryMapper.getMappedFields(new Document(field, 1), entity).keySet().iterator().next();
 
-		while (cursor.hasNext()) {
-			T object = cursor.next();
-			result.add(object);
+		Class<?> mongoDriverCompatibleType = getMongoDbFactory().getCodecFor(resultClass).map(Codec::getEncoderClass)
+				.orElse((Class) BsonValue.class);
+
+		MongoIterable<?> result = execute((db) -> {
+
+			DistinctIterable<?> iterable = db.getCollection(collectionName).distinct(mappedFieldName, mappedQuery,
+					mongoDriverCompatibleType);
+
+			return query.getCollation().isPresent()
+					? iterable.collation(query.getCollation().map(Collation::toMongoCollation).get()) : iterable;
+		});
+
+		if (resultClass == Object.class || mongoDriverCompatibleType != resultClass) {
+			result = result.map(mapDistinctResult(getMostSpecificConversionTargetType(resultClass, entityClass, field)));
 		}
 
-		return result;
+		try {
+			return (List<T>) result.into(new ArrayList<>());
+		} catch (RuntimeException e) {
+			throw potentiallyConvertRuntimeException(e, exceptionTranslator);
+		}
+	}
+
+	/**
+	 * @param userType must not be {@literal null}.
+	 * @param domainType must not be {@literal null}.
+	 * @param field must not be {@literal null}.
+	 * @return the most specific conversion target type depending on user preference and domain type property.
+	 * @since 2.1
+	 */
+	private Class<?> getMostSpecificConversionTargetType(Class<?> userType, Class<?> domainType, String field) {
+
+		Class<?> conversionTargetType = userType;
+		try {
+
+			Class<?> propertyType = PropertyPath.from(field, domainType).getLeafProperty().getLeafType();
+
+			// use the more specific type but favor UserType over property one
+			if (ClassUtils.isAssignable(userType, propertyType)) {
+				conversionTargetType = propertyType;
+			}
+
+		} catch (PropertyReferenceException e) {
+			// just don't care about it as we default to Object.class anyway.
+		}
+
+		return conversionTargetType;
+	}
+
+	/**
+	 * @param targetType the desired conversion target type.
+	 * @return new {@link Function} converting {@link BsonValue} into desired target type.
+	 * @since 2.1
+	 */
+	private <S, T> Function<S, T> mapDistinctResult(Class<T> targetType) {
+		return (source) -> getConverter().mapValueToTargetType(targetType, new DefaultDbRefResolver(mongoDbFactory))
+				.apply(source);
 	}
 
 	@Override
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperation.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperation.java
index 1a42d7e52..4e28f850f 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperation.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperation.java
@@ -156,7 +156,7 @@ public interface ReactiveFindOperation {
 	/**
 	 * Result type override (optional).
 	 */
-	interface FindWithProjection<T> extends FindWithQuery<T> {
+	interface FindWithProjection<T> extends FindWithQuery<T>, FindDistinct {
 
 		/**
 		 * Define the target type fields should be mapped to. <br />
@@ -170,8 +170,101 @@ public interface ReactiveFindOperation {
 		<R> FindWithQuery<R> as(Class<R> resultType);
 	}
 
+	/**
+	 * Distinct Find support.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface FindDistinct {
+
+		/**
+		 * Finds the distinct values for a specified {@literal field} across a single
+		 * {@link com.mongodb.reactivestreams.client.MongoCollection} or view.
+		 *
+		 * @param field name of the field. Must not be {@literal null}.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if field is {@literal null}.
+		 */
+		TerminatingDistinct<Object> distinct(String field);
+	}
+
+	/**
+	 * Result type override. Optional.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface DistinctWithProjection {
+
+		/**
+		 * Define the target type the result should be mapped to. <br />
+		 * Skip this step if you are anyway fine with the default conversion.
+		 * <dl>
+		 * <dt>{@link Object} (the default)</dt>
+		 * <dd>Result is mapped according to the {@link org.bson.BsonType} converting eg. {@link org.bson.BsonString} into
+		 * plain {@link String}, {@link org.bson.BsonInt64} to {@link Long}, etc. always picking the most concrete type with
+		 * respect to the domain types property.<br />
+		 * Any {@link org.bson.BsonType#DOCUMENT} is run through the {@link org.springframework.data.convert.EntityReader}
+		 * to obtain the domain type. <br />
+		 * Using {@link Object} also works for non strictly typed fields. Eg. a mixture different types like fields using
+		 * {@link String} in one {@link org.bson.Document} while {@link Long} in another.</dd>
+		 * <dt>Any Simple type like {@link String}, {@link Long}, ...</dt>
+		 * <dd>The result is mapped directly by the MongoDB Java driver and the {@link org.bson.codecs.CodeCodec Codecs} in
+		 * place. This works only for results where all documents considered for the operation use the very same type for
+		 * the field.</dd>
+		 * <dt>Any Domain type</dt>
+		 * <dd>Domain types can only be mapped if the if the result of the actual {@code distinct()} operation returns
+		 * {@link org.bson.BsonType#DOCUMENT}.</dd>
+		 * <dt>{@link org.bson.BsonValue}</dt>
+		 * <dd>Using {@link org.bson.BsonValue} allows retrieval of the raw driver specific format, which returns eg.
+		 * {@link org.bson.BsonString}.</dd>
+		 * </dl>
+		 *
+		 * @param resultType must not be {@literal null}.
+		 * @param <R> result type.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if resultType is {@literal null}.
+		 */
+		<R> TerminatingDistinct<R> as(Class<R> resultType);
+	}
+
+	/**
+	 * Result restrictions. Optional.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface DistinctWithQuery<T> extends DistinctWithProjection {
+
+		/**
+		 * Set the filter query to be used.
+		 *
+		 * @param query must not be {@literal null}.
+		 * @return new instance of {@link TerminatingDistinct}.
+		 * @throws IllegalArgumentException if resultType is {@literal null}.
+		 */
+		TerminatingDistinct<T> matching(Query query);
+	}
+
+	/**
+	 * Terminating distinct find operations.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface TerminatingDistinct<T> extends DistinctWithQuery<T> {
+
+		/**
+		 * Get all matching distinct field values.
+		 *
+		 * @return empty {@link Flux} if not match found. Never {@literal null}.
+		 */
+		Flux<T> all();
+	}
+
 	/**
 	 * {@link ReactiveFind} provides methods for constructing lookup operations in a fluent way.
 	 */
-	interface ReactiveFind<T> extends FindWithCollection<T>, FindWithProjection<T> {}
+	interface ReactiveFind<T> extends FindWithCollection<T>, FindWithProjection<T>, FindDistinct {}
 }
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupport.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupport.java
index dda069778..d72f30505 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupport.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupport.java
@@ -19,7 +19,6 @@ import lombok.AccessLevel;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import lombok.experimental.FieldDefaults;
-import org.springframework.lang.Nullable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -28,6 +27,7 @@ import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.SerializationUtils;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -196,6 +196,18 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 			return template.exists(query, domainType, getCollectionName());
 		}
 
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ReactiveFindOperation.FindDistinct#distinct(java.lang.String)
+		 */
+		@Override
+		public TerminatingDistinct<Object> distinct(String field) {
+
+			Assert.notNull(field, "Field must not be null!");
+
+			return new DistinctOperationSupport<>(this, field);
+		}
+
 		private Flux<T> doFind(@Nullable FindPublisherPreparer preparer) {
 
 			Document queryObject = query.getQueryObject();
@@ -205,6 +217,12 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 					preparer != null ? preparer : getCursorPreparer(query));
 		}
 
+		private Flux<T> doFindDistinct(String field) {
+
+			return template.findDistinct(query, field, getCollectionName(), domainType,
+					returnType == domainType ? (Class<T>) Object.class : returnType);
+		}
+
 		private FindPublisherPreparer getCursorPreparer(Query query) {
 			return template.new QueryFindPublisherPreparer(query, domainType);
 		}
@@ -216,5 +234,54 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation {
 		private String asString() {
 			return SerializationUtils.serializeToJsonSafely(query);
 		}
+
+		/**
+		 * @author Christoph Strobl
+		 * @since 2.1
+		 */
+		static class DistinctOperationSupport<T> implements TerminatingDistinct<T> {
+
+			private final String field;
+			private final ReactiveFindSupport delegate;
+
+			public DistinctOperationSupport(ReactiveFindSupport delegate, String field) {
+
+				this.delegate = delegate;
+				this.field = field;
+			}
+
+			/*
+			 * (non-Javadoc)
+			 * @see org.springframework.data.mongodb.core.ReactiveFindOperation.DistinctWithProjection#as(java.lang.Class)
+			 */
+			@Override
+			public <R> TerminatingDistinct<R> as(Class<R> resultType) {
+
+				Assert.notNull(resultType, "ResultType must not be null!");
+
+				return new DistinctOperationSupport((ReactiveFindSupport) delegate.as(resultType), field);
+			}
+
+			/*
+			 * (non-Javadoc)
+			 * @see org.springframework.data.mongodb.core.ReactiveFindOperation.DistinctWithQuery#matching(org.springframework.data.mongodb.core.query.Query)
+			 */
+			@Override
+			public TerminatingDistinct<T> matching(Query query) {
+
+				Assert.notNull(query, "Query must not be null!");
+
+				return new DistinctOperationSupport((ReactiveFindSupport) delegate.matching(query), field);
+			}
+
+			/*
+			 * (non-Javadoc)
+			 * @see org.springframework.data.mongodb.core..ReactiveFindOperation.TerminatingDistinct#all()
+			 */
+			@Override
+			public Flux<T> all() {
+				return delegate.doFindDistinct(field);
+			}
+		}
 	}
 }
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java
index 6d255edef..20161a02c 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java
@@ -377,6 +377,66 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
 	 */
 	<T> Mono<T> findById(Object id, Class<T> entityClass, String collectionName);
 
+	/**
+	 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view and
+	 * returns the results in a {@link List}.
+	 *
+	 * @param field the name of the field to inspect for distinct values. Must not be {@literal null}.
+	 * @param entityClass the domain type used for determining the actual {@link MongoCollection}. Must not be
+	 *          {@literal null}.
+	 * @param resultClass the result type. Must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	default <T> Flux<T> findDistinct(String field, Class<?> entityClass, Class<T> resultClass) {
+		return findDistinct(new Query(), field, entityClass, resultClass);
+	}
+
+	/**
+	 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view and
+	 * returns the results in a {@link List}.
+	 *
+	 * @param query filter {@link Query} to restrict search. Must not be {@literal null}.
+	 * @param field the name of the field to inspect for distinct values. Must not be {@literal null}.
+	 * @param entityClass the domain type used for determining the actual {@link MongoCollection} and mapping the
+	 *          {@link Query} to the domain type fields. Must not be {@literal null}.
+	 * @param resultClass the result type. Must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	<T> Flux<T> findDistinct(Query query, String field, Class<?> entityClass, Class<T> resultClass);
+
+	/**
+	 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view and
+	 * returns the results in a {@link List}.
+	 *
+	 * @param query filter {@link Query} to restrict search. Must not be {@literal null}.
+	 * @param field the name of the field to inspect for distinct values. Must not be {@literal null}.
+	 * @param collectionName the explicit name of the actual {@link MongoCollection}. Must not be {@literal null}.
+	 * @param entityClass the domain type used for mapping the {@link Query} to the domain type fields.
+	 * @param resultClass the result type. Must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	<T> Flux<T> findDistinct(Query query, String field, String collectionName, Class<?> entityClass,
+			Class<T> resultClass);
+
+	/**
+	 * Finds the distinct values for a specified {@literal field} across a single {@link MongoCollection} or view and
+	 * returns the results in a {@link List}.
+	 *
+	 * @param query filter {@link Query} to restrict search. Must not be {@literal null}.
+	 * @param field the name of the field to inspect for distinct values. Must not be {@literal null}.
+	 * @param collection the explicit name of the actual {@link MongoCollection}. Must not be {@literal null}.
+	 * @param resultClass the result type. Must not be {@literal null}.
+	 * @param <T>
+	 * @return
+	 * @since 2.1
+	 */
+	default <T> Flux<T> findDistinct(Query query, String field, String collection, Class<T> resultClass) {
+		return findDistinct(query, field, collection, Object.class, resultClass);
+	}
+
 	/**
 	 * Execute an aggregation operation.
 	 * <p>
@@ -613,8 +673,8 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
 	 * If you object has an "Id' property, it will be set with the generated Id from MongoDB. If your Id property is a
 	 * String then MongoDB ObjectId will be used to populate that string. Otherwise, the conversion from ObjectId to your
 	 * property type will be handled by Spring's BeanWrapper class that leverages Type Conversion API. See
-	 * <a href="http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation" >
-	 * Spring's Type Conversion"</a> for more details.
+	 * <a href="http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation" > Spring's Type
+	 * Conversion"</a> for more details.
 	 * <p/>
 	 * <p/>
 	 * Insert is used to initially store the object into the database. To update an existing object use the save method.
@@ -673,8 +733,8 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
 	 * If you object has an "Id' property, it will be set with the generated Id from MongoDB. If your Id property is a
 	 * String then MongoDB ObjectId will be used to populate that string. Otherwise, the conversion from ObjectId to your
 	 * property type will be handled by Spring's BeanWrapper class that leverages Type Conversion API. See
-	 * <a href="http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation" >
-	 * Spring's Type Conversion"</a> for more details.
+	 * <a href="http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation" > Spring's Type
+	 * Conversion"</a> for more details.
 	 * <p/>
 	 * <p/>
 	 * Insert is used to initially store the object into the database. To update an existing object use the save method.
@@ -721,8 +781,8 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
 	 * If you object has an "Id' property, it will be set with the generated Id from MongoDB. If your Id property is a
 	 * String then MongoDB ObjectId will be used to populate that string. Otherwise, the conversion from ObjectId to your
 	 * property type will be handled by Spring's BeanWrapper class that leverages Type Conversion API. See
-	 * <a href="http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation" >
-	 * Spring's Type Conversion"</a> for more details.
+	 * <a href="http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation" > Spring's Type
+	 * Conversion"</a> for more details.
 	 *
 	 * @param objectToSave the object to store in the collection. Must not be {@literal null}.
 	 * @return the saved object.
@@ -739,8 +799,8 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
 	 * If you object has an "Id' property, it will be set with the generated Id from MongoDB. If your Id property is a
 	 * String then MongoDB ObjectId will be used to populate that string. Otherwise, the conversion from ObjectId to your
 	 * property type will be handled by Spring's BeanWrapper class that leverages Type Conversion API. See <a
-	 * http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation">Spring's
-	 * Type Conversion"</a> for more details.
+	 * http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation">Spring's Type
+	 * Conversion"</a> for more details.
 	 *
 	 * @param objectToSave the object to store in the collection. Must not be {@literal null}.
 	 * @param collectionName name of the collection to store the object in. Must not be {@literal null}.
@@ -758,8 +818,8 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
 	 * If you object has an "Id' property, it will be set with the generated Id from MongoDB. If your Id property is a
 	 * String then MongoDB ObjectId will be used to populate that string. Otherwise, the conversion from ObjectId to your
 	 * property type will be handled by Spring's BeanWrapper class that leverages Type Conversion API. See
-	 * <a href="http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation" >
-	 * Spring's Type Conversion"</a> for more details.
+	 * <a href="http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation" > Spring's Type
+	 * Conversion"</a> for more details.
 	 *
 	 * @param objectToSave the object to store in the collection. Must not be {@literal null}.
 	 * @return the saved object.
@@ -776,8 +836,8 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
 	 * If you object has an "Id' property, it will be set with the generated Id from MongoDB. If your Id property is a
 	 * String then MongoDB ObjectId will be used to populate that string. Otherwise, the conversion from ObjectId to your
 	 * property type will be handled by Spring's BeanWrapper class that leverages Type Conversion API. See <a
-	 * http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation">Spring's
-	 * Type Conversion"</a> for more details.
+	 * http://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#validation">Spring's Type
+	 * Conversion"</a> for more details.
 	 *
 	 * @param objectToSave the object to store in the collection. Must not be {@literal null}.
 	 * @param collectionName name of the collection to store the object in. Must not be {@literal null}.
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java
index 5a67d8e9c..ba8af473c 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java
@@ -40,7 +40,9 @@ import java.util.stream.Collectors;
 
 import javax.annotation.Nonnull;
 
+import org.bson.BsonValue;
 import org.bson.Document;
+import org.bson.codecs.Codec;
 import org.bson.conversions.Bson;
 import org.bson.types.ObjectId;
 import org.reactivestreams.Publisher;
@@ -65,6 +67,8 @@ import org.springframework.data.geo.GeoResult;
 import org.springframework.data.geo.Metric;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.mapping.PropertyPath;
+import org.springframework.data.mapping.PropertyReferenceException;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
 import org.springframework.data.mongodb.MongoDbFactory;
@@ -133,6 +137,7 @@ import com.mongodb.client.model.UpdateOptions;
 import com.mongodb.client.result.DeleteResult;
 import com.mongodb.client.result.UpdateResult;
 import com.mongodb.reactivestreams.client.AggregatePublisher;
+import com.mongodb.reactivestreams.client.DistinctPublisher;
 import com.mongodb.reactivestreams.client.FindPublisher;
 import com.mongodb.reactivestreams.client.MongoClient;
 import com.mongodb.reactivestreams.client.MongoCollection;
@@ -675,6 +680,77 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		return doFindOne(collectionName, new Document(idKey, id), null, entityClass, null);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findDistinct(org.springframework.data.mongodb.core.query.Query, java.lang.String, java.lang.Class, java.lang.Class)
+	 */
+	public <T> Flux<T> findDistinct(Query query, String field, Class<?> entityClass, Class<T> resultClass) {
+		return findDistinct(query, field, determineCollectionName(entityClass), entityClass, resultClass);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findDistinct(org.springframework.data.mongodb.core.query.Query, java.lang.String, java.lang.String, java.lang.Class, java.lang.Class)
+	 */
+	public <T> Flux<T> findDistinct(Query query, String field, String collectionName, Class<?> entityClass,
+			Class<T> resultClass) {
+
+		Assert.notNull(query, "Query must not be null!");
+		Assert.notNull(field, "Field must not be null!");
+		Assert.notNull(collectionName, "CollectionName must not be null!");
+		Assert.notNull(entityClass, "EntityClass must not be null!");
+		Assert.notNull(resultClass, "ResultClass must not be null!");
+
+		MongoPersistentEntity<?> entity = getPersistentEntity(entityClass);
+
+		Document mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), entity);
+		String mappedFieldName = queryMapper.getMappedFields(new Document(field, 1), entity).keySet().iterator().next();
+
+		Class<?> mongoDriverCompatibleType = mongoDatabaseFactory.getCodecFor(resultClass).map(Codec::getEncoderClass)
+				.orElse((Class) BsonValue.class);
+
+		Flux result = execute(collectionName, collection -> {
+
+			DistinctPublisher publisher = collection.distinct(mappedFieldName, mappedQuery, mongoDriverCompatibleType);
+
+			return query.getCollation().isPresent()
+					? publisher.collation(query.getCollation().map(Collation::toMongoCollation).get()) : publisher;
+		});
+
+		if (resultClass == Object.class || mongoDriverCompatibleType != resultClass) {
+			result = result.map(
+					getConverter().mapValueToTargetType(getMostSpecificConversionTargetType(resultClass, entityClass, field), NO_OP_REF_RESOLVER));
+		}
+
+		return result;
+	}
+
+	/**
+	 * @param userType must not be {@literal null}.
+	 * @param domainType must not be {@literal null}.
+	 * @param field must not be {@literal null}.
+	 * @return the most specific conversion target type depending on user preference and domain type property.
+	 * @since 2.1
+	 */
+	private Class<?> getMostSpecificConversionTargetType(Class<?> userType, Class<?> domainType, String field) {
+
+		Class<?> conversionTargetType = userType;
+		try {
+
+			Class<?> propertyType = PropertyPath.from(field, domainType).getLeafProperty().getLeafType();
+
+			// use the more specific type but favor UserType over property one
+			if (ClassUtils.isAssignable(userType, propertyType)) {
+				conversionTargetType = propertyType;
+			}
+
+		} catch (PropertyReferenceException e) {
+			// just don't care about it as we default to Object.class anyway.
+		}
+
+		return conversionTargetType;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.TypedAggregation, java.lang.String, java.lang.Class)
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/MongoConverter.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/MongoConverter.java
index 36d4ae5dc..442f4cc6a 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/MongoConverter.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/MongoConverter.java
@@ -15,6 +15,9 @@
  */
 package org.springframework.data.mongodb.core.convert;
 
+import java.util.function.Function;
+
+import org.bson.BsonValue;
 import org.bson.Document;
 import org.bson.conversions.Bson;
 import org.springframework.data.convert.EntityConverter;
@@ -22,6 +25,11 @@ import org.springframework.data.convert.EntityReader;
 import org.springframework.data.convert.TypeMapper;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
+import org.springframework.data.mongodb.util.BsonUtils;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+import com.mongodb.DBRef;
 
 /**
  * Central Mongo specific converter interface which combines {@link MongoWriter} and {@link MongoReader}.
@@ -41,4 +49,59 @@ public interface MongoConverter
 	 * @return will never be {@literal null}.
 	 */
 	MongoTypeMapper getTypeMapper();
+
+	/**
+	 * Mapping function capable of converting values into a desired target type by eg. extracting the actual java type
+	 * from a given {@link BsonValue}.
+	 *
+	 * @param targetType must not be {@literal null}.
+	 * @param dbRefResolver must not be {@literal null}.
+	 * @param <S>
+	 * @param <T>
+	 * @return new typed {@link com.mongodb.Function}.
+	 * @throws IllegalArgumentException if {@literal targetType} is {@literal null}.
+	 * @since 2.1
+	 */
+	default <S, T> Function<S, T> mapValueToTargetType(Class<T> targetType, DbRefResolver dbRefResolver) {
+
+		Assert.notNull(targetType, "TargetType must not be null!");
+		Assert.notNull(dbRefResolver, "DbRefResolver must not be null!");
+
+		return (source) -> {
+
+			if (targetType != Object.class && ClassUtils.isAssignable(targetType, source.getClass())) {
+				return (T) source;
+			}
+
+			if (source instanceof BsonValue) {
+
+				Object value = BsonUtils.toJavaType((BsonValue) source);
+
+				if (value instanceof Document) {
+
+					Document sourceDocument = (Document) value;
+
+					if (sourceDocument.containsKey("$ref") && sourceDocument.containsKey("$id")) {
+
+						sourceDocument = dbRefResolver
+								.fetch(new DBRef(sourceDocument.getString("$ref"), sourceDocument.get("$id")));
+						if (sourceDocument == null) {
+							return null;
+						}
+					}
+
+					return read(targetType, sourceDocument);
+				} else {
+					if (!ClassUtils.isAssignable(targetType, value.getClass())) {
+						if (getConversionService().canConvert(value.getClass(), targetType)) {
+							return getConversionService().convert(value, targetType);
+						}
+					}
+				}
+
+				return (T) value;
+			}
+			return (T) getConversionService().convert(source, targetType);
+		};
+	}
 }
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/BsonUtils.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/BsonUtils.java
index 1a50d3e09..29d09524d 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/BsonUtils.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/BsonUtils.java
@@ -15,14 +15,17 @@
  */
 package org.springframework.data.mongodb.util;
 
+import java.util.Date;
 import java.util.Map;
 
+import org.bson.BsonValue;
 import org.bson.Document;
 import org.bson.conversions.Bson;
 import org.springframework.lang.Nullable;
 
 import com.mongodb.BasicDBObject;
 import com.mongodb.DBObject;
+import com.mongodb.DBRef;
 
 /**
  * @author Christoph Strobl
@@ -59,4 +62,46 @@ public class BsonUtils {
 		}
 		throw new IllegalArgumentException("o_O what's that? Cannot add value to " + bson.getClass());
 	}
+
+	/**
+	 * Extract the corresponding plain value from {@link BsonValue}. Eg. plain {@link String} from
+	 * {@link org.bson.BsonString}.
+	 *
+	 * @param value must not be {@literal null}.
+	 * @return
+	 * @since 2.1
+	 */
+	public static Object toJavaType(BsonValue value) {
+
+		switch (value.getBsonType()) {
+			case INT32:
+				return value.asInt32().getValue();
+			case INT64:
+				return value.asInt64().getValue();
+			case STRING:
+				return value.asString().getValue();
+			case DECIMAL128:
+				return value.asDecimal128().doubleValue();
+			case DOUBLE:
+				return value.asDouble().getValue();
+			case BOOLEAN:
+				return value.asBoolean().getValue();
+			case OBJECT_ID:
+				return value.asObjectId().getValue();
+			case DB_POINTER:
+				return new DBRef(value.asDBPointer().getNamespace(), value.asDBPointer().getId());
+			case BINARY:
+				return value.asBinary().getData();
+			case DATE_TIME:
+				return new Date(value.asDateTime().getValue());
+			case SYMBOL:
+				return value.asSymbol().getSymbol();
+			case ARRAY:
+				return value.asArray().toArray();
+			case DOCUMENT:
+				return Document.parse(value.asDocument().toJson());
+			default:
+				return value;
+		}
+	}
 }
diff --git a/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ExecutableFindOperationExtensions.kt b/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ExecutableFindOperationExtensions.kt
index f51621f5d..83585111f 100644
--- a/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ExecutableFindOperationExtensions.kt
+++ b/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ExecutableFindOperationExtensions.kt
@@ -25,7 +25,7 @@ import kotlin.reflect.KClass
  * @since 2.0
  */
 fun <T : Any> ExecutableFindOperation.query(entityClass: KClass<T>): ExecutableFindOperation.ExecutableFind<T> =
-        query(entityClass.java)
+		query(entityClass.java)
 
 /**
  * Extension for [ExecutableFindOperation.query] leveraging reified type parameters.
@@ -35,7 +35,7 @@ fun <T : Any> ExecutableFindOperation.query(entityClass: KClass<T>): ExecutableF
  * @since 2.0
  */
 inline fun <reified T : Any> ExecutableFindOperation.query(): ExecutableFindOperation.ExecutableFind<T> =
-        query(T::class.java)
+		query(T::class.java)
 
 
 /**
@@ -46,7 +46,7 @@ inline fun <reified T : Any> ExecutableFindOperation.query(): ExecutableFindOper
  * @since 2.0
  */
 fun <T : Any> ExecutableFindOperation.FindWithProjection<T>.asType(resultType: KClass<T>): ExecutableFindOperation.FindWithQuery<T> =
-        `as`(resultType.java)
+		`as`(resultType.java)
 
 /**
  * Extension for [ExecutableFindOperation.FindWithProjection. as] leveraging reified type parameters.
@@ -56,6 +56,13 @@ fun <T : Any> ExecutableFindOperation.FindWithProjection<T>.asType(resultType: K
  * @since 2.0
  */
 inline fun <reified T : Any> ExecutableFindOperation.FindWithProjection<T>.asType(): ExecutableFindOperation.FindWithQuery<T> =
-        `as`(T::class.java)
-
+		`as`(T::class.java)
 
+/**
+ * Extension for [ExecutableFindOperation.DistinctWithProjection. as] providing a [KClass] based variant.
+ *
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+fun <T : Any> ExecutableFindOperation.DistinctWithProjection.asType(resultType: KClass<T>): ExecutableFindOperation.TerminatingDistinct<T> =
+		`as`(resultType.java);
diff --git a/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/MongoOperationsExtensions.kt b/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/MongoOperationsExtensions.kt
index df29d436d..e0bb3ec85 100644
--- a/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/MongoOperationsExtensions.kt
+++ b/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/MongoOperationsExtensions.kt
@@ -322,6 +322,42 @@ inline fun <reified T : Any> MongoOperations.findById(id: Any, collectionName: S
 		if (collectionName != null) findById(id, T::class.java, collectionName)
 		else findById(id, T::class.java)
 
+/**
+ * Extension for [MongoOperations.findDistinct] leveraging reified type parameters.
+ *
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+inline fun <reified T : Any> MongoOperations.findDistinct(field: String, entityClass: KClass<*>): List<T> =
+		findDistinct(field, entityClass.java, T::class.java);
+
+/**
+ * Extension for [MongoOperations.findDistinct] leveraging reified type parameters.
+ *
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+inline fun <reified T : Any> MongoOperations.findDistinct(query: Query, field: String, entityClass: KClass<*>): List<T> =
+		findDistinct(query, field, entityClass.java, T::class.java);
+
+/**
+ * Extension for [MongoOperations.findDistinct] leveraging reified type parameters.
+ *
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+inline fun <reified T : Any> MongoOperations.findDistinct(query: Query, field: String, collectionName: String?, entityClass: KClass<*>): List<T> =
+		findDistinct(query, field, collectionName, entityClass.java, T::class.java);
+
+/**
+ * Extension for [MongoOperations.findDistinct] leveraging reified type parameters.
+ *
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+inline fun <reified T : Any> MongoOperations.findDistinct(query: Query, field: String, collectionName: String?): List<T> =
+		findDistinct(query, field, collectionName, T::class.java);
+
 /**
  * Extension for [MongoOperations.findAndModify] leveraging reified type parameters.
  *
diff --git a/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveFindOperationExtensions.kt b/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveFindOperationExtensions.kt
index 354bae8e5..37531f4ae 100644
--- a/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveFindOperationExtensions.kt
+++ b/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveFindOperationExtensions.kt
@@ -53,4 +53,11 @@ fun <T : Any> ReactiveFindOperation.FindWithProjection<T>.asType(resultType: KCl
 inline fun <reified T : Any> ReactiveFindOperation.FindWithProjection<T>.asType(): ReactiveFindOperation.FindWithQuery<T> =
 		`as`(T::class.java)
 
-
+/**
+ * Extension for [ExecutableFindOperation.DistinctWithProjection. as] providing a [KClass] based variant.
+ *
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+fun <T : Any> ReactiveFindOperation.DistinctWithProjection.asType(resultType: KClass<T>): ReactiveFindOperation.TerminatingDistinct<T> =
+		`as`(resultType.java);
diff --git a/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveMongoOperationsExtensions.kt b/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveMongoOperationsExtensions.kt
index 366fdafc6..e06d0f40e 100644
--- a/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveMongoOperationsExtensions.kt
+++ b/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveMongoOperationsExtensions.kt
@@ -165,6 +165,42 @@ inline fun <reified T : Any> ReactiveMongoOperations.find(query: Query, collecti
 inline fun <reified T : Any> ReactiveMongoOperations.findById(id: Any, collectionName: String? = null): Mono<T> =
 		if (collectionName != null) findById(id, T::class.java, collectionName) else findById(id, T::class.java)
 
+/**
+ * Extension for [MongoOperations.findDistinct] leveraging reified type parameters.
+ *
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+inline fun <reified T : Any> ReactiveMongoOperations.findDistinct(field: String, entityClass: KClass<*>): Flux<T> =
+		findDistinct(field, entityClass.java, T::class.java);
+
+/**
+ * Extension for [MongoOperations.findDistinct] leveraging reified type parameters.
+ *
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+inline fun <reified T : Any> ReactiveMongoOperations.findDistinct(query: Query, field: String, entityClass: KClass<*>): Flux<T> =
+		findDistinct(query, field, entityClass.java, T::class.java);
+
+/**
+ * Extension for [MongoOperations.findDistinct] leveraging reified type parameters.
+ *
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+inline fun <reified T : Any> ReactiveMongoOperations.findDistinct(query: Query, field: String, collectionName: String?, entityClass: KClass<*>): Flux<T> =
+		findDistinct(query, field, collectionName, entityClass.java, T::class.java);
+
+/**
+ * Extension for [MongoOperations.findDistinct] leveraging reified type parameters.
+ *
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+inline fun <reified T : Any> ReactiveMongoOperations.findDistinct(query: Query, field: String, collectionName: String?): Flux<T> =
+		findDistinct(query, field, collectionName, T::class.java);
+
 /**
  * Extension for [ReactiveMongoOperations.geoNear] leveraging reified type parameters.
  *
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupportTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupportTests.java
index ffa8b15a2..fd7d3407c 100644
--- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupportTests.java
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupportTests.java
@@ -21,13 +21,19 @@ import static org.springframework.data.mongodb.core.query.Query.*;
 
 import lombok.AllArgsConstructor;
 import lombok.Data;
+import lombok.NoArgsConstructor;
 
+import java.util.Date;
 import java.util.stream.Stream;
 
+import org.bson.BsonString;
+import org.bson.BsonValue;
+import org.bson.Document;
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.geo.GeoResults;
 import org.springframework.data.geo.Point;
@@ -347,13 +353,170 @@ public class ExecutableFindOperationSupportTests {
 		assertThat(template.query(Person.class).as(Contact.class).all()).allMatch(it -> it instanceof Person);
 	}
 
+	@Test // DATAMONGO-1761
+	public void distinctReturnsEmptyListIfNoMatchFound() {
+		assertThat(template.query(Person.class).distinct("actually-not-property-in-use").as(String.class).all()).isEmpty();
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsSimpleFieldValuesCorrectlyForCollectionHavingReturnTypeSpecifiedThatCanBeConvertedDirectlyByACodec() {
+
+		Person anakin = new Person();
+		anakin.firstname = "anakin";
+		anakin.lastname = luke.lastname;
+
+		template.save(anakin);
+
+		assertThat(template.query(Person.class).distinct("lastname").as(String.class).all())
+				.containsExactlyInAnyOrder("solo", "skywalker");
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsSimpleFieldValuesCorrectlyForCollectionHavingNoReturnTypeSpecified() {
+
+		Person anakin = new Person();
+		anakin.firstname = "anakin";
+		anakin.ability = "dark-lord";
+
+		Person padme = new Person();
+		padme.firstname = "padme";
+		padme.ability = 42L;
+
+		Person jaja = new Person();
+		jaja.firstname = "jaja";
+		jaja.ability = new Date();
+
+		template.save(anakin);
+		template.save(padme);
+		template.save(jaja);
+
+		assertThat(template.query(Person.class).distinct("ability").all()).containsExactlyInAnyOrder(anakin.ability,
+				padme.ability, jaja.ability);
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsComplexValuesCorrectlyForCollectionHavingNoReturnTypeSpecified() {
+
+		Sith sith = new Sith();
+		sith.rank = "lord";
+
+		Person anakin = new Person();
+		anakin.firstname = "anakin";
+		anakin.ability = sith;
+
+		template.save(anakin);
+
+		assertThat(template.query(Person.class).distinct("ability").all()).containsExactlyInAnyOrder(anakin.ability);
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsComplexValuesCorrectlyForCollectionHavingReturnTypeSpecified() {
+
+		Sith sith = new Sith();
+		sith.rank = "lord";
+
+		Person anakin = new Person();
+		anakin.firstname = "anakin";
+		anakin.ability = sith;
+
+		template.save(anakin);
+
+		assertThat(template.query(Person.class).distinct("ability").as(Sith.class).all())
+				.containsExactlyInAnyOrder((Sith) anakin.ability);
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsComplexValuesCorrectlyForCollectionHavingReturnTypeDocumentSpecified() {
+
+		Sith sith = new Sith();
+		sith.rank = "lord";
+
+		Person anakin = new Person();
+		anakin.firstname = "anakin";
+		anakin.ability = sith;
+
+		template.save(anakin);
+
+		assertThat(template.query(Person.class).distinct("ability").as(Document.class).all())
+				.containsExactlyInAnyOrder(new Document("rank", "lord").append("_class", Sith.class.getName()));
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctMapsFieldNameCorrectly() {
+
+		assertThat(template.query(Jedi.class).inCollection(STAR_WARS).distinct("name").as(String.class).all())
+				.containsExactlyInAnyOrder("han", "luke");
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsRawValuesIfReturnTypeIsBsonValue() {
+
+		assertThat(template.query(Person.class).distinct("lastname").as(BsonValue.class).all())
+				.containsExactlyInAnyOrder(new BsonString("solo"), new BsonString("skywalker"));
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsValuesMappedToTheirJavaTypeEvenWhenNotExplicitlyDefinedByTheDomainType() {
+
+		template.save(new Document("darth", "vader"), STAR_WARS);
+
+		assertThat(template.query(Person.class).distinct("darth").all()).containsExactlyInAnyOrder("vader");
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsMappedDomainTypeForProjections() {
+
+		luke.father = new Person();
+		luke.father.firstname = "anakin";
+
+		template.save(luke);
+
+		assertThat(template.query(Person.class).distinct("father").as(Jedi.class).all())
+				.containsExactlyInAnyOrder(new Jedi("anakin"));
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctAlllowsQueryUsingObjectSourceType() {
+
+		luke.father = new Person();
+		luke.father.firstname = "anakin";
+
+		template.save(luke);
+
+		assertThat(template.query(Object.class).inCollection(STAR_WARS).distinct("father").as(Jedi.class).all())
+				.containsExactlyInAnyOrder(new Jedi("anakin"));
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsMappedDomainTypeExtractedFromPropertyWhenNoExplicitTypePresent() {
+
+		luke.father = new Person();
+		luke.father.firstname = "anakin";
+
+		template.save(luke);
+
+		Person expected = new Person();
+		expected.firstname = luke.father.firstname;
+
+		assertThat(template.query(Person.class).distinct("father").all()).containsExactlyInAnyOrder(expected);
+	}
+
+	@Test(expected = InvalidDataAccessApiUsageException.class) // DATAMONGO-1761
+	public void distinctThrowsExceptionWhenExplicitMappingTypeCannotBeApplied() {
+		template.query(Person.class).distinct("firstname").as(Long.class).all();
+	}
+
 	interface Contact {}
 
 	@Data
 	@org.springframework.data.mongodb.core.mapping.Document(collection = STAR_WARS)
 	static class Person implements Contact {
+
 		@Id String id;
 		String firstname;
+		String lastname;
+		Object ability;
+		Person father;
 	}
 
 	interface PersonProjection {
@@ -372,11 +535,19 @@ public class ExecutableFindOperationSupportTests {
 	}
 
 	@Data
+	@AllArgsConstructor
+	@NoArgsConstructor
 	static class Jedi {
 
 		@Field("firstname") String name;
 	}
 
+	@Data
+	static class Sith {
+
+		String rank;
+	}
+
 	@Data
 	@AllArgsConstructor
 	@org.springframework.data.mongodb.core.mapping.Document(collection = STAR_WARS_PLANETS)
@@ -400,10 +571,12 @@ public class ExecutableFindOperationSupportTests {
 
 		han = new Person();
 		han.firstname = "han";
+		han.lastname = "solo";
 		han.id = "id-1";
 
 		luke = new Person();
 		luke.firstname = "luke";
+		luke.lastname = "skywalker";
 		luke.id = "id-2";
 
 		template.save(han);
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java
index d2f60e04a..6eb9f202a 100644
--- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java
@@ -727,8 +727,9 @@ public class MongoTemplateTests {
 		assertThat(notFound, nullValue());
 	}
 
-	@Test
+	@Test // DATAMONGO-1761
 	public void testDistinct() {
+
 		Address address1 = new Address();
 		address1.state = "PA";
 		address1.city = "Philadelphia";
@@ -748,16 +749,51 @@ public class MongoTemplateTests {
 		template.save(person1);
 		template.save(person2);
 
-		List<String> nameList = template.distinct("name", MyPerson.class, String.class);
-		assertTrue(nameList.containsAll(Arrays.asList(person1.getName(), person2.getName())));
+		assertThat(template.findDistinct("name", MyPerson.class, String.class)).containsExactlyInAnyOrder(person1.getName(),
+				person2.getName());
+		assertThat(template.findDistinct(new BasicQuery("{'address.state' : 'PA'}"), "name", MyPerson.class, String.class))
+				.containsExactlyInAnyOrder(person1.getName(), person2.getName());
+		assertThat(template.findDistinct(new BasicQuery("{'address.state' : 'PA'}"), "name",
+				template.determineCollectionName(MyPerson.class), MyPerson.class, String.class))
+						.containsExactlyInAnyOrder(person1.getName(), person2.getName());
+	}
 
-		Query query = new BasicQuery("{'address.state' : 'PA'}");
-		nameList = template.distinct(query, "name", MyPerson.class, String.class);
-		assertTrue(nameList.containsAll(Arrays.asList(person1.getName(), person2.getName())));
+	@Test // DATAMONGO-1761
+	public void testDistinctCovertsResultToPropertyTargetTypeCorrectly() {
 
-		String collectionName = template.determineCollectionName(MyPerson.class);
-		nameList = template.distinct(query, "name", collectionName, String.class);
-		assertTrue(nameList.containsAll(Arrays.asList(person1.getName(), person2.getName())));
+		template.insert(new Person("garvin"));
+
+		assertThat(template.findDistinct("firstName", Person.class, Object.class))
+				.allSatisfy(val -> instanceOf(String.class));
+	}
+
+	@Test // DATAMONGO-1761
+	public void testDistinctResolvesDbRefsCorrectly() {
+
+		SomeContent content1 = new SomeContent();
+		content1.text = "content-1";
+
+		SomeContent content2 = new SomeContent();
+		content2.text = "content-2";
+
+		template.save(content1);
+		template.save(content2);
+
+		SomeTemplate t1 = new SomeTemplate();
+		t1.content = content1;
+
+		SomeTemplate t2 = new SomeTemplate();
+		t2.content = content2;
+
+		SomeTemplate t3 = new SomeTemplate();
+		t3.content = content2;
+
+		template.insert(t1);
+		template.insert(t2);
+		template.insert(t3);
+
+		assertThat(template.findDistinct("content", SomeTemplate.class, SomeContent.class))
+				.containsExactlyInAnyOrder(content1, content2);
 	}
 
 	@Test
@@ -3609,6 +3645,7 @@ public class MongoTemplateTests {
 		}
 	}
 
+	@EqualsAndHashCode
 	public static class SomeContent {
 
 		String id;
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupportTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupportTests.java
index d4272f407..fdea09cda 100644
--- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupportTests.java
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupportTests.java
@@ -21,12 +21,20 @@ import static org.springframework.data.mongodb.core.query.Query.*;
 
 import lombok.AllArgsConstructor;
 import lombok.Data;
+import lombok.NoArgsConstructor;
 import reactor.test.StepVerifier;
 
+import java.util.Date;
+import java.util.function.Consumer;
+
+import org.bson.BsonString;
+import org.bson.BsonValue;
+import org.bson.Document;
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.geo.Point;
 import org.springframework.data.mongodb.core.index.GeoSpatialIndexType;
@@ -61,10 +69,12 @@ public class ReactiveFindOperationSupportTests {
 
 		han = new Person();
 		han.firstname = "han";
+		han.lastname = "solo";
 		han.id = "id-1";
 
 		luke = new Person();
 		luke.firstname = "luke";
+		luke.lastname = "skywalker";
 		luke.id = "id-2";
 
 		blocking.save(han);
@@ -310,13 +320,177 @@ public class ReactiveFindOperationSupportTests {
 				.expectNext(false).verifyComplete();
 	}
 
+	@Test // DATAMONGO-1761
+	public void distinctReturnsEmptyListIfNoMatchFound() {
+
+		StepVerifier.create(template.query(Person.class).distinct("actually-not-property-in-use").as(String.class).all())
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsSimpleFieldValuesCorrectlyForCollectionHavingReturnTypeSpecifiedThatCanBeConvertedDirectlyByACodec() {
+
+		Person anakin = new Person();
+		anakin.firstname = "anakin";
+		anakin.lastname = luke.lastname;
+
+		blocking.save(anakin);
+
+		StepVerifier.create(template.query(Person.class).distinct("lastname").as(String.class).all())
+				.assertNext(in("solo", "skywalker")).assertNext(in("solo", "skywalker")).verifyComplete();
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsSimpleFieldValuesCorrectlyForCollectionHavingNoReturnTypeSpecified() {
+
+		Person anakin = new Person();
+		anakin.firstname = "anakin";
+		anakin.ability = "dark-lord";
+
+		Person padme = new Person();
+		padme.firstname = "padme";
+		padme.ability = 42L;
+
+		Person jaja = new Person();
+		jaja.firstname = "jaja";
+		jaja.ability = new Date();
+
+		blocking.save(anakin);
+		blocking.save(padme);
+		blocking.save(jaja);
+
+		Consumer<Object> containedInAbilities = in(anakin.ability, padme.ability, jaja.ability);
+
+		StepVerifier.create(template.query(Person.class).distinct("ability").all()).assertNext(containedInAbilities)
+				.assertNext(containedInAbilities).assertNext(containedInAbilities).verifyComplete();
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsComplexValuesCorrectlyForCollectionHavingNoReturnTypeSpecified() {
+
+		Sith sith = new Sith();
+		sith.rank = "lord";
+
+		Person anakin = new Person();
+		anakin.firstname = "anakin";
+		anakin.ability = sith;
+
+		blocking.save(anakin);
+
+		StepVerifier.create(template.query(Person.class).distinct("ability").all()).expectNext(anakin.ability)
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsComplexValuesCorrectlyForCollectionHavingReturnTypeSpecified() {
+
+		Sith sith = new Sith();
+		sith.rank = "lord";
+
+		Person anakin = new Person();
+		anakin.firstname = "anakin";
+		anakin.ability = sith;
+
+		blocking.save(anakin);
+
+		StepVerifier.create(template.query(Person.class).distinct("ability").as(Sith.class).all()).expectNext(sith)
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsComplexValuesCorrectlyForCollectionHavingReturnTypeDocumentSpecified() {
+
+		Sith sith = new Sith();
+		sith.rank = "lord";
+
+		Person anakin = new Person();
+		anakin.firstname = "anakin";
+		anakin.ability = sith;
+
+		blocking.save(anakin);
+
+		StepVerifier.create(template.query(Person.class).distinct("ability").as(Document.class).all())
+				.expectNext(new Document("rank", "lord").append("_class", Sith.class.getName())).verifyComplete();
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctMapsFieldNameCorrectly() {
+
+		StepVerifier.create(template.query(Jedi.class).inCollection(STAR_WARS).distinct("name").as(String.class).all())
+				.assertNext(in("han", "luke")).assertNext(in("han", "luke")).verifyComplete();
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsRawValuesIfReturnTypeIsBsonValue() {
+
+		Consumer<BsonValue> inValues = in(new BsonString("solo"), new BsonString("skywalker"));
+		StepVerifier.create(template.query(Person.class).distinct("lastname").as(BsonValue.class).all())
+				.assertNext(inValues).assertNext(inValues).verifyComplete();
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsValuesMappedToTheirJavaTypeEvenWhenNotExplicitlyDefinedByTheDomainType() {
+
+		blocking.save(new Document("darth", "vader"), STAR_WARS);
+
+		StepVerifier.create(template.query(Person.class).distinct("darth").all()).expectNext("vader").verifyComplete();
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsMappedDomainTypeForProjections() {
+
+		luke.father = new Person();
+		luke.father.firstname = "anakin";
+
+		blocking.save(luke);
+
+		StepVerifier.create(template.query(Person.class).distinct("father").as(Jedi.class).all())
+				.expectNext(new Jedi("anakin")).verifyComplete();
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctAlllowsQueryUsingObjectSourceType() {
+
+		luke.father = new Person();
+		luke.father.firstname = "anakin";
+
+		blocking.save(luke);
+
+		StepVerifier.create(template.query(Object.class).inCollection(STAR_WARS).distinct("father").as(Jedi.class).all())
+				.expectNext(new Jedi("anakin")).verifyComplete();
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctReturnsMappedDomainTypeExtractedFromPropertyWhenNoExplicitTypePresent() {
+
+		luke.father = new Person();
+		luke.father.firstname = "anakin";
+
+		blocking.save(luke);
+
+		Person expected = new Person();
+		expected.firstname = luke.father.firstname;
+
+		StepVerifier.create(template.query(Person.class).distinct("father").all()).expectNext(expected).verifyComplete();
+	}
+
+	@Test // DATAMONGO-1761
+	public void distinctThrowsExceptionWhenExplicitMappingTypeCannotBeApplied() {
+		StepVerifier.create(template.query(Person.class).distinct("firstname").as(Long.class).all())
+				.expectError(InvalidDataAccessApiUsageException.class).verify();
+	}
+
 	interface Contact {}
 
 	@Data
 	@org.springframework.data.mongodb.core.mapping.Document(collection = STAR_WARS)
 	static class Person implements Contact {
+
 		@Id String id;
 		String firstname;
+		String lastname;
+		Object ability;
+		Person father;
 	}
 
 	interface PersonProjection {
@@ -335,11 +509,19 @@ public class ReactiveFindOperationSupportTests {
 	}
 
 	@Data
+	@NoArgsConstructor
+	@AllArgsConstructor
 	static class Jedi {
 
 		@Field("firstname") String name;
 	}
 
+	@Data
+	static class Sith {
+
+		String rank;
+	}
+
 	@Data
 	@AllArgsConstructor
 	@org.springframework.data.mongodb.core.mapping.Document(collection = STAR_WARS)
@@ -358,4 +540,10 @@ public class ReactiveFindOperationSupportTests {
 		@Value("#{target.name}")
 		String getId();
 	}
+
+	static <T> Consumer<T> in(T... values) {
+		return (val) -> {
+			assertThat(values).contains(val);
+		};
+	}
 }
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java
index 43c0cbc4f..cfc36441b 100644
--- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java
@@ -924,6 +924,21 @@ public class ReactiveMongoTemplateTests {
 		assertThat(documents.poll(1, TimeUnit.SECONDS), is(nullValue()));
 	}
 
+	@Test // DATAMONGO-1761
+	public void testDistinct() {
+
+		Person person1 = new Person("Christoph", 38);
+		Person person2 = new Person("Christine", 39);
+		Person person3 = new Person("Christoph", 37);
+
+		StepVerifier.create(template.save(person1)).expectNextCount(1).verifyComplete();
+		StepVerifier.create(template.save(person2)).expectNextCount(1).verifyComplete();
+		StepVerifier.create(template.save(person3)).expectNextCount(1).verifyComplete();
+
+		StepVerifier.create(template.findDistinct("firstName", Person.class, String.class)).expectNextCount(2)
+				.verifyComplete();
+	}
+
 	private PersonWithAList createPersonWithAList(String firstname, int age) {
 
 		PersonWithAList p = new PersonWithAList();
diff --git a/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ExecutableFindOperationExtensionsTests.kt b/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ExecutableFindOperationExtensionsTests.kt
index 7d61ee20b..4f8a60d34 100644
--- a/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ExecutableFindOperationExtensionsTests.kt
+++ b/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ExecutableFindOperationExtensionsTests.kt
@@ -36,6 +36,9 @@ class ExecutableFindOperationExtensionsTests {
     @Mock(answer = Answers.RETURNS_MOCKS)
     lateinit var operationWithProjection: ExecutableFindOperation.FindWithProjection<First>
 
+    @Mock(answer = Answers.RETURNS_MOCKS)
+    lateinit var distinctWithProjection: ExecutableFindOperation.DistinctWithProjection
+
     @Test // DATAMONGO-1689
     fun `ExecutableFindOperation#query(KClass) extension should call its Java counterpart`() {
 
@@ -64,4 +67,10 @@ class ExecutableFindOperationExtensionsTests {
         verify(operationWithProjection).`as`(First::class.java)
     }
 
+    @Test // DATAMONGO-1761
+    fun `ExecutableFindOperation#DistinctWithProjection#asType(KClass) extension should call its Java counterpart`() {
+
+        distinctWithProjection.asType(First::class)
+        verify(distinctWithProjection).`as`(First::class.java)
+    }
 }
diff --git a/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/MongoOperationsExtensionsTests.kt b/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/MongoOperationsExtensionsTests.kt
index da60730e2..76008cac5 100644
--- a/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/MongoOperationsExtensionsTests.kt
+++ b/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/MongoOperationsExtensionsTests.kt
@@ -678,4 +678,38 @@ class MongoOperationsExtensionsTests {
 		operations.findAllAndRemove<First>(query)
 		verify(operations).findAllAndRemove(query, First::class.java)
 	}
+
+	@Test // DATAMONGO-1761
+	fun `findDistinct(String, KClass) should call java counterpart`() {
+
+		operations.findDistinct<String>("field", First::class)
+		verify(operations).findDistinct("field", First::class.java, String::class.java)
+	}
+
+	@Test // DATAMONGO-1761
+	fun `findDistinct(Query, String, KClass) should call java counterpart`() {
+
+		val query = mock<Query>()
+
+		operations.findDistinct<String>(query, "field", First::class)
+		verify(operations).findDistinct(query, "field", First::class.java, String::class.java)
+	}
+
+	@Test // DATAMONGO-1761
+	fun `findDistinct(Query, String, String, KClass) should call java counterpart`() {
+
+		val query = mock<Query>()
+
+		operations.findDistinct<String>(query, "field", "collection", First::class)
+		verify(operations).findDistinct(query, "field", "collection", First::class.java, String::class.java)
+	}
+
+	@Test // DATAMONGO-1761
+	fun `findDistinct(Query, String, String) should call java counterpart`() {
+
+		val query = mock<Query>()
+
+		operations.findDistinct<String>(query, "field", "collection")
+		verify(operations).findDistinct(query, "field", "collection", String::class.java)
+	}
 }
diff --git a/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveFindOperationExtensionsTests.kt b/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveFindOperationExtensionsTests.kt
index fda680cb7..7e0ee6c97 100644
--- a/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveFindOperationExtensionsTests.kt
+++ b/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveFindOperationExtensionsTests.kt
@@ -35,6 +35,9 @@ class ReactiveFindOperationExtensionsTests {
 	@Mock(answer = Answers.RETURNS_MOCKS)
 	lateinit var operationWithProjection: ReactiveFindOperation.FindWithProjection<First>
 
+	@Mock(answer = Answers.RETURNS_MOCKS)
+	lateinit var distinctWithProjection: ReactiveFindOperation.DistinctWithProjection
+
 	@Test // DATAMONGO-1719
 	fun `ReactiveFind#query(KClass) extension should call its Java counterpart`() {
 
@@ -62,4 +65,11 @@ class ReactiveFindOperationExtensionsTests {
 		operationWithProjection.asType()
 		verify(operationWithProjection).`as`(First::class.java)
 	}
+
+	@Test // DATAMONGO-1761
+	fun `ReactiveFind#DistinctWithProjection#asType(KClass) extension should call its Java counterpart`() {
+
+		distinctWithProjection.asType(First::class)
+		verify(distinctWithProjection).`as`(First::class.java)
+	}
 }
diff --git a/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveMongoOperationsExtensionsTests.kt b/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveMongoOperationsExtensionsTests.kt
index 187da8224..b2f0ecaaf 100644
--- a/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveMongoOperationsExtensionsTests.kt
+++ b/spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveMongoOperationsExtensionsTests.kt
@@ -526,4 +526,38 @@ class ReactiveMongoOperationsExtensionsTests {
 		operations.tail<First>(query, collectionName)
 		verify(operations).tail(query, First::class.java, collectionName)
 	}
+
+	@Test // DATAMONGO-1761
+	fun `findDistinct(String, KClass) should call java counterpart`() {
+
+		operations.findDistinct<String>("field", First::class)
+		verify(operations).findDistinct("field", First::class.java, String::class.java)
+	}
+
+	@Test // DATAMONGO-1761
+	fun `findDistinct(Query, String, KClass) should call java counterpart`() {
+
+		val query = mock<Query>()
+
+		operations.findDistinct<String>(query, "field", First::class)
+		verify(operations).findDistinct(query, "field", First::class.java, String::class.java)
+	}
+
+	@Test // DATAMONGO-1761
+	fun `findDistinct(Query, String, String, KClass) should call java counterpart`() {
+
+		val query = mock<Query>()
+
+		operations.findDistinct<String>(query, "field", "collection", First::class)
+		verify(operations).findDistinct(query, "field", "collection", First::class.java, String::class.java)
+	}
+
+	@Test // DATAMONGO-1761
+	fun `findDistinct(Query, String, String) should call java counterpart`() {
+
+		val query = mock<Query>()
+
+		operations.findDistinct<String>(query, "field", "collection")
+		verify(operations).findDistinct(query, "field", "collection", String::class.java)
+	}
 }
diff --git a/src/main/asciidoc/reference/mongodb.adoc b/src/main/asciidoc/reference/mongodb.adoc
index de79c5d95..70ea97a32 100644
--- a/src/main/asciidoc/reference/mongodb.adoc
+++ b/src/main/asciidoc/reference/mongodb.adoc
@@ -1087,6 +1087,44 @@ The query methods need to specify the target type T that will be returned and th
 * *find* Map the results of an ad-hoc query on the collection to a List of the specified type.
 * *findAndRemove* Map the results of an ad-hoc query on the collection to a single instance of an object of the specified type. The first document that matches the query is returned and also removed from the collection in the database.
 
+[[mongo-template.query.distinct]]
+=== Query distinct values
+
+MongoDB allows obtaining distinct field values for a single field. The stored values do not have to have the same data type to be considered, nor is the feature limited to simple types.
+However when retrieving distinct values the actual result type does matter for the sake of conversion.
+
+.Retrieving distinct values
+====
+[source,java]
+----
+template.query(Person.class)  <1>
+  .distinct("lastname")       <2>
+  .all();                     <3>
+---
+<1> Query the collection of `Person`.
+<2> Select _distinct_  values of the `lastname` field. The fieldname will be mapped according to the domain types property declaration, taking potential `@Field` annotations into account.
+<3> Retrieve all distinct values as `List` of `Object` due to no explicit result type specification.
+====
+
+Retrieving distinct values into a `Collection` of `Object.class` is the most flexible way as it will try to determine the property value of the domain type converting results to the desired type or mapping `Document` structures.
+
+Sometimes, when all values of the desired field are fixed to a certain type, it is more convenient to directly obtain a correctly typed `Collection`
+
+.Retrieving strongly typed distinct values
+====
+[source,java]
+----
+template.query(Person.class)  <1>
+  .distinct("lastname")       <2>
+  .as(String.class)           <3>
+  .all();                     <4>
+---
+<1> Query the collection of `Person`.
+<2> Select _distinct_  values of the `lastname` field. The fieldname will be mapped according to the domain types property declaration, taking potential `@Field` annotations into account.
+<3> Retrieved values will be converted into the desired target type. In this case `String`. It would also be possible to map the values to a more complex type if the stored field contains a document.
+<4> Retrieve all distinct values as a `List` of `String`. Throws a `DataAccessException` if the type cannot be converted into the desired target type.
+===
+
 [[mongo.geospatial]]
 === GeoSpatial Queries