Implement update by query

Author: faridfaoudi

src/main/java/org/springframework/data/elasticsearch/client/reactive/DefaultReactiveElasticsearchClient.java

@@ -24,6 +24,7 @@
 import io.netty.handler.timeout.WriteTimeoutHandler;
 import org.elasticsearch.client.indices.GetFieldMappingsRequest;
 import org.elasticsearch.client.indices.GetFieldMappingsResponse;
+import org.elasticsearch.index.reindex.UpdateByQueryRequest;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.netty.http.client.HttpClient;
@@ -521,6 +522,13 @@ public Mono<BulkByScrollResponse> deleteBy(HttpHeaders headers, DeleteByQueryReq
 				.next();
 	}
 
+	@Override
+	public Mono<org.springframework.data.elasticsearch.core.query.BulkByScrollResponse> updateBy(HttpHeaders headers, UpdateByQueryRequest updateRequest) {
+		return sendRequest(updateRequest, requestCreator.updateByQuery(), BulkByScrollResponse.class, headers) //
+				.next() //
+				.map(org.springframework.data.elasticsearch.core.query.BulkByScrollResponse::translateFrom);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient#bulk(org.springframework.http.HttpHeaders, org.elasticsearch.action.bulk.BulkRequest)

src/main/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClient.java

@@ -17,6 +17,7 @@
 
 import org.elasticsearch.client.indices.GetFieldMappingsRequest;
 import org.elasticsearch.client.indices.GetFieldMappingsResponse;
+import org.elasticsearch.index.reindex.UpdateByQueryRequest;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -596,6 +597,44 @@ default Mono<BulkByScrollResponse> deleteBy(DeleteByQueryRequest deleteRequest)
 	 */
 	Mono<BulkByScrollResponse> deleteBy(HttpHeaders headers, DeleteByQueryRequest deleteRequest);
 
+	/**
+	 * Execute a {@link UpdateByQueryRequest} against the {@literal update by query} API.
+	 *
+	 * @param consumer never {@literal null}.
+	 * @see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update-by-query.html">Update By
+	 * 	 *      Query API on elastic.co</a>
+	 * @return a {@link Mono} emitting operation response.
+	 */
+	default Mono<org.springframework.data.elasticsearch.core.query.BulkByScrollResponse> updateBy(Consumer<UpdateByQueryRequest> consumer){
+
+		final UpdateByQueryRequest request = new UpdateByQueryRequest();
+		consumer.accept(request);
+		return updateBy(request);
+	}
+
+	/**
+	 * Execute a {@link UpdateByQueryRequest} against the {@literal update by query} API.
+	 *
+	 * @param updateRequest must not be {@literal null}.
+	 * @see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update-by-query.html">Update By
+	 * 	 *      Query API on elastic.co</a>
+	 * @return a {@link Mono} emitting operation response.
+	 */
+	default Mono<org.springframework.data.elasticsearch.core.query.BulkByScrollResponse> updateBy(UpdateByQueryRequest updateRequest){
+		return updateBy(HttpHeaders.EMPTY, updateRequest);
+	}
+
+	/**
+	 * Execute a {@link UpdateByQueryRequest} against the {@literal update by query} API.
+	 *
+	 * @param headers Use {@link HttpHeaders} to provide eg. authentication data. Must not be {@literal null}.
+	 * @param updateRequest must not be {@literal null}.
+	 * @see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update-by-query.html">Update By
+	 * 	 *      Query API on elastic.co</a>
+	 * @return a {@link Mono} emitting operation response.
+	 */
+	Mono<org.springframework.data.elasticsearch.core.query.BulkByScrollResponse> updateBy(HttpHeaders headers, UpdateByQueryRequest updateRequest);
+
 	/**
 	 * Execute a {@link BulkRequest} against the {@literal bulk} API.
 	 *

src/main/java/org/springframework/data/elasticsearch/client/reactive/RequestCreator.java

@@ -33,6 +33,7 @@
 import org.elasticsearch.client.indices.IndexTemplatesExistRequest;
 import org.elasticsearch.client.indices.PutIndexTemplateRequest;
 import org.elasticsearch.index.reindex.DeleteByQueryRequest;
+import org.elasticsearch.index.reindex.UpdateByQueryRequest;
 import org.springframework.data.elasticsearch.UncategorizedElasticsearchException;
 import org.springframework.data.elasticsearch.client.util.RequestConverters;
 
@@ -91,6 +92,10 @@ default Function<DeleteByQueryRequest, Request> deleteByQuery() {
 		return RequestConverters::deleteByQuery;
 	}
 
+	default Function<UpdateByQueryRequest, Request> updateByQuery() {
+		return RequestConverters::updateByQuery;
+	}
+
 	default Function<BulkRequest, Request> bulk() {
 
 		return request -> {

src/main/java/org/springframework/data/elasticsearch/client/util/RequestConverters.java

@@ -540,8 +540,10 @@ private static Request prepareReindexRequest(ReindexRequest reindexRequest, bool
 	public static Request updateByQuery(UpdateByQueryRequest updateByQueryRequest) {
 		String endpoint = endpoint(updateByQueryRequest.indices(), updateByQueryRequest.getDocTypes(), "_update_by_query");
 		Request request = new Request(HttpMethod.POST.name(), endpoint);
-		Params params = new Params(request).withRouting(updateByQueryRequest.getRouting())
-				.withPipeline(updateByQueryRequest.getPipeline()).withRefresh(updateByQueryRequest.isRefresh())
+		Params params = new Params(request)
+				.withRouting(updateByQueryRequest.getRouting())
+				.withPipeline(updateByQueryRequest.getPipeline())
+				.withRefresh(updateByQueryRequest.isRefresh())
 				.withTimeout(updateByQueryRequest.getTimeout())
 				.withWaitForActiveShards(updateByQueryRequest.getWaitForActiveShards())
 				.withRequestsPerSecond(updateByQueryRequest.getRequestsPerSecond())
@@ -555,8 +557,8 @@ public static Request updateByQuery(UpdateByQueryRequest updateByQueryRequest) {
 		if (updateByQueryRequest.getScrollTime() != AbstractBulkByScrollRequest.DEFAULT_SCROLL_TIMEOUT) {
 			params.putParam("scroll", updateByQueryRequest.getScrollTime());
 		}
-		if (updateByQueryRequest.getSize() > 0) {
-			params.putParam("size", Integer.toString(updateByQueryRequest.getSize()));
+		if (updateByQueryRequest.getMaxDocs() > 0) {
+			params.putParam("max_docs", Integer.toString(updateByQueryRequest.getMaxDocs()));
 		}
 		request.setEntity(createEntity(updateByQueryRequest, REQUEST_BODY_CONTENT_TYPE));
 		return request;

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

@@ -18,6 +18,7 @@
 import java.util.List;
 
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.data.elasticsearch.core.query.BulkByScrollResponse;
 import org.springframework.data.elasticsearch.core.query.BulkOptions;
 import org.springframework.data.elasticsearch.core.query.DeleteQuery;
 import org.springframework.data.elasticsearch.core.query.GetQuery;
@@ -32,6 +33,7 @@
  * <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs.html">Elasticsearch Document APIs</a>.
  *
  * @author Peter-Josef Meisch
+ * @author Farid Faoudi
  * @since 4.0
  */
 public interface DocumentOperations {
@@ -297,6 +299,16 @@ default String delete(String id, IndexCoordinates index) {
 	 */
 	UpdateResponse update(UpdateQuery updateQuery, IndexCoordinates index);
 
+	/**
+	 * Update document(s) by query
+	 *
+	 * @param updateQuery query defining the update
+	 * @param index the index where to update the records
+	 * @return the update response
+	 * @since 4.2
+	 */
+	BulkByScrollResponse updateByQuery(UpdateQuery updateQuery, IndexCoordinates index);
+
 	// region deprecated
 	/**
 	 * Delete all records matching the query.

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

@@ -38,6 +38,7 @@
 import org.elasticsearch.client.RestHighLevelClient;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.reindex.DeleteByQueryRequest;
+import org.elasticsearch.index.reindex.UpdateByQueryRequest;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 import org.elasticsearch.search.suggest.SuggestBuilder;
 import org.slf4j.Logger;
@@ -46,6 +47,7 @@
 import org.springframework.data.elasticsearch.core.document.DocumentAdapters;
 import org.springframework.data.elasticsearch.core.document.SearchDocumentResponse;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.data.elasticsearch.core.query.BulkByScrollResponse;
 import org.springframework.data.elasticsearch.core.query.BulkOptions;
 import org.springframework.data.elasticsearch.core.query.DeleteQuery;
 import org.springframework.data.elasticsearch.core.query.IndexQuery;
@@ -87,6 +89,7 @@
  * @author Mathias Teier
  * @author Gyula Attila Csorogi
  * @author Massimiliano Poggi
+ * @author Farid Faoudi
  */
 public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 
@@ -223,6 +226,13 @@ public UpdateResponse update(UpdateQuery query, IndexCoordinates index) {
 		return new UpdateResponse(result);
 	}
 
+	@Override
+	public BulkByScrollResponse updateByQuery(UpdateQuery query, IndexCoordinates index) {
+		final UpdateByQueryRequest updateByQueryRequest = requestFactory.updateByQueryRequest(query, index);
+		final org.elasticsearch.index.reindex.BulkByScrollResponse bulkByScrollResponse = execute(client -> client.updateByQuery(updateByQueryRequest, RequestOptions.DEFAULT));
+		return BulkByScrollResponse.translateFrom(bulkByScrollResponse);
+	}
+
 	public List<IndexedObjectInformation> doBulkOperation(List<?> queries, BulkOptions bulkOptions,
 			IndexCoordinates index) {
 		BulkRequest bulkRequest = prepareWriteRequest(requestFactory.bulkRequest(queries, bulkOptions, index));

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

@@ -36,6 +36,7 @@
 import org.elasticsearch.action.update.UpdateRequestBuilder;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.index.reindex.UpdateByQueryRequestBuilder;
 import org.elasticsearch.search.suggest.SuggestBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -44,6 +45,7 @@
 import org.springframework.data.elasticsearch.core.document.DocumentAdapters;
 import org.springframework.data.elasticsearch.core.document.SearchDocumentResponse;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.data.elasticsearch.core.query.BulkByScrollResponse;
 import org.springframework.data.elasticsearch.core.query.BulkOptions;
 import org.springframework.data.elasticsearch.core.query.DeleteQuery;
 import org.springframework.data.elasticsearch.core.query.IndexQuery;
@@ -81,6 +83,7 @@
  * @author Farid Azaza
  * @author Gyula Attila Csorogi
  * @author Roman Puchkovskiy
+ * @author Farid Faoudi
  * @deprecated as of 4.0
  */
 @Deprecated
@@ -241,6 +244,13 @@ public UpdateResponse update(UpdateQuery query, IndexCoordinates index) {
 		return new UpdateResponse(result);
 	}
 
+	@Override
+	public BulkByScrollResponse updateByQuery(UpdateQuery query, IndexCoordinates index) {
+		final UpdateByQueryRequestBuilder updateByQueryRequestBuilder = requestFactory.updateByQueryRequestBuilderFor(client, query, index);
+		final org.elasticsearch.index.reindex.BulkByScrollResponse bulkByScrollResponse = updateByQueryRequestBuilder.execute().actionGet();
+		return BulkByScrollResponse.translateFrom(bulkByScrollResponse);
+	}
+
 	public List<IndexedObjectInformation> doBulkOperation(List<?> queries, BulkOptions bulkOptions,
 			IndexCoordinates index) {
 		BulkRequestBuilder bulkRequestBuilder = requestFactory.bulkRequestBuilder(client, queries, bulkOptions, index);

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.elasticsearch.core;
 
+import org.springframework.data.elasticsearch.core.query.BulkByScrollResponse;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -36,6 +37,7 @@
  * @author Peter-Josef Meisch
  * @author Aleksei Arsenev
  * @author Roman Puchkovskiy
+ * @author Farid Faoudi
  * @since 4.0
  */
 public interface ReactiveDocumentOperations {
@@ -336,4 +338,13 @@ default Mono<String> delete(String id, Class<?> entityType, IndexCoordinates ind
 	 * @since 4.1
 	 */
 	Mono<UpdateResponse> update(UpdateQuery updateQuery, IndexCoordinates index);
+
+	/**
+	 * Update document(s) by query.
+	 * @param updateQuery query defining the update
+	 * @param index the index where to update the records
+	 * @return a {@link Mono} emitting the update response
+	 * @since 4.2
+	 */
+	Mono<BulkByScrollResponse> updateByQuery(UpdateQuery updateQuery, IndexCoordinates index);
 }

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.elasticsearch.core;
 
+import org.elasticsearch.index.reindex.UpdateByQueryRequest;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.util.function.Tuple2;
@@ -95,6 +96,7 @@
  * @author Roman Puchkovskiy
  * @author Russell Parry
  * @author Thomas Geese
+ * @author Farid Faoudi
  * @since 3.2
  */
 public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOperations, ApplicationContextAware {
@@ -539,6 +541,18 @@ public Mono<UpdateResponse> update(UpdateQuery updateQuery, IndexCoordinates ind
 		});
 	}
 
+	@Override
+	public Mono<org.springframework.data.elasticsearch.core.query.BulkByScrollResponse> updateByQuery(UpdateQuery updateQuery, IndexCoordinates index) {
+
+		Assert.notNull(updateQuery, "UpdateQuery must not be null");
+		Assert.notNull(index, "Index must not be null");
+
+		return Mono.defer(() -> {
+			final UpdateByQueryRequest request = requestFactory.updateByQueryRequest(updateQuery, index);
+			return Mono.from(execute(client -> client.updateBy(request)));
+		});
+	}
+
 	@Override
 	public Mono<Long> delete(Query query, Class<?> entityType) {
 		return delete(query, entityType, getIndexCoordinatesFor(entityType));