Remove Elasticsearch classes from suggest response data.

Original Pull Request #1940 
Closes #1302

src/main/asciidoc/reference/elasticsearch-migration-guide-4.2-4.3.adoc

@@ -21,6 +21,14 @@ Check the sections on <<elasticsearch-migration-guide-4.2-4.3.deprecations>> and
 [[elasticsearch-migration-guide-4.2-4.3.deprecations]]
 == Deprecations
 
+=== suggest methods
+
+In `SearchOperations`, and so in `ElasticsearchOperations` as well, the `suggest` methods taking a `org.elasticsearch.search.suggest.SuggestBuilder` as argument and returning a `org.elasticsearch.action.search.SearchResponse` have been deprecated.
+Use `SearchHits<T> search(Query query, Class<T> clazz)` instead, passing in a `NativeSearchQuery` which can contain a `SuggestBuilder` and read the suggest results from the returned `SearchHit<T>`.
+
+In `ReactiveSearchOperations` the new `suggest` methods return a `Mono<org.springframework.data.elasticsearch.core.suggest.response.Suggest>` now.
+Here as well the old methods are deprecated.
+
 [[elasticsearch-migration-guide-4.2-4.3.breaking-changes]]
 == Breaking Changes
 
@@ -59,3 +67,7 @@ Some properties of the `org.springframework.data.elasticsearch.core.query.BulkOp
 === IndicesOptions change
 
 Spring Data Elasticsearch now uses `org.springframework.data.elasticsearch.core.query.IndicesOptions` instead of `org.elasticsearch.action.support.IndicesOptions`.
+
+=== Completion classes
+
+The classes from the package `org.springframework.data.elasticsearch.core.completion` have been moved to `org.springframework.data.elasticsearch.core.suggest`.

src/main/asciidoc/reference/elasticsearch-operations.adoc

@@ -20,11 +20,11 @@ The default implementations of the interfaces offer:
 [NOTE]
 ====
 .Index management and automatic creation of indices and mappings.
+The `IndexOperations` interface and the provided implementation which can be obtained from an `ElasticsearchOperations` instance - for example with a call to `operations.indexOps(clazz)`- give the user the ability to create indices, put mappings or store template and alias information in the Elasticsearch cluster.
+Details of the index that will be created can be set by using the `@Setting` annotation, refer to <<elasticsearc.misc.index.settings>> for further information.
 
-The `IndexOperations` interface and the provided implementation which can be obtained from an `ElasticsearchOperations` instance - for example with a call to `operations.indexOps(clazz)`- give the user the ability to create indices, put mappings or store template and alias information in the Elasticsearch cluster. Details of the index that will be created
-can be set by using the `@Setting` annotation, refer to <<elasticsearc.misc.index.settings>> for further information.
-
-**None of these operations are done automatically** by the implementations of `IndexOperations` or `ElasticsearchOperations`. It is the user's responsibility to call the methods.
+**None of these operations are done automatically** by the implementations of `IndexOperations` or `ElasticsearchOperations`.
+It is the user's responsibility to call the methods.
 
 There is support for automatic creation of indices and writing the mappings when using Spring Data Elasticsearch repositories, see <<elasticsearch.repositories.autocreation>>
 
@@ -58,6 +58,7 @@ public class TransportClientConfig extends ElasticsearchConfigurationSupport {
   }
 }
 ----
+
 <1> Setting up the <<elasticsearch.clients.transport>>.
 Deprecated as of version 4.0.
 <2> Creating the `ElasticsearchTemplate` bean, offering both names, _elasticsearchOperations_ and _elasticsearchTemplate_.
@@ -82,6 +83,7 @@ public class RestClientConfig extends AbstractElasticsearchConfiguration {
   // no special bean creation needed                       <2>
 }
 ----
+
 <1> Setting up the <<elasticsearch.clients.rest>>.
 <2> The base class `AbstractElasticsearchConfiguration` already provides the `elasticsearchTemplate` bean.
 ====
@@ -127,6 +129,7 @@ public class TestController {
 }
 
 ----
+
 <1> Let Spring inject the provided `ElasticsearchOperations` bean in the constructor.
 <2> Store some entity in the Elasticsearch cluster.
 <3> Retrieve the entity with a query by id.
@@ -164,6 +167,7 @@ Contains the following information:
 * Maximum score
 * A list of `SearchHit<T>` objects
 * Returned aggregations
+* Returned suggest results
 
 .SearchPage<T>
 Defines a Spring Data `Page` that contains a `SearchHits<T>` element and can be used for paging access using repository methods.
@@ -182,12 +186,12 @@ Almost all of the methods defined in the `SearchOperations` and `ReactiveSearchO
 [[elasticsearch.operations.criteriaquery]]
 === CriteriaQuery
 
-`CriteriaQuery` based queries allow the creation of queries to search for data without knowing the syntax or basics of Elasticsearch queries. They allow the user to build queries by simply chaining and combining `Criteria` objects that specifiy the criteria the searched documents must fulfill.
+`CriteriaQuery` based queries allow the creation of queries to search for data without knowing the syntax or basics of Elasticsearch queries.
+They allow the user to build queries by simply chaining and combining `Criteria` objects that specifiy the criteria the searched documents must fulfill.
 
 NOTE: when talking about AND or OR when combining criteria keep in mind, that in Elasticsearch AND are converted to a **must** condition and OR to a **should**
 
-`Criteria` and their usage are best explained by example
-(let's assume we have a `Book` entity with a `price` property):
+`Criteria` and their usage are best explained by example (let's assume we have a `Book` entity with a `price` property):
 
 .Get books with a given price
 ====
@@ -211,7 +215,7 @@ Query query = new CriteriaQuery(criteria);
 
 When chaining `Criteria`, by default a AND logic is used:
 
-.Get all persons with first name _James_  and last name _Miller_:
+.Get all persons with first name _James_ and last name _Miller_:
 ====
 [source,java]
 ----
@@ -219,11 +223,13 @@ Criteria criteria = new Criteria("lastname").is("Miller") <1>
   .and("firstname").is("James")                           <2>
 Query query = new CriteriaQuery(criteria);
 ----
+
 <1> the first `Criteria`
 <2> the and() creates a new `Criteria` and chaines it to the first one.
 ====
 
-If you want to create nested queries, you need to use subqueries for this. Let's assume we want to find all persons with a last name of _Miller_ and a first name of either _Jack_ or _John_:
+If you want to create nested queries, you need to use subqueries for this.
+Let's assume we want to find all persons with a last name of _Miller_ and a first name of either _Jack_ or _John_:
 
 .Nested subqueries
 ====
@@ -236,6 +242,7 @@ Criteria miller = new Criteria("lastName").is("Miller")  <.>
   );
 Query query = new CriteriaQuery(criteria);
 ----
+
 <.> create a first `Criteria` for the last name
 <.> this is combined with AND to a subCriteria
 <.> This sub Criteria is an OR combination for the first name _John_
@@ -281,5 +288,3 @@ Query query = new NativeSearchQueryBuilder()
 SearchHits<Person> searchHits = operations.search(query, Person.class);
 ----
 ====
-
-

src/main/java/org/springframework/data/elasticsearch/core/AbstractElasticsearchRestTransportTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2021 the original author or authors.
+ * 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.
@@ -32,6 +32,7 @@ import org.elasticsearch.action.search.MultiSearchResponse;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.index.query.MoreLikeThisQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.suggest.SuggestBuilder;
 import org.springframework.data.elasticsearch.BulkFailureException;
 import org.springframework.data.elasticsearch.core.document.SearchDocumentResponse;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
@@ -103,11 +104,12 @@ public abstract class AbstractElasticsearchRestTransportTemplate extends Abstrac
 
 		MultiSearchResponse.Item[] items = getMultiSearchResult(request);
 
+		ReadDocumentCallback<T> documentCallback = new ReadDocumentCallback<T>(elasticsearchConverter, clazz, index);
 		SearchDocumentResponseCallback<SearchHits<T>> callback = new ReadSearchDocumentResponseCallback<>(clazz, index);
 		List<SearchHits<T>> res = new ArrayList<>(queries.size());
 		int c = 0;
 		for (Query query : queries) {
-			res.add(callback.doWith(SearchDocumentResponse.from(items[c++].getResponse())));
+			res.add(callback.doWith(SearchDocumentResponse.from(items[c++].getResponse(), documentCallback::doWith)));
 		}
 		return res;
 	}
@@ -134,11 +136,13 @@ public abstract class AbstractElasticsearchRestTransportTemplate extends Abstrac
 		for (Query query : queries) {
 			Class entityClass = it1.next();
 
+			IndexCoordinates index = getIndexCoordinatesFor(entityClass);
+			ReadDocumentCallback<?> documentCallback = new ReadDocumentCallback<>(elasticsearchConverter, entityClass, index);
 			SearchDocumentResponseCallback<SearchHits<?>> callback = new ReadSearchDocumentResponseCallback<>(entityClass,
-					getIndexCoordinatesFor(entityClass));
+					index);
 
 			SearchResponse response = items[c++].getResponse();
-			res.add(callback.doWith(SearchDocumentResponse.from(response)));
+			res.add(callback.doWith(SearchDocumentResponse.from(response, documentCallback::doWith)));
 		}
 		return res;
 	}
@@ -166,11 +170,12 @@ public abstract class AbstractElasticsearchRestTransportTemplate extends Abstrac
 		for (Query query : queries) {
 			Class entityClass = it1.next();
 
+			ReadDocumentCallback<?> documentCallback = new ReadDocumentCallback<>(elasticsearchConverter, entityClass, index);
 			SearchDocumentResponseCallback<SearchHits<?>> callback = new ReadSearchDocumentResponseCallback<>(entityClass,
 					index);
 
 			SearchResponse response = items[c++].getResponse();
-			res.add(callback.doWith(SearchDocumentResponse.from(response)));
+			res.add(callback.doWith(SearchDocumentResponse.from(response, documentCallback::doWith)));
 		}
 		return res;
 	}
@@ -204,5 +209,11 @@ public abstract class AbstractElasticsearchRestTransportTemplate extends Abstrac
 		return Version.CURRENT.toString();
 	}
 
+	@Override
+	@Deprecated
+	public SearchResponse suggest(SuggestBuilder suggestion, Class<?> clazz) {
+		return suggest(suggestion, getIndexCoordinatesFor(clazz));
+	}
+
 	// endregion
 }

src/main/java/org/springframework/data/elasticsearch/core/AbstractElasticsearchTemplate.java

@@ -22,8 +22,6 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.stream.Collectors;
 
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.search.suggest.SuggestBuilder;
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -432,11 +430,6 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 	 */
 	abstract protected void searchScrollClear(List<String> scrollIds);
 
-	@Override
-	public SearchResponse suggest(SuggestBuilder suggestion, Class<?> clazz) {
-		return suggest(suggestion, getIndexCoordinatesFor(clazz));
-	}
-
 	// endregion
 
 	// region Helper methods
@@ -758,6 +751,7 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 			this.type = type;
 		}
 
+		@NonNull
 		@Override
 		public SearchHits<T> doWith(SearchDocumentResponse response) {
 			List<T> entities = response.getSearchDocuments().stream().map(delegate::doWith).collect(Collectors.toList());
@@ -778,6 +772,7 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 			this.type = type;
 		}
 
+		@NonNull
 		@Override
 		public SearchScrollHits<T> doWith(SearchDocumentResponse response) {
 			List<T> entities = response.getSearchDocuments().stream().map(delegate::doWith).collect(Collectors.toList());

src/main/java/org/springframework/data/elasticsearch/core/ElasticsearchRestTemplate.java

@@ -317,8 +317,10 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchRestTranspor
 		SearchRequest searchRequest = requestFactory.searchRequest(query, clazz, index);
 		SearchResponse response = execute(client -> client.search(searchRequest, RequestOptions.DEFAULT));
 
+		ReadDocumentCallback<T> documentCallback = new ReadDocumentCallback<T>(elasticsearchConverter, clazz, index);
 		SearchDocumentResponseCallback<SearchHits<T>> callback = new ReadSearchDocumentResponseCallback<>(clazz, index);
-		return callback.doWith(SearchDocumentResponse.from(response));
+
+		return callback.doWith(SearchDocumentResponse.from(response, documentCallback::doWith));
 	}
 
 	@Override
@@ -332,9 +334,10 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchRestTranspor
 
 		SearchResponse response = execute(client -> client.search(searchRequest, RequestOptions.DEFAULT));
 
+		ReadDocumentCallback<T> documentCallback = new ReadDocumentCallback<T>(elasticsearchConverter, clazz, index);
 		SearchDocumentResponseCallback<SearchScrollHits<T>> callback = new ReadSearchScrollDocumentResponseCallback<>(clazz,
 				index);
-		return callback.doWith(SearchDocumentResponse.from(response));
+		return callback.doWith(SearchDocumentResponse.from(response, documentCallback::doWith));
 	}
 
 	@Override
@@ -346,9 +349,10 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchRestTranspor
 
 		SearchResponse response = execute(client -> client.scroll(request, RequestOptions.DEFAULT));
 
-		SearchDocumentResponseCallback<SearchScrollHits<T>> callback = //
-				new ReadSearchScrollDocumentResponseCallback<>(clazz, index);
-		return callback.doWith(SearchDocumentResponse.from(response));
+		ReadDocumentCallback<T> documentCallback = new ReadDocumentCallback<T>(elasticsearchConverter, clazz, index);
+		SearchDocumentResponseCallback<SearchScrollHits<T>> callback = new ReadSearchScrollDocumentResponseCallback<>(clazz,
+				index);
+		return callback.doWith(SearchDocumentResponse.from(response, documentCallback::doWith));
 	}
 
 	@Override

src/main/java/org/springframework/data/elasticsearch/core/ElasticsearchTemplate.java

@@ -353,8 +353,9 @@ public class ElasticsearchTemplate extends AbstractElasticsearchRestTransportTem
 		SearchRequestBuilder searchRequestBuilder = requestFactory.searchRequestBuilder(client, query, clazz, index);
 		SearchResponse response = getSearchResponse(searchRequestBuilder);
 
+		ReadDocumentCallback<T> documentCallback = new ReadDocumentCallback<T>(elasticsearchConverter, clazz, index);
 		SearchDocumentResponseCallback<SearchHits<T>> callback = new ReadSearchDocumentResponseCallback<>(clazz, index);
-		return callback.doWith(SearchDocumentResponse.from(response));
+		return callback.doWith(SearchDocumentResponse.from(response, documentCallback::doWith));
 	}
 
 	@Override
@@ -369,9 +370,10 @@ public class ElasticsearchTemplate extends AbstractElasticsearchRestTransportTem
 
 		SearchResponse response = getSearchResponseWithTimeout(action);
 
+		ReadDocumentCallback<T> documentCallback = new ReadDocumentCallback<T>(elasticsearchConverter, clazz, index);
 		SearchDocumentResponseCallback<SearchScrollHits<T>> callback = new ReadSearchScrollDocumentResponseCallback<>(clazz,
 				index);
-		return callback.doWith(SearchDocumentResponse.from(response));
+		return callback.doWith(SearchDocumentResponse.from(response, documentCallback::doWith));
 	}
 
 	@Override
@@ -385,9 +387,10 @@ public class ElasticsearchTemplate extends AbstractElasticsearchRestTransportTem
 
 		SearchResponse response = getSearchResponseWithTimeout(action);
 
+		ReadDocumentCallback<T> documentCallback = new ReadDocumentCallback<T>(elasticsearchConverter, clazz, index);
 		SearchDocumentResponseCallback<SearchScrollHits<T>> callback = new ReadSearchScrollDocumentResponseCallback<>(clazz,
 				index);
-		return callback.doWith(SearchDocumentResponse.from(response));
+		return callback.doWith(SearchDocumentResponse.from(response, documentCallback::doWith));
 	}
 
 	@Override

src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplate.java

@@ -23,6 +23,7 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.elasticsearch.Version;
@@ -44,7 +45,6 @@ import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
 import org.elasticsearch.index.reindex.DeleteByQueryRequest;
 import org.elasticsearch.index.reindex.UpdateByQueryRequest;
-import org.elasticsearch.search.suggest.Suggest;
 import org.elasticsearch.search.suggest.SuggestBuilder;
 import org.reactivestreams.Publisher;
 import org.slf4j.Logger;
@@ -84,6 +84,7 @@ import org.springframework.data.elasticsearch.core.query.UpdateQuery;
 import org.springframework.data.elasticsearch.core.query.UpdateResponse;
 import org.springframework.data.elasticsearch.core.routing.DefaultRoutingResolver;
 import org.springframework.data.elasticsearch.core.routing.RoutingResolver;
+import org.springframework.data.elasticsearch.core.suggest.response.Suggest;
 import org.springframework.data.elasticsearch.support.VersionInfo;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.mapping.callback.ReactiveEntityCallbacks;
@@ -375,8 +376,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 	protected Flux<BulkItemResponse> doBulkOperation(List<?> queries, BulkOptions bulkOptions, IndexCoordinates index) {
 		BulkRequest bulkRequest = prepareWriteRequest(requestFactory.bulkRequest(queries, bulkOptions, index));
 		return client.bulk(bulkRequest) //
-				.onErrorMap(
-						e -> new UncategorizedElasticsearchException("Error while bulk for request: " + bulkRequest.toString(), e)) //
+				.onErrorMap(e -> new UncategorizedElasticsearchException("Error while bulk for request: " + bulkRequest, e)) //
 				.flatMap(this::checkForBulkOperationFailure) //
 				.flatMapMany(response -> Flux.fromArray(response.getItems()));
 	}
@@ -658,7 +658,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 	}
 
 	/**
-	 * Customization hook to modify a generated {@link DeleteRequest} prior to its execution. Eg. by setting the
+	 * Customization hook to modify a generated {@link DeleteRequest} prior to its execution. E.g. by setting the
 	 * {@link WriteRequest#setRefreshPolicy(String) refresh policy} if applicable.
 	 *
 	 * @param request the generated {@link DeleteRequest}.
@@ -669,7 +669,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 	}
 
 	/**
-	 * Customization hook to modify a generated {@link DeleteByQueryRequest} prior to its execution. Eg. by setting the
+	 * Customization hook to modify a generated {@link DeleteByQueryRequest} prior to its execution. E.g. by setting the
 	 * {@link WriteRequest#setRefreshPolicy(String) refresh policy} if applicable.
 	 *
 	 * @param request the generated {@link DeleteByQueryRequest}.
@@ -694,7 +694,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 	}
 
 	/**
-	 * Customization hook to modify a generated {@link IndexRequest} prior to its execution. Eg. by setting the
+	 * Customization hook to modify a generated {@link IndexRequest} prior to its execution. E.g. by setting the
 	 * {@link WriteRequest#setRefreshPolicy(String) refresh policy} if applicable.
 	 *
 	 * @param source the source object the {@link IndexRequest} was derived from.
@@ -706,7 +706,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 	}
 
 	/**
-	 * Pre process the write request before it is sent to the server, eg. by setting the
+	 * Preprocess the write request before it is sent to the server, e.g. by setting the
 	 * {@link WriteRequest#setRefreshPolicy(String) refresh policy} if applicable.
 	 *
 	 * @param request must not be {@literal null}.
@@ -777,7 +777,10 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 		return Mono.defer(() -> {
 			SearchRequest request = requestFactory.searchRequest(query, clazz, index);
 			request = prepareSearchRequest(request, false);
-			return doFindForResponse(request);
+
+			SearchDocumentCallback<?> documentCallback = new ReadSearchDocumentCallback<>(clazz, index);
+
+			return doFindForResponse(request, searchDocument -> documentCallback.toEntity(searchDocument).block());
 		});
 	}
 
@@ -788,27 +791,11 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 
 	@Override
 	public Flux<AggregationContainer<?>> aggregate(Query query, Class<?> entityType, IndexCoordinates index) {
-		return doAggregate(query, entityType, index);
-	}
 
-	@Override
-	public Flux<Suggest> suggest(SuggestBuilder suggestion, Class<?> entityType) {
-		return doSuggest(suggestion, getIndexCoordinatesFor(entityType));
-	}
+		Assert.notNull(query, "query must not be null");
+		Assert.notNull(entityType, "entityType must not be null");
+		Assert.notNull(index, "index must not be null");
 
-	@Override
-	public Flux<Suggest> suggest(SuggestBuilder suggestion, IndexCoordinates index) {
-		return doSuggest(suggestion, index);
-	}
-
-	private Flux<Suggest> doSuggest(SuggestBuilder suggestion, IndexCoordinates index) {
-		return Flux.defer(() -> {
-			SearchRequest request = requestFactory.searchRequest(suggestion, index);
-			return Flux.from(execute(client -> client.suggest(request)));
-		});
-	}
-
-	private Flux<AggregationContainer<?>> doAggregate(Query query, Class<?> entityType, IndexCoordinates index) {
 		return Flux.defer(() -> {
 			SearchRequest request = requestFactory.searchRequest(query, entityType, index);
 			request = prepareSearchRequest(request, false);
@@ -816,6 +803,61 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 		});
 	}
 
+	/**
+	 * Customization hook on the actual execution result {@link Publisher}. <br />
+	 *
+	 * @param request the already prepared {@link SearchRequest} ready to be executed.
+	 * @return a {@link Flux} emitting the result of the operation.
+	 */
+	protected Flux<AggregationContainer<?>> doAggregate(SearchRequest request) {
+
+		if (QUERY_LOGGER.isDebugEnabled()) {
+			QUERY_LOGGER.debug("Executing doCount: {}", request);
+		}
+
+		return Flux.from(execute(client -> client.aggregate(request))) //
+				.onErrorResume(NoSuchIndexException.class, it -> Flux.empty()).map(ElasticsearchAggregation::new);
+	}
+
+	@Override
+	public Mono<Suggest> suggest(Query query, Class<?> entityType) {
+		return suggest(query, entityType, getIndexCoordinatesFor(entityType));
+	}
+
+	@Override
+	public Mono<Suggest> suggest(Query query, Class<?> entityType, IndexCoordinates index) {
+
+		Assert.notNull(query, "query must not be null");
+		Assert.notNull(entityType, "entityType must not be null");
+		Assert.notNull(index, "index must not be null");
+
+		return doFindForResponse(query, entityType, index).mapNotNull(searchDocumentResponse -> {
+			Suggest suggest = searchDocumentResponse.getSuggest();
+			SearchHitMapping.mappingFor(entityType, converter).mapHitsInCompletionSuggestion(suggest);
+			return suggest;
+		});
+	}
+
+	@Override
+	@Deprecated
+	public Flux<org.elasticsearch.search.suggest.Suggest> suggest(SuggestBuilder suggestion, Class<?> entityType) {
+		return doSuggest(suggestion, getIndexCoordinatesFor(entityType));
+	}
+
+	@Override
+	@Deprecated
+	public Flux<org.elasticsearch.search.suggest.Suggest> suggest(SuggestBuilder suggestion, IndexCoordinates index) {
+		return doSuggest(suggestion, index);
+	}
+
+	@Deprecated
+	private Flux<org.elasticsearch.search.suggest.Suggest> doSuggest(SuggestBuilder suggestion, IndexCoordinates index) {
+		return Flux.defer(() -> {
+			SearchRequest request = requestFactory.searchRequest(suggestion, index);
+			return Flux.from(execute(client -> client.suggest(request)));
+		});
+	}
+
 	@Override
 	public Mono<Long> count(Query query, Class<?> entityType) {
 		return count(query, entityType, getIndexCoordinatesFor(entityType));
@@ -855,31 +897,19 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 	 * Customization hook on the actual execution result {@link Mono}. <br />
 	 *
 	 * @param request the already prepared {@link SearchRequest} ready to be executed.
+	 * @param suggestEntityCreator
 	 * @return a {@link Mono} emitting the result of the operation converted to s {@link SearchDocumentResponse}.
 	 */
-	protected Mono<SearchDocumentResponse> doFindForResponse(SearchRequest request) {
+	protected Mono<SearchDocumentResponse> doFindForResponse(SearchRequest request,
+			Function<SearchDocument, ? extends Object> suggestEntityCreator) {
 
 		if (QUERY_LOGGER.isDebugEnabled()) {
 			QUERY_LOGGER.debug("Executing doFindForResponse: {}", request);
 		}
 
-		return Mono.from(execute(client1 -> client1.searchForResponse(request))).map(SearchDocumentResponse::from);
-	}
-
-	/**
-	 * Customization hook on the actual execution result {@link Publisher}. <br />
-	 *
-	 * @param request the already prepared {@link SearchRequest} ready to be executed.
-	 * @return a {@link Flux} emitting the result of the operation.
-	 */
-	protected Flux<AggregationContainer<?>> doAggregate(SearchRequest request) {
-
-		if (QUERY_LOGGER.isDebugEnabled()) {
-			QUERY_LOGGER.debug("Executing doCount: {}", request);
-		}
-
-		return Flux.from(execute(client -> client.aggregate(request))) //
-				.onErrorResume(NoSuchIndexException.class, it -> Flux.empty()).map(ElasticsearchAggregation::new);
+		return Mono.from(execute(client1 -> client1.searchForResponse(request))).map(searchResponse -> {
+			return SearchDocumentResponse.from(searchResponse, suggestEntityCreator);
+		});
 	}
 
 	/**
@@ -915,7 +945,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 	}
 
 	/**
-	 * Customization hook to modify a generated {@link SearchRequest} prior to its execution. Eg. by setting the
+	 * Customization hook to modify a generated {@link SearchRequest} prior to its execution. E.g. by setting the
 	 * {@link SearchRequest#indicesOptions(IndicesOptions) indices options} if applicable.
 	 *
 	 * @param request the generated {@link SearchRequest}.
@@ -941,7 +971,8 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 	// region Helper methods
 	protected Mono<String> getClusterVersion() {
 		try {
-			return Mono.from(execute(client -> client.info())).map(mainResponse -> mainResponse.getVersion().toString());
+			return Mono.from(execute(ReactiveElasticsearchClient::info))
+					.map(mainResponse -> mainResponse.getVersion().toString());
 		} catch (Exception ignored) {}
 		return Mono.empty();
 	}
@@ -1163,11 +1194,9 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 
 	protected interface SearchDocumentCallback<T> {
 
-		@NonNull
-		Mono<T> toEntity(@NonNull SearchDocument response);
+		Mono<T> toEntity(SearchDocument response);
 
-		@NonNull
-		Mono<SearchHit<T>> toSearchHit(@NonNull SearchDocument response);
+		Mono<SearchHit<T>> toSearchHit(SearchDocument response);
 	}
 
 	protected class ReadSearchDocumentCallback<T> implements SearchDocumentCallback<T> {
@@ -1210,7 +1239,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 		}
 
 		private T entityAt(long index) {
-			// it's safe to cast to int because the original indexed colleciton was fitting in memory
+			// it's safe to cast to int because the original indexed collection was fitting in memory
 			int intIndex = (int) index;
 			return entities.get(intIndex);
 		}

src/main/java/org/springframework/data/elasticsearch/core/ReactiveSearchOperations.java

@@ -20,11 +20,11 @@ import reactor.core.publisher.Mono;
 
 import java.util.List;
 
-import org.elasticsearch.search.suggest.Suggest;
 import org.elasticsearch.search.suggest.SuggestBuilder;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.elasticsearch.core.query.Query;
+import org.springframework.data.elasticsearch.core.suggest.response.Suggest;
 
 /**
  * The reactive operations for the
@@ -202,18 +202,47 @@ public interface ReactiveSearchOperations {
 	 *
 	 * @param suggestion the query
 	 * @param entityType must not be {@literal null}.
-	 * @return the suggest response
+	 * @return the suggest response (Elasticsearch library classes)
+	 * @deprecated since 4.3, use {@link #suggest(Query, Class)}
 	 */
-	Flux<Suggest> suggest(SuggestBuilder suggestion, Class<?> entityType);
+	@Deprecated
+	Flux<org.elasticsearch.search.suggest.Suggest> suggest(SuggestBuilder suggestion, Class<?> entityType);
 
 	/**
 	 * Does a suggest query
 	 *
 	 * @param suggestion the query
 	 * @param index the index to run the query against
-	 * @return the suggest response
+	 * @return the suggest response (Elasticsearch library classes)
+	 * @deprecated since 4.3, use {@link #suggest(Query, Class, IndexCoordinates)}
 	 */
-	Flux<Suggest> suggest(SuggestBuilder suggestion, IndexCoordinates index);
+	@Deprecated
+	Flux<org.elasticsearch.search.suggest.Suggest> suggest(SuggestBuilder suggestion, IndexCoordinates index);
+
+	/**
+	 * Does a suggest query.
+	 *
+	 * @param query the Query containing the suggest definition. Must be currently a
+	 *          {@link org.springframework.data.elasticsearch.core.query.NativeSearchQuery}, must not be {@literal null}.
+	 * @param entityType the type of the entities that might be returned for a completion suggestion, must not be
+	 *          {@literal null}.
+	 * @return suggest data
+	 * @since 4.3
+	 */
+	Mono<Suggest> suggest(Query query, Class<?> entityType);
+
+	/**
+	 * Does a suggest query.
+	 *
+	 * @param query the Query containing the suggest definition. Must be currently a
+	 *          {@link org.springframework.data.elasticsearch.core.query.NativeSearchQuery}, must not be {@literal null}.
+	 * @param entityType the type of the entities that might be returned for a completion suggestion, must not be
+	 *          {@literal null}.
+	 * @param index the index to run the query against, must not be {@literal null}.
+	 * @return suggest data
+	 * @since 4.3
+	 */
+	Mono<Suggest> suggest(Query query, Class<?> entityType, IndexCoordinates index);
 
 	// region helper
 	/**

src/main/java/org/springframework/data/elasticsearch/core/RequestFactory.java

@@ -817,7 +817,8 @@ class RequestFactory {
 			if (query instanceof NativeSearchQuery) {
 				NativeSearchQuery searchQuery = (NativeSearchQuery) query;
 
-				if (searchQuery.getHighlightFields() != null || searchQuery.getHighlightBuilder() != null) {
+				if ((searchQuery.getHighlightFields() != null && searchQuery.getHighlightFields().length > 0)
+						|| searchQuery.getHighlightBuilder() != null) {
 					highlightBuilder = searchQuery.getHighlightBuilder();
 
 					if (highlightBuilder == null) {
@@ -1140,6 +1141,9 @@ class RequestFactory {
 			query.getPipelineAggregations().forEach(sourceBuilder::aggregation);
 		}
 
+		if (query.getSuggestBuilder() != null) {
+			sourceBuilder.suggest(query.getSuggestBuilder());
+		}
 	}
 
 	private void prepareNativeSearch(SearchRequestBuilder searchRequestBuilder, NativeSearchQuery nativeSearchQuery) {
@@ -1166,6 +1170,10 @@ class RequestFactory {
 		if (!isEmpty(nativeSearchQuery.getPipelineAggregations())) {
 			nativeSearchQuery.getPipelineAggregations().forEach(searchRequestBuilder::addAggregation);
 		}
+
+		if (nativeSearchQuery.getSuggestBuilder() != null) {
+			searchRequestBuilder.suggest(nativeSearchQuery.getSuggestBuilder());
+		}
 	}
 
 	@SuppressWarnings("rawtypes")

src/main/java/org/springframework/data/elasticsearch/core/SearchHitMapping.java

@@ -31,6 +31,8 @@ import org.springframework.data.elasticsearch.core.document.SearchDocument;
 import org.springframework.data.elasticsearch.core.document.SearchDocumentResponse;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentEntity;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentProperty;
+import org.springframework.data.elasticsearch.core.suggest.response.CompletionSuggestion;
+import org.springframework.data.elasticsearch.core.suggest.response.Suggest;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -97,7 +99,27 @@ class SearchHitMapping<T> {
 		AggregationsContainer<?> aggregations = searchDocumentResponse.getAggregations();
 		TotalHitsRelation totalHitsRelation = TotalHitsRelation.valueOf(searchDocumentResponse.getTotalHitsRelation());
 
-		return new SearchHitsImpl<>(totalHits, totalHitsRelation, maxScore, scrollId, searchHits, aggregations);
+		Suggest suggest = searchDocumentResponse.getSuggest();
+		mapHitsInCompletionSuggestion(suggest);
+
+		return new SearchHitsImpl<>(totalHits, totalHitsRelation, maxScore, scrollId, searchHits, aggregations, suggest);
+	}
+
+	@SuppressWarnings("unchecked")
+	public void mapHitsInCompletionSuggestion(@Nullable Suggest suggest) {
+		if (suggest != null) {
+			for (Suggest.Suggestion<? extends Suggest.Suggestion.Entry<? extends Suggest.Suggestion.Entry.Option>> suggestion : suggest
+					.getSuggestions()) {
+				if (suggestion instanceof CompletionSuggestion) {
+					CompletionSuggestion<T> completionSuggestion = (CompletionSuggestion<T>) suggestion;
+					for (CompletionSuggestion.Entry<T> entry : completionSuggestion.getEntries()) {
+						for (CompletionSuggestion.Entry.Option<T> option : entry.getOptions()) {
+							option.updateSearchHit(this::mapHit);
+						}
+					}
+				}
+			}
+		}
 	}
 
 	SearchHit<T> mapHit(SearchDocument searchDocument, T content) {
@@ -213,7 +235,8 @@ class SearchHitMapping<T> {
 						searchHits.getMaxScore(), //
 						scrollId, //
 						convertedSearchHits, //
-						searchHits.getAggregations());
+						searchHits.getAggregations(), //
+						searchHits.getSuggest());
 			}
 		} catch (Exception e) {
 			LOGGER.warn("Could not map inner_hits", e);

src/main/java/org/springframework/data/elasticsearch/core/SearchHits.java

@@ -18,6 +18,7 @@ package org.springframework.data.elasticsearch.core;
 import java.util.Iterator;
 import java.util.List;
 
+import org.springframework.data.elasticsearch.core.suggest.response.Suggest;
 import org.springframework.data.util.Streamable;
 import org.springframework.lang.Nullable;
 
@@ -77,6 +78,21 @@ public interface SearchHits<T> extends Streamable<SearchHit<T>> {
 		return !getSearchHits().isEmpty();
 	}
 
+	/**
+	 * @return the suggest response
+	 * @since 4.3
+	 */
+	@Nullable
+	Suggest getSuggest();
+
+	/**
+	 * @return wether the {@link SearchHits} has a suggest response.
+	 * @since 4.3
+	 */
+	default boolean hasSuggest() {
+		return getSuggest() != null;
+	}
+
 	/**
 	 * @return an iterator for {@link SearchHit}
 	 */

src/main/java/org/springframework/data/elasticsearch/core/SearchHitsImpl.java

@@ -18,6 +18,7 @@ package org.springframework.data.elasticsearch.core;
 import java.util.Collections;
 import java.util.List;
 
+import org.springframework.data.elasticsearch.core.suggest.response.Suggest;
 import org.springframework.data.util.Lazy;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -39,6 +40,7 @@ public class SearchHitsImpl<T> implements SearchScrollHits<T> {
 	private final List<? extends SearchHit<T>> searchHits;
 	private final Lazy<List<SearchHit<T>>> unmodifiableSearchHits;
 	@Nullable private final AggregationsContainer<?> aggregations;
+	@Nullable private final Suggest suggest;
 
 	/**
 	 * @param totalHits the number of total hits for the search
@@ -49,7 +51,8 @@ public class SearchHitsImpl<T> implements SearchScrollHits<T> {
 	 * @param aggregations the aggregations if available
 	 */
 	public SearchHitsImpl(long totalHits, TotalHitsRelation totalHitsRelation, float maxScore, @Nullable String scrollId,
-			List<? extends SearchHit<T>> searchHits, @Nullable AggregationsContainer<?> aggregations) {
+			List<? extends SearchHit<T>> searchHits, @Nullable AggregationsContainer<?> aggregations,
+			@Nullable Suggest suggest) {
 
 		Assert.notNull(searchHits, "searchHits must not be null");
 
@@ -59,6 +62,7 @@ public class SearchHitsImpl<T> implements SearchScrollHits<T> {
 		this.scrollId = scrollId;
 		this.searchHits = searchHits;
 		this.aggregations = aggregations;
+		this.suggest = suggest;
 		this.unmodifiableSearchHits = Lazy.of(() -> Collections.unmodifiableList(searchHits));
 	}
 
@@ -88,14 +92,23 @@ public class SearchHitsImpl<T> implements SearchScrollHits<T> {
 	public List<SearchHit<T>> getSearchHits() {
 		return unmodifiableSearchHits.get();
 	}
-	// endregion
 
-	// region SearchHit access
 	@Override
 	public SearchHit<T> getSearchHit(int index) {
 		return searchHits.get(index);
 	}
-	// endregion
+
+	@Override
+	@Nullable
+	public AggregationsContainer<?> getAggregations() {
+		return aggregations;
+	}
+
+	@Override
+	@Nullable
+	public Suggest getSuggest() {
+		return suggest;
+	}
 
 	@Override
 	public String toString() {
@@ -108,12 +121,4 @@ public class SearchHitsImpl<T> implements SearchScrollHits<T> {
 				", aggregations=" + aggregations + //
 				'}';
 	}
-
-	// region aggregations
-	@Override
-	@Nullable
-	public AggregationsContainer<?> getAggregations() {
-		return aggregations;
-	}
-	// endregion
 }

src/main/java/org/springframework/data/elasticsearch/core/SearchOperations.java

@@ -71,7 +71,11 @@ public interface SearchOperations {
 	 * @param clazz the entity class
 	 * @return the suggest response
 	 * @since 4.1
+	 * @deprecated since 4.3 use a {@link org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder} with
+	 *             {@link org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder#withSuggestBuilder(SuggestBuilder)},
+	 *             call {@link #search(Query, Class)} and get the suggest from {@link SearchHits#getSuggest()}
 	 */
+	@Deprecated
 	SearchResponse suggest(SuggestBuilder suggestion, Class<?> clazz);
 
 	/**
@@ -80,7 +84,11 @@ public interface SearchOperations {
 	 * @param suggestion the query
 	 * @param index the index to run the query against
 	 * @return the suggest response
+	 * @deprecated since 4.3 use a {@link org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder} with
+	 *             {@link org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder#withSuggestBuilder(SuggestBuilder)},
+	 *             call {@link #search(Query, Class)} and get the suggest from {@link SearchHits#getSuggest()}
 	 */
+	@Deprecated
 	SearchResponse suggest(SuggestBuilder suggestion, IndexCoordinates index);
 
 	/**

src/main/java/org/springframework/data/elasticsearch/core/document/DocumentAdapters.java

@@ -177,8 +177,8 @@ public final class DocumentAdapters {
 		Map<String, SearchHits> sourceInnerHits = source.getInnerHits();
 
 		if (sourceInnerHits != null) {
-			sourceInnerHits
-					.forEach((name, searchHits) -> innerHits.put(name, SearchDocumentResponse.from(searchHits, null, null)));
+			sourceInnerHits.forEach((name, searchHits) -> innerHits.put(name,
+					SearchDocumentResponse.from(searchHits, null, null, null, searchDocument -> null)));
 		}
 
 		NestedMetaData nestedMetaData = from(source.getNestedIdentity());
@@ -186,12 +186,13 @@ public final class DocumentAdapters {
 		List<String> matchedQueries = from(source.getMatchedQueries());
 
 		BytesReference sourceRef = source.getSourceRef();
+		Map<String, DocumentField> sourceFields = source.getFields();
 
 		if (sourceRef == null || sourceRef.length() == 0) {
 			return new SearchDocumentAdapter(
 					fromDocumentFields(source, source.getIndex(), source.getId(), source.getVersion(), source.getSeqNo(),
 							source.getPrimaryTerm()),
-					source.getScore(), source.getSortValues(), source.getFields(), highlightFields, innerHits, nestedMetaData,
+					source.getScore(), source.getSortValues(), sourceFields, highlightFields, innerHits, nestedMetaData,
 					explanation, matchedQueries);
 		}
 
@@ -205,8 +206,8 @@ public final class DocumentAdapters {
 		document.setSeqNo(source.getSeqNo());
 		document.setPrimaryTerm(source.getPrimaryTerm());
 
-		return new SearchDocumentAdapter(document, source.getScore(), source.getSortValues(), source.getFields(),
-				highlightFields, innerHits, nestedMetaData, explanation, matchedQueries);
+		return new SearchDocumentAdapter(document, source.getScore(), source.getSortValues(), sourceFields, highlightFields,
+				innerHits, nestedMetaData, explanation, matchedQueries);
 	}
 
 	@Nullable

src/main/java/org/springframework/data/elasticsearch/core/document/SearchDocumentResponse.java

@@ -17,20 +17,28 @@ package org.springframework.data.elasticsearch.core.document;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Function;
 
 import org.apache.lucene.search.TotalHits;
 import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.common.text.Text;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.SearchHits;
 import org.elasticsearch.search.aggregations.Aggregations;
 import org.springframework.data.elasticsearch.core.AggregationsContainer;
 import org.springframework.data.elasticsearch.core.clients.elasticsearch7.ElasticsearchAggregations;
+import org.springframework.data.elasticsearch.core.suggest.response.CompletionSuggestion;
+import org.springframework.data.elasticsearch.core.suggest.response.PhraseSuggestion;
+import org.springframework.data.elasticsearch.core.suggest.response.SortBy;
+import org.springframework.data.elasticsearch.core.suggest.response.Suggest;
+import org.springframework.data.elasticsearch.core.suggest.response.TermSuggestion;
+import org.springframework.data.elasticsearch.support.ScoreDoc;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
  * This represents the complete search response from Elasticsearch, including the returned documents. Instances must be
- * created with the {@link #from(SearchResponse)} method.
+ * created with the {@link #from(SearchResponse,Function)} method.
  *
  * @author Peter-Josef Meisch
  * @since 4.0
@@ -42,16 +50,18 @@ public class SearchDocumentResponse {
 	private final float maxScore;
 	private final String scrollId;
 	private final List<SearchDocument> searchDocuments;
-	private final AggregationsContainer<?> aggregations;
+	@Nullable private final AggregationsContainer<?> aggregations;
+	@Nullable private final Suggest suggest;
 
 	private SearchDocumentResponse(long totalHits, String totalHitsRelation, float maxScore, String scrollId,
-			List<SearchDocument> searchDocuments, Aggregations aggregations) {
+			List<SearchDocument> searchDocuments, @Nullable Aggregations aggregations, @Nullable Suggest suggest) {
 		this.totalHits = totalHits;
 		this.totalHitsRelation = totalHitsRelation;
 		this.maxScore = maxScore;
 		this.scrollId = scrollId;
 		this.searchDocuments = searchDocuments;
-		this.aggregations = new ElasticsearchAggregations(aggregations);
+		this.aggregations = aggregations != null ? new ElasticsearchAggregations(aggregations) : null;
+		this.suggest = suggest;
 	}
 
 	public long getTotalHits() {
@@ -74,40 +84,53 @@ public class SearchDocumentResponse {
 		return searchDocuments;
 	}
 
+	@Nullable
 	public AggregationsContainer<?> getAggregations() {
 		return aggregations;
 	}
 
+	@Nullable
+	public Suggest getSuggest() {
+		return suggest;
+	}
+
 	/**
 	 * creates a SearchDocumentResponse from the {@link SearchResponse}
 	 *
 	 * @param searchResponse must not be {@literal null}
+	 * @param suggestEntityCreator function to create an entity from a {@link SearchDocument}
+	 * @param <T> entity type
 	 * @return the SearchDocumentResponse
 	 */
-	public static SearchDocumentResponse from(SearchResponse searchResponse) {
+	public static <T> SearchDocumentResponse from(SearchResponse searchResponse,
+			Function<SearchDocument, T> suggestEntityCreator) {
 
 		Assert.notNull(searchResponse, "searchResponse must not be null");
 
-		Aggregations aggregations = searchResponse.getAggregations();
-		String scrollId = searchResponse.getScrollId();
-
 		SearchHits searchHits = searchResponse.getHits();
+		String scrollId = searchResponse.getScrollId();
+		Aggregations aggregations = searchResponse.getAggregations();
+		org.elasticsearch.search.suggest.Suggest suggest = searchResponse.getSuggest();
 
-		SearchDocumentResponse searchDocumentResponse = from(searchHits, scrollId, aggregations);
-		return searchDocumentResponse;
+		return from(searchHits, scrollId, aggregations, suggest, suggestEntityCreator);
 	}
 
 	/**
-	 * creates a {@link SearchDocumentResponse} from {@link SearchHits} with the given scrollId and aggregations
+	 * creates a {@link SearchDocumentResponse} from {@link SearchHits} with the given scrollId aggregations and suggest
 	 *
 	 * @param searchHits the {@link SearchHits} to process
 	 * @param scrollId scrollId
 	 * @param aggregations aggregations
+	 * @param suggestES the suggestion response from Elasticsearch
+	 * @param suggestEntityCreator function to create an entity from a {@link SearchDocument}
+	 * @param <T> entity type
 	 * @return the {@link SearchDocumentResponse}
-	 * @since 4.1
+	 * @since 4.3
 	 */
-	public static SearchDocumentResponse from(SearchHits searchHits, @Nullable String scrollId,
-			@Nullable Aggregations aggregations) {
+	public static <T> SearchDocumentResponse from(SearchHits searchHits, @Nullable String scrollId,
+			@Nullable Aggregations aggregations, @Nullable org.elasticsearch.search.suggest.Suggest suggestES,
+			Function<SearchDocument, T> suggestEntityCreator) {
+
 		TotalHits responseTotalHits = searchHits.getTotalHits();
 
 		long totalHits;
@@ -130,7 +153,105 @@ public class SearchDocumentResponse {
 			}
 		}
 
-		return new SearchDocumentResponse(totalHits, totalHitsRelation, maxScore, scrollId, searchDocuments, aggregations);
+		Suggest suggest = suggestFrom(suggestES, suggestEntityCreator);
+		return new SearchDocumentResponse(totalHits, totalHitsRelation, maxScore, scrollId, searchDocuments, aggregations,
+				suggest);
 	}
 
+	@Nullable
+	private static <T> Suggest suggestFrom(@Nullable org.elasticsearch.search.suggest.Suggest suggestES,
+			Function<SearchDocument, T> entityCreator) {
+
+		if (suggestES == null) {
+			return null;
+		}
+
+		List<Suggest.Suggestion<? extends Suggest.Suggestion.Entry<? extends Suggest.Suggestion.Entry.Option>>> suggestions = new ArrayList<>();
+
+		for (org.elasticsearch.search.suggest.Suggest.Suggestion<? extends org.elasticsearch.search.suggest.Suggest.Suggestion.Entry<? extends org.elasticsearch.search.suggest.Suggest.Suggestion.Entry.Option>> suggestionES : suggestES) {
+
+			if (suggestionES instanceof org.elasticsearch.search.suggest.term.TermSuggestion) {
+				org.elasticsearch.search.suggest.term.TermSuggestion termSuggestionES = (org.elasticsearch.search.suggest.term.TermSuggestion) suggestionES;
+
+				List<TermSuggestion.Entry> entries = new ArrayList<>();
+				for (org.elasticsearch.search.suggest.term.TermSuggestion.Entry entryES : termSuggestionES) {
+
+					List<TermSuggestion.Entry.Option> options = new ArrayList<>();
+					for (org.elasticsearch.search.suggest.term.TermSuggestion.Entry.Option optionES : entryES) {
+						options.add(new TermSuggestion.Entry.Option(textToString(optionES.getText()),
+								textToString(optionES.getHighlighted()), optionES.getScore(), optionES.collateMatch(),
+								optionES.getFreq()));
+					}
+
+					entries.add(new TermSuggestion.Entry(textToString(entryES.getText()), entryES.getOffset(),
+							entryES.getLength(), options));
+				}
+
+				suggestions.add(new TermSuggestion(termSuggestionES.getName(), termSuggestionES.getSize(), entries,
+						suggestFrom(termSuggestionES.getSort())));
+			}
+
+			if (suggestionES instanceof org.elasticsearch.search.suggest.phrase.PhraseSuggestion) {
+				org.elasticsearch.search.suggest.phrase.PhraseSuggestion phraseSuggestionES = (org.elasticsearch.search.suggest.phrase.PhraseSuggestion) suggestionES;
+
+				List<PhraseSuggestion.Entry> entries = new ArrayList<>();
+				for (org.elasticsearch.search.suggest.phrase.PhraseSuggestion.Entry entryES : phraseSuggestionES) {
+
+					List<PhraseSuggestion.Entry.Option> options = new ArrayList<>();
+					for (org.elasticsearch.search.suggest.phrase.PhraseSuggestion.Entry.Option optionES : entryES) {
+						options.add(new PhraseSuggestion.Entry.Option(textToString(optionES.getText()),
+								textToString(optionES.getHighlighted()), optionES.getScore(), optionES.collateMatch()));
+					}
+
+					entries.add(new PhraseSuggestion.Entry(textToString(entryES.getText()), entryES.getOffset(),
+							entryES.getLength(), options, entryES.getCutoffScore()));
+				}
+
+				suggestions.add(new PhraseSuggestion(phraseSuggestionES.getName(), phraseSuggestionES.getSize(), entries));
+			}
+
+			if (suggestionES instanceof org.elasticsearch.search.suggest.completion.CompletionSuggestion) {
+				org.elasticsearch.search.suggest.completion.CompletionSuggestion completionSuggestionES = (org.elasticsearch.search.suggest.completion.CompletionSuggestion) suggestionES;
+
+				List<CompletionSuggestion.Entry<T>> entries = new ArrayList<>();
+				for (org.elasticsearch.search.suggest.completion.CompletionSuggestion.Entry entryES : completionSuggestionES) {
+
+					List<CompletionSuggestion.Entry.Option<T>> options = new ArrayList<>();
+					for (org.elasticsearch.search.suggest.completion.CompletionSuggestion.Entry.Option optionES : entryES) {
+						SearchDocument searchDocument = optionES.getHit() != null ? DocumentAdapters.from(optionES.getHit()) : null;
+						T hitEntity = searchDocument != null ? entityCreator.apply(searchDocument) : null;
+						options.add(new CompletionSuggestion.Entry.Option<T>(textToString(optionES.getText()),
+								textToString(optionES.getHighlighted()), optionES.getScore(), optionES.collateMatch(),
+								optionES.getContexts(), scoreDocFrom(optionES.getDoc()), searchDocument, hitEntity));
+					}
+
+					entries.add(new CompletionSuggestion.Entry<T>(textToString(entryES.getText()), entryES.getOffset(),
+							entryES.getLength(), options));
+				}
+
+				suggestions.add(
+						new CompletionSuggestion<T>(completionSuggestionES.getName(), completionSuggestionES.getSize(), entries));
+			}
+		}
+
+		return new Suggest(suggestions, suggestES.hasScoreDocs());
+	}
+
+	private static SortBy suggestFrom(org.elasticsearch.search.suggest.SortBy sort) {
+		return SortBy.valueOf(sort.name().toUpperCase());
+	}
+
+	@Nullable
+	private static ScoreDoc scoreDocFrom(@Nullable org.apache.lucene.search.ScoreDoc scoreDoc) {
+
+		if (scoreDoc == null) {
+			return null;
+		}
+
+		return new ScoreDoc(scoreDoc.score, scoreDoc.doc, scoreDoc.shardIndex);
+	}
+
+	private static String textToString(@Nullable Text text) {
+		return text != null ? text.string() : "";
+	}
 }

src/main/java/org/springframework/data/elasticsearch/core/mapping/SimpleElasticsearchPersistentProperty.java

@@ -30,7 +30,7 @@ import org.springframework.data.elasticsearch.annotations.GeoPointField;
 import org.springframework.data.elasticsearch.annotations.GeoShapeField;
 import org.springframework.data.elasticsearch.annotations.MultiField;
 import org.springframework.data.elasticsearch.core.Range;
-import org.springframework.data.elasticsearch.core.completion.Completion;
+import org.springframework.data.elasticsearch.core.suggest.Completion;
 import org.springframework.data.elasticsearch.core.convert.DatePersistentPropertyConverter;
 import org.springframework.data.elasticsearch.core.convert.DateRangePersistentPropertyConverter;
 import org.springframework.data.elasticsearch.core.convert.ElasticsearchDateConverter;

src/main/java/org/springframework/data/elasticsearch/core/query/NativeSearchQuery.java

@@ -26,6 +26,7 @@ import org.elasticsearch.search.aggregations.PipelineAggregationBuilder;
 import org.elasticsearch.search.collapse.CollapseBuilder;
 import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
 import org.elasticsearch.search.sort.SortBuilder;
+import org.elasticsearch.search.suggest.SuggestBuilder;
 import org.springframework.lang.Nullable;
 
 /**
@@ -54,6 +55,7 @@ public class NativeSearchQuery extends AbstractQuery {
 	@Nullable private HighlightBuilder.Field[] highlightFields;
 	@Nullable private List<IndexBoost> indicesBoost;
 	@Nullable private SearchTemplateRequestBuilder searchTemplate;
+	@Nullable private SuggestBuilder suggestBuilder;
 
 	public NativeSearchQuery(@Nullable QueryBuilder query) {
 
@@ -184,4 +186,19 @@ public class NativeSearchQuery extends AbstractQuery {
 	public void setSearchTemplate(@Nullable SearchTemplateRequestBuilder searchTemplate) {
 		this.searchTemplate = searchTemplate;
 	}
+
+	/**
+	 * @since 4.3
+	 */
+	public void setSuggestBuilder(SuggestBuilder suggestBuilder) {
+		this.suggestBuilder = suggestBuilder;
+	}
+
+	/**
+	 * @since 4.3
+	 */
+	@Nullable
+	public SuggestBuilder getSuggestBuilder() {
+		return suggestBuilder;
+	}
 }

src/main/java/org/springframework/data/elasticsearch/core/query/NativeSearchQueryBuilder.java

@@ -31,6 +31,7 @@ import org.elasticsearch.search.aggregations.PipelineAggregationBuilder;
 import org.elasticsearch.search.collapse.CollapseBuilder;
 import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
 import org.elasticsearch.search.sort.SortBuilder;
+import org.elasticsearch.search.suggest.SuggestBuilder;
 import org.springframework.data.domain.Pageable;
 import org.springframework.lang.Nullable;
 
@@ -76,6 +77,7 @@ public class NativeSearchQueryBuilder {
 	@Nullable private Boolean trackTotalHits;
 	@Nullable private Duration timeout;
 	private final List<RescorerQuery> rescorerQueries = new ArrayList<>();
+	@Nullable private SuggestBuilder suggestBuilder;
 
 	public NativeSearchQueryBuilder withQuery(QueryBuilder queryBuilder) {
 		this.queryBuilder = queryBuilder;
@@ -311,6 +313,14 @@ public class NativeSearchQueryBuilder {
 		return this;
 	}
 
+	/**
+	 * @since 4.3
+	 */
+	public NativeSearchQueryBuilder withSuggestBuilder(SuggestBuilder suggestBuilder) {
+		this.suggestBuilder = suggestBuilder;
+		return this;
+	}
+
 	public NativeSearchQuery build() {
 
 		NativeSearchQuery nativeSearchQuery = new NativeSearchQuery( //
@@ -393,6 +403,9 @@ public class NativeSearchQueryBuilder {
 			nativeSearchQuery.setRescorerQueries(rescorerQueries);
 		}
 
+		if (suggestBuilder != null) {
+			nativeSearchQuery.setSuggestBuilder(suggestBuilder);
+		}
 		return nativeSearchQuery;
 	}
 }

src/main/java/org/springframework/data/elasticsearch/core/suggest/Completion.java

@@ -1,4 +1,19 @@
-package org.springframework.data.elasticsearch.core.completion;
+/*
+ * 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.elasticsearch.core.suggest;
 
 import java.util.List;
 import java.util.Map;

src/main/java/org/springframework/data/elasticsearch/core/suggest/package-info.java

@@ -0,0 +1,18 @@
+/*
+ * 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.
+ */
+@org.springframework.lang.NonNullApi
+@org.springframework.lang.NonNullFields
+package org.springframework.data.elasticsearch.core.suggest;

src/main/java/org/springframework/data/elasticsearch/core/suggest/response/CompletionSuggestion.java

@@ -0,0 +1,80 @@
+/*
+ * 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.elasticsearch.core.suggest.response;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+
+import org.springframework.data.elasticsearch.core.SearchHit;
+import org.springframework.data.elasticsearch.core.document.SearchDocument;
+import org.springframework.data.elasticsearch.support.ScoreDoc;
+import org.springframework.lang.Nullable;
+
+/**
+ * @author Peter-Josef Meisch
+ * @since 4.3
+ */
+public class CompletionSuggestion<T> extends Suggest.Suggestion<CompletionSuggestion.Entry<T>> {
+
+	public CompletionSuggestion(String name, int size, List<Entry<T>> entries) {
+		super(name, size, entries);
+	}
+
+	public static class Entry<T> extends Suggest.Suggestion.Entry<Entry.Option<T>> {
+
+		public Entry(String text, int offset, int length, List<Option<T>> options) {
+			super(text, offset, length, options);
+		}
+
+		public static class Option<T> extends Suggest.Suggestion.Entry.Option {
+
+			private final Map<String, Set<String>> contexts;
+			private final ScoreDoc scoreDoc;
+			@Nullable private final SearchDocument searchDocument;
+			@Nullable private final T hitEntity;
+			@Nullable private SearchHit<T> searchHit;
+
+			public Option(String text, String highlighted, float score, Boolean collateMatch,
+					Map<String, Set<String>> contexts, ScoreDoc scoreDoc, @Nullable SearchDocument searchDocument,
+					@Nullable T hitEntity) {
+				super(text, highlighted, score, collateMatch);
+				this.contexts = contexts;
+				this.scoreDoc = scoreDoc;
+				this.searchDocument = searchDocument;
+				this.hitEntity = hitEntity;
+			}
+
+			public Map<String, Set<String>> getContexts() {
+				return contexts;
+			}
+
+			public ScoreDoc getScoreDoc() {
+				return scoreDoc;
+			}
+
+			@Nullable
+			public SearchHit<T> getSearchHit() {
+				return searchHit;
+			}
+
+			public void updateSearchHit(BiFunction<SearchDocument, T, SearchHit<T>> mapper) {
+				searchHit = mapper.apply(searchDocument, hitEntity);
+			}
+		}
+	}
+}

src/main/java/org/springframework/data/elasticsearch/core/suggest/response/PhraseSuggestion.java

@@ -0,0 +1,50 @@
+/*
+ * 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.elasticsearch.core.suggest.response;
+
+import java.util.List;
+
+/**
+ * @author Peter-Josef Meisch
+ * @since 4.3
+ */
+public class PhraseSuggestion extends Suggest.Suggestion<PhraseSuggestion.Entry> {
+
+	public PhraseSuggestion(String name, int size, List<Entry> entries) {
+		super(name, size, entries);
+	}
+
+	public static class Entry extends Suggest.Suggestion.Entry<Entry.Option> {
+
+		private final double cutoffScore;
+
+		public Entry(String text, int offset, int length, List<Option> options, double cutoffScore) {
+			super(text, offset, length, options);
+			this.cutoffScore = cutoffScore;
+		}
+
+		public double getCutoffScore() {
+			return cutoffScore;
+		}
+
+		public static class Option extends Suggest.Suggestion.Entry.Option {
+
+			public Option(String text, String highlighted, float score, Boolean collateMatch) {
+				super(text, highlighted, score, collateMatch);
+			}
+		}
+	}
+}

src/main/java/org/springframework/data/elasticsearch/core/suggest/response/SortBy.java

@@ -0,0 +1,24 @@
+/*
+ * 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.elasticsearch.core.suggest.response;
+
+/**
+ * @author Peter-Josef Meisch
+ * @since 4.3
+ */
+public enum SortBy {
+	SCORE, FREQUENCY
+}

src/main/java/org/springframework/data/elasticsearch/core/suggest/response/Suggest.java

@@ -0,0 +1,139 @@
+/*
+ * 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.elasticsearch.core.suggest.response;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.lang.Nullable;
+
+/**
+ * Class structure mirroring the Elasticsearch classes for a suggest response.
+ *
+ * @author Peter-Josef Meisch
+ * @since 4.3
+ */
+public class Suggest {
+	private final List<Suggestion<? extends Suggestion.Entry<? extends Suggestion.Entry.Option>>> suggestions;
+	private final Map<String, Suggestion<? extends Suggestion.Entry<? extends Suggestion.Entry.Option>>> suggestionsMap;
+	private final boolean hasScoreDocs;
+
+	public Suggest(List<Suggestion<? extends Suggestion.Entry<? extends Suggestion.Entry.Option>>> suggestions,
+			boolean hasScoreDocs) {
+		this.suggestions = suggestions;
+		this.suggestionsMap = new HashMap<>();
+		suggestions.forEach(suggestion -> suggestionsMap.put(suggestion.getName(), suggestion));
+		this.hasScoreDocs = hasScoreDocs;
+	}
+
+	public List<Suggestion<? extends Suggestion.Entry<? extends Suggestion.Entry.Option>>> getSuggestions() {
+		return suggestions;
+	}
+
+	public Suggestion<? extends Suggestion.Entry<? extends Suggestion.Entry.Option>> getSuggestion(String name) {
+		return suggestionsMap.get(name);
+	}
+
+	public boolean hasScoreDocs() {
+		return hasScoreDocs;
+	}
+
+	public abstract static class Suggestion<E extends Suggestion.Entry<? extends Suggestion.Entry.Option>> {
+		private final String name;
+		private final int size;
+		private final List<E> entries;
+
+		public Suggestion(String name, int size, List<E> entries) {
+			this.name = name;
+			this.size = size;
+			this.entries = entries;
+		}
+
+		public String getName() {
+			return name;
+		}
+
+		public int getSize() {
+			return size;
+		}
+
+		public List<E> getEntries() {
+			return entries;
+		}
+
+		public abstract static class Entry<O extends Entry.Option> {
+			private final String text;
+			private final int offset;
+			private final int length;
+			private final List<O> options;
+
+			public Entry(String text, int offset, int length, List<O> options) {
+				this.text = text;
+				this.offset = offset;
+				this.length = length;
+				this.options = options;
+			}
+
+			public String getText() {
+				return text;
+			}
+
+			public int getOffset() {
+				return offset;
+			}
+
+			public int getLength() {
+				return length;
+			}
+
+			public List<O> getOptions() {
+				return options;
+			}
+
+			public abstract static class Option {
+				private final String text;
+				private final String highlighted;
+				private final float score;
+				@Nullable private final Boolean collateMatch;
+
+				public Option(String text, String highlighted, float score, @Nullable Boolean collateMatch) {
+					this.text = text;
+					this.highlighted = highlighted;
+					this.score = score;
+					this.collateMatch = collateMatch;
+				}
+
+				public String getText() {
+					return text;
+				}
+
+				public String getHighlighted() {
+					return highlighted;
+				}
+
+				public float getScore() {
+					return score;
+				}
+
+				@Nullable
+				public Boolean getCollateMatch() {
+					return collateMatch;
+				}
+			}
+		}
+	}
+}

src/main/java/org/springframework/data/elasticsearch/core/suggest/response/TermSuggestion.java

@@ -0,0 +1,56 @@
+/*
+ * 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.elasticsearch.core.suggest.response;
+
+import java.util.List;
+
+/**
+ * @author Peter-Josef Meisch
+ */
+public class TermSuggestion extends Suggest.Suggestion<TermSuggestion.Entry> {
+
+	private final SortBy sort;
+
+	public TermSuggestion(String name, int size, List<Entry> entries, SortBy sort) {
+		super(name, size, entries);
+		this.sort = sort;
+	}
+
+	public SortBy getSort() {
+		return sort;
+	}
+
+	public static class Entry extends Suggest.Suggestion.Entry<Entry.Option> {
+
+		public Entry(String text, int offset, int length, List<Option> options) {
+			super(text, offset, length, options);
+		}
+
+		public static class Option extends Suggest.Suggestion.Entry.Option {
+
+			private final int freq;
+
+			public Option(String text, String highlighted, float score, Boolean collateMatch, int freq) {
+				super(text, highlighted, score, collateMatch);
+				this.freq = freq;
+			}
+
+			public int getFreq() {
+				return freq;
+			}
+		}
+	}
+}

src/main/java/org/springframework/data/elasticsearch/core/suggest/response/package-info.java

@@ -1,3 +1,3 @@
 @org.springframework.lang.NonNullApi
 @org.springframework.lang.NonNullFields
-package org.springframework.data.elasticsearch.core.completion;
+package org.springframework.data.elasticsearch.core.suggest.response;

src/main/java/org/springframework/data/elasticsearch/repository/query/ElasticsearchPartQuery.java

@@ -113,7 +113,8 @@ public class ElasticsearchPartQuery extends AbstractElasticsearchRepositoryQuery
 				int itemCount = (int) elasticsearchOperations.count(query, clazz, index);
 
 				if (itemCount == 0) {
-					result = new SearchHitsImpl<>(0, TotalHitsRelation.EQUAL_TO, Float.NaN, null, Collections.emptyList(), null);
+					result = new SearchHitsImpl<>(0, TotalHitsRelation.EQUAL_TO, Float.NaN, null, Collections.emptyList(), null,
+							null);
 				} else {
 					query.setPageable(PageRequest.of(0, Math.max(1, itemCount)));
 				}

src/main/java/org/springframework/data/elasticsearch/support/ScoreDoc.java

@@ -0,0 +1,45 @@
+/*
+ * 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.elasticsearch.support;
+
+/**
+ * @author Peter-Josef Meisch
+ * @since 4.3
+ */
+public class ScoreDoc {
+
+	private final float score;
+	private final int doc;
+	private final int shardIndex;
+
+	public ScoreDoc(float score, int doc, int shardIndex) {
+		this.score = score;
+		this.doc = doc;
+		this.shardIndex = shardIndex;
+	}
+
+	public float getScore() {
+		return score;
+	}
+
+	public int getDoc() {
+		return doc;
+	}
+
+	public int getShardIndex() {
+		return shardIndex;
+	}
+}

src/test/java/org/springframework/data/elasticsearch/annotations/ComposableAnnotationsUnitTest.java

@@ -32,7 +32,7 @@ import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.springframework.core.annotation.AliasFor;
 import org.springframework.data.annotation.Id;
-import org.springframework.data.elasticsearch.core.completion.Completion;
+import org.springframework.data.elasticsearch.core.suggest.Completion;
 import org.springframework.data.elasticsearch.core.convert.MappingElasticsearchConverter;
 import org.springframework.data.elasticsearch.core.index.MappingBuilder;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentProperty;

src/test/java/org/springframework/data/elasticsearch/core/SearchHitSupportTest.java

@@ -65,7 +65,7 @@ class SearchHitSupportTest {
 		hits.add(new SearchHit<>(null, null, null, 0, null, null, null, null, null, null, "five"));
 
 		SearchHits<String> originalSearchHits = new SearchHitsImpl<>(hits.size(), TotalHitsRelation.EQUAL_TO, 0, "scroll",
-				hits, null);
+				hits, null, null);
 
 		SearchPage<String> searchPage = SearchHitSupport.searchPageFor(originalSearchHits, PageRequest.of(0, 3));
 		SearchHits<String> searchHits = searchPage.getSearchHits();

src/test/java/org/springframework/data/elasticsearch/core/StreamQueriesTest.java

@@ -83,6 +83,7 @@ public class StreamQueriesTest {
 
 		assertThat(clearScrollCalled).isTrue();
 	}
+
 	private SearchHit<String> getOneSearchHit() {
 		return new SearchHit<String>(null, null, null, 0, null, null, null, null, null, null, "one");
 	}
@@ -179,6 +180,6 @@ public class StreamQueriesTest {
 	}
 
 	private SearchScrollHits<String> newSearchScrollHits(List<SearchHit<String>> hits, String scrollId) {
-		return new SearchHitsImpl<String>(hits.size(), TotalHitsRelation.EQUAL_TO, 0, scrollId, hits, null);
+		return new SearchHitsImpl<String>(hits.size(), TotalHitsRelation.EQUAL_TO, 0, scrollId, hits, null, null);
 	}
 }

src/test/java/org/springframework/data/elasticsearch/core/index/MappingBuilderIntegrationTests.java

@@ -47,7 +47,7 @@ import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
 import org.springframework.data.elasticsearch.core.IndexOperations;
 import org.springframework.data.elasticsearch.core.MappingContextBaseTests;
 import org.springframework.data.elasticsearch.core.SearchHits;
-import org.springframework.data.elasticsearch.core.completion.Completion;
+import org.springframework.data.elasticsearch.core.suggest.Completion;
 import org.springframework.data.elasticsearch.core.geo.GeoPoint;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.elasticsearch.core.query.IndexQuery;

src/test/java/org/springframework/data/elasticsearch/core/index/MappingBuilderUnitTests.java

@@ -42,7 +42,7 @@ import org.springframework.data.annotation.Id;
 import org.springframework.data.annotation.Transient;
 import org.springframework.data.elasticsearch.annotations.*;
 import org.springframework.data.elasticsearch.core.MappingContextBaseTests;
-import org.springframework.data.elasticsearch.core.completion.Completion;
+import org.springframework.data.elasticsearch.core.suggest.Completion;
 import org.springframework.data.elasticsearch.core.geo.GeoPoint;
 import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMappingContext;
 import org.springframework.data.elasticsearch.core.query.SeqNoPrimaryTerm;

src/test/java/org/springframework/data/elasticsearch/core/suggest/ElasticsearchTemplateCompletionTests.java

@@ -13,19 +13,16 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.data.elasticsearch.core.completion;
+package org.springframework.data.elasticsearch.core.suggest;
 
 import static org.assertj.core.api.Assertions.*;
 
 import java.util.ArrayList;
 import java.util.List;
 
-import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.common.unit.Fuzziness;
 import org.elasticsearch.search.suggest.SuggestBuilder;
 import org.elasticsearch.search.suggest.SuggestBuilders;
-import org.elasticsearch.search.suggest.SuggestionBuilder;
-import org.elasticsearch.search.suggest.completion.CompletionSuggestion;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -35,10 +32,14 @@ import org.springframework.context.annotation.Import;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.elasticsearch.annotations.CompletionField;
 import org.springframework.data.elasticsearch.annotations.Document;
-import org.springframework.data.elasticsearch.core.AbstractElasticsearchTemplate;
 import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
+import org.springframework.data.elasticsearch.core.SearchHits;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.elasticsearch.core.query.IndexQuery;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
+import org.springframework.data.elasticsearch.core.suggest.response.CompletionSuggestion;
+import org.springframework.data.elasticsearch.core.suggest.response.Suggest;
 import org.springframework.data.elasticsearch.junit.jupiter.ElasticsearchRestTemplateConfiguration;
 import org.springframework.data.elasticsearch.junit.jupiter.SpringIntegrationTest;
 import org.springframework.data.elasticsearch.utils.IndexInitializer;
@@ -131,23 +132,26 @@ public class ElasticsearchTemplateCompletionTests {
 	@Test
 	public void shouldFindSuggestionsForGivenCriteriaQueryUsingCompletionEntity() {
 
-		// given
 		loadCompletionObjectEntities();
+		NativeSearchQuery query = new NativeSearchQueryBuilder().withSuggestBuilder(new SuggestBuilder()
+				.addSuggestion("test-suggest", SuggestBuilders.completionSuggestion("suggest").prefix("m", Fuzziness.AUTO)))
+				.build();
 
-		SuggestionBuilder completionSuggestionFuzzyBuilder = SuggestBuilders.completionSuggestion("suggest").prefix("m",
-				Fuzziness.AUTO);
+		SearchHits<CompletionEntity> searchHits = operations.search(query, CompletionEntity.class);
 
-		// when
-		SearchResponse suggestResponse = ((AbstractElasticsearchTemplate) operations).suggest(
-				new SuggestBuilder().addSuggestion("test-suggest", completionSuggestionFuzzyBuilder),
-				IndexCoordinates.of("test-index-core-completion"));
-		CompletionSuggestion completionSuggestion = suggestResponse.getSuggest().getSuggestion("test-suggest");
-		List<CompletionSuggestion.Entry.Option> options = completionSuggestion.getEntries().get(0).getOptions();
-
-		// then
+		assertThat(searchHits.hasSuggest()).isTrue();
+		Suggest suggest = searchHits.getSuggest();
+		// noinspection ConstantConditions
+		Suggest.Suggestion<? extends Suggest.Suggestion.Entry<? extends Suggest.Suggestion.Entry.Option>> suggestion = suggest
+				.getSuggestion("test-suggest");
+		assertThat(suggestion).isNotNull();
+		assertThat(suggestion).isInstanceOf(CompletionSuggestion.class);
+		// noinspection unchecked
+		List<CompletionSuggestion.Entry.Option<AnnotatedCompletionEntity>> options = ((CompletionSuggestion<AnnotatedCompletionEntity>) suggestion)
+				.getEntries().get(0).getOptions();
 		assertThat(options).hasSize(2);
-		assertThat(options.get(0).getText().string()).isIn("Marchand", "Mohsin");
-		assertThat(options.get(1).getText().string()).isIn("Marchand", "Mohsin");
+		assertThat(options.get(0).getText()).isIn("Marchand", "Mohsin");
+		assertThat(options.get(1).getText()).isIn("Marchand", "Mohsin");
 	}
 
 	@Test // DATAES-754
@@ -160,43 +164,52 @@ public class ElasticsearchTemplateCompletionTests {
 	@Test
 	public void shouldFindSuggestionsForGivenCriteriaQueryUsingAnnotatedCompletionEntity() {
 
-		// given
 		loadAnnotatedCompletionObjectEntities();
-		SuggestionBuilder completionSuggestionFuzzyBuilder = SuggestBuilders.completionSuggestion("suggest").prefix("m",
-				Fuzziness.AUTO);
+		NativeSearchQuery query = new NativeSearchQueryBuilder().withSuggestBuilder(new SuggestBuilder()
+				.addSuggestion("test-suggest", SuggestBuilders.completionSuggestion("suggest").prefix("m", Fuzziness.AUTO)))
+				.build();
 
-		// when
-		SearchResponse suggestResponse = ((AbstractElasticsearchTemplate) operations).suggest(
-				new SuggestBuilder().addSuggestion("test-suggest", completionSuggestionFuzzyBuilder),
-				IndexCoordinates.of("test-index-annotated-completion"));
-		CompletionSuggestion completionSuggestion = suggestResponse.getSuggest().getSuggestion("test-suggest");
-		List<CompletionSuggestion.Entry.Option> options = completionSuggestion.getEntries().get(0).getOptions();
+		SearchHits<AnnotatedCompletionEntity> searchHits = operations.search(query, AnnotatedCompletionEntity.class);
 
-		// then
+		assertThat(searchHits.hasSuggest()).isTrue();
+		Suggest suggest = searchHits.getSuggest();
+		// noinspection ConstantConditions
+		Suggest.Suggestion<? extends Suggest.Suggestion.Entry<? extends Suggest.Suggestion.Entry.Option>> suggestion = suggest
+				.getSuggestion("test-suggest");
+		assertThat(suggestion).isNotNull();
+		assertThat(suggestion).isInstanceOf(CompletionSuggestion.class);
+		// noinspection unchecked
+		List<CompletionSuggestion.Entry.Option<AnnotatedCompletionEntity>> options = ((CompletionSuggestion<AnnotatedCompletionEntity>) suggestion)
+				.getEntries().get(0).getOptions();
 		assertThat(options).hasSize(2);
-		assertThat(options.get(0).getText().string()).isIn("Marchand", "Mohsin");
-		assertThat(options.get(1).getText().string()).isIn("Marchand", "Mohsin");
+		assertThat(options.get(0).getText()).isIn("Marchand", "Mohsin");
+		assertThat(options.get(1).getText()).isIn("Marchand", "Mohsin");
 	}
 
 	@Test
 	public void shouldFindSuggestionsWithWeightsForGivenCriteriaQueryUsingAnnotatedCompletionEntity() {
 
-		// given
 		loadAnnotatedCompletionObjectEntitiesWithWeights();
-		SuggestionBuilder completionSuggestionFuzzyBuilder = SuggestBuilders.completionSuggestion("suggest").prefix("m",
-				Fuzziness.AUTO);
+		NativeSearchQuery query = new NativeSearchQueryBuilder().withSuggestBuilder(new SuggestBuilder()
+				.addSuggestion("test-suggest", SuggestBuilders.completionSuggestion("suggest").prefix("m", Fuzziness.AUTO)))
+				.build();
 
-		// when
-		SearchResponse suggestResponse = ((AbstractElasticsearchTemplate) operations).suggest(
-				new SuggestBuilder().addSuggestion("test-suggest", completionSuggestionFuzzyBuilder),
-				IndexCoordinates.of("test-index-annotated-completion"));
-		CompletionSuggestion completionSuggestion = suggestResponse.getSuggest().getSuggestion("test-suggest");
-		List<CompletionSuggestion.Entry.Option> options = completionSuggestion.getEntries().get(0).getOptions();
+		SearchHits<AnnotatedCompletionEntity> searchHits = operations.search(query, AnnotatedCompletionEntity.class);
+
+		assertThat(searchHits.hasSuggest()).isTrue();
+		Suggest suggest = searchHits.getSuggest();
+		// noinspection ConstantConditions
+		Suggest.Suggestion<? extends Suggest.Suggestion.Entry<? extends Suggest.Suggestion.Entry.Option>> suggestion = suggest
+				.getSuggestion("test-suggest");
+		assertThat(suggestion).isNotNull();
+		assertThat(suggestion).isInstanceOf(CompletionSuggestion.class);
+		// noinspection unchecked
+		List<CompletionSuggestion.Entry.Option<AnnotatedCompletionEntity>> options = ((CompletionSuggestion<AnnotatedCompletionEntity>) suggestion)
+				.getEntries().get(0).getOptions();
 
-		// then
 		assertThat(options).hasSize(4);
-		for (CompletionSuggestion.Entry.Option option : options) {
-			switch (option.getText().string()) {
+		for (CompletionSuggestion.Entry.Option<AnnotatedCompletionEntity> option : options) {
+			switch (option.getText()) {
 				case "Mewes Kochheim1":
 					assertThat(option.getScore()).isEqualTo(4);
 					break;
@@ -216,10 +229,6 @@ public class ElasticsearchTemplateCompletionTests {
 		}
 	}
 
-	/**
-	 * @author Rizwan Idrees
-	 * @author Mohsin Husen
-	 */
 	static class NonDocumentEntity {
 
 		@Nullable @Id private String someId;
@@ -245,9 +254,6 @@ public class ElasticsearchTemplateCompletionTests {
 		}
 	}
 
-	/**
-	 * @author Mewes Kochheim
-	 */
 	@Document(indexName = "test-index-core-completion")
 	static class CompletionEntity {
 
@@ -291,9 +297,6 @@ public class ElasticsearchTemplateCompletionTests {
 		}
 	}
 
-	/**
-	 * @author Mewes Kochheim
-	 */
 	static class CompletionEntityBuilder {
 
 		private CompletionEntity result;

src/test/java/org/springframework/data/elasticsearch/core/suggest/ElasticsearchTemplateCompletionTransportTests.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.data.elasticsearch.core.completion;
+package org.springframework.data.elasticsearch.core.suggest;
 
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;

src/test/java/org/springframework/data/elasticsearch/core/suggest/ElasticsearchTemplateCompletionWithContextsTests.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.data.elasticsearch.core.completion;
+package org.springframework.data.elasticsearch.core.suggest;
 
 import static org.assertj.core.api.Assertions.*;
 

src/test/java/org/springframework/data/elasticsearch/core/suggest/ElasticsearchTemplateCompletionWithContextsTransportTests.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.data.elasticsearch.core.completion;
+package org.springframework.data.elasticsearch.core.suggest;
 
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;

src/test/java/org/springframework/data/elasticsearch/core/suggest/ReactiveElasticsearchTemplateSuggestIntegrationTests.java

@@ -0,0 +1,212 @@
+/*
+ * 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.elasticsearch.core.suggest;
+
+import static org.assertj.core.api.Assertions.*;
+
+import reactor.test.StepVerifier;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.elasticsearch.common.unit.Fuzziness;
+import org.elasticsearch.search.suggest.SuggestBuilder;
+import org.elasticsearch.search.suggest.SuggestBuilders;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.elasticsearch.annotations.CompletionField;
+import org.springframework.data.elasticsearch.annotations.Document;
+import org.springframework.data.elasticsearch.core.ReactiveElasticsearchOperations;
+import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.data.elasticsearch.core.query.IndexQuery;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
+import org.springframework.data.elasticsearch.core.suggest.response.CompletionSuggestion;
+import org.springframework.data.elasticsearch.core.suggest.response.Suggest;
+import org.springframework.data.elasticsearch.junit.jupiter.ReactiveElasticsearchRestTemplateConfiguration;
+import org.springframework.data.elasticsearch.junit.jupiter.SpringIntegrationTest;
+import org.springframework.data.elasticsearch.utils.IndexNameProvider;
+import org.springframework.lang.Nullable;
+
+/**
+ * @author Peter-Josef Meisch
+ */
+@SuppressWarnings("SpringJavaAutowiredMembersInspection")
+@SpringIntegrationTest
+public class ReactiveElasticsearchTemplateSuggestIntegrationTests {
+	@Configuration
+	@Import({ ReactiveElasticsearchRestTemplateConfiguration.class })
+	static class Config {
+		@Bean
+		IndexNameProvider indexNameProvider() {
+			return new IndexNameProvider("reactive-template-suggest");
+		}
+	}
+
+	@Autowired private ReactiveElasticsearchOperations operations;
+
+	@Autowired private IndexNameProvider indexNameProvider;
+
+	// region Setup
+	@BeforeEach
+	public void beforeEach() {
+
+		indexNameProvider.increment();
+		operations.indexOps(CompletionEntity.class).createWithMapping().block();
+	}
+
+	@Test // #1302
+	@DisplayName("should do some test")
+	void shouldDoSomeTest() {
+
+		assertThat(operations).isNotNull();
+	}
+
+	@Test // #1302
+	@DisplayName("should find suggestions for given prefix completion")
+	void shouldFindSuggestionsForGivenPrefixCompletion() {
+
+		loadCompletionObjectEntities();
+
+		NativeSearchQuery query = new NativeSearchQueryBuilder().withSuggestBuilder(new SuggestBuilder()
+				.addSuggestion("test-suggest", SuggestBuilders.completionSuggestion("suggest").prefix("m", Fuzziness.AUTO)))
+				.build();
+
+		operations.suggest(query, CompletionEntity.class) //
+				.as(StepVerifier::create) //
+				.assertNext(suggest -> {
+					Suggest.Suggestion<? extends Suggest.Suggestion.Entry<? extends Suggest.Suggestion.Entry.Option>> suggestion = suggest
+							.getSuggestion("test-suggest");
+					assertThat(suggestion).isNotNull();
+					assertThat(suggestion).isInstanceOf(CompletionSuggestion.class);
+					// noinspection unchecked
+					List<CompletionSuggestion.Entry.Option<ElasticsearchTemplateCompletionTests.AnnotatedCompletionEntity>> options = ((CompletionSuggestion<ElasticsearchTemplateCompletionTests.AnnotatedCompletionEntity>) suggestion)
+							.getEntries().get(0).getOptions();
+					assertThat(options).hasSize(2);
+					assertThat(options.get(0).getText()).isIn("Marchand", "Mohsin");
+					assertThat(options.get(1).getText()).isIn("Marchand", "Mohsin");
+
+				}) //
+				.verifyComplete();
+	}
+
+	// region helper functions
+	private void loadCompletionObjectEntities() {
+
+		CompletionEntity rizwan_idrees = new CompletionEntityBuilder("1").name("Rizwan Idrees")
+				.suggest(new String[] { "Rizwan Idrees" }).build();
+		CompletionEntity franck_marchand = new CompletionEntityBuilder("2").name("Franck Marchand")
+				.suggest(new String[] { "Franck", "Marchand" }).build();
+		CompletionEntity mohsin_husen = new CompletionEntityBuilder("3").name("Mohsin Husen")
+				.suggest(new String[] { "Mohsin", "Husen" }).build();
+		CompletionEntity artur_konczak = new CompletionEntityBuilder("4").name("Artur Konczak")
+				.suggest(new String[] { "Artur", "Konczak" }).build();
+		List<CompletionEntity> entities = new ArrayList<>(
+				Arrays.asList(rizwan_idrees, franck_marchand, mohsin_husen, artur_konczak));
+		IndexCoordinates index = IndexCoordinates.of(indexNameProvider.indexName());
+		operations.saveAll(entities, index).blockLast();
+	}
+	// endregion
+
+	// region Entities
+	@Document(indexName = "#{@indexNameProvider.indexName()}")
+	static class CompletionEntity {
+
+		@Nullable @Id private String id;
+
+		@Nullable private String name;
+
+		@Nullable @CompletionField(maxInputLength = 100) private Completion suggest;
+
+		private CompletionEntity() {}
+
+		public CompletionEntity(String id) {
+			this.id = id;
+		}
+
+		@Nullable
+		public String getId() {
+			return id;
+		}
+
+		public void setId(String id) {
+			this.id = id;
+		}
+
+		@Nullable
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+		@Nullable
+		public Completion getSuggest() {
+			return suggest;
+		}
+
+		public void setSuggest(Completion suggest) {
+			this.suggest = suggest;
+		}
+	}
+
+	static class CompletionEntityBuilder {
+
+		private CompletionEntity result;
+
+		public CompletionEntityBuilder(String id) {
+			result = new CompletionEntity(id);
+		}
+
+		public CompletionEntityBuilder name(String name) {
+			result.setName(name);
+			return this;
+		}
+
+		public CompletionEntityBuilder suggest(String[] input) {
+			return suggest(input, null);
+		}
+
+		public CompletionEntityBuilder suggest(String[] input, Integer weight) {
+			Completion suggest = new Completion(input);
+			suggest.setWeight(weight);
+
+			result.setSuggest(suggest);
+			return this;
+		}
+
+		public CompletionEntity build() {
+			return result;
+		}
+
+		public IndexQuery buildIndex() {
+			IndexQuery indexQuery = new IndexQuery();
+			indexQuery.setId(result.getId());
+			indexQuery.setObject(result);
+			return indexQuery;
+		}
+	}
+	// endregion
+}