Support Delete by query with es parameters.

Signed-off-by: SilentDemonSD <drxxstrange@gmail.com>

Author: youssef3wi

src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchTemplate.java

@@ -54,6 +54,7 @@
 import org.springframework.data.elasticsearch.core.query.BaseQueryBuilder;
 import org.springframework.data.elasticsearch.core.query.BulkOptions;
 import org.springframework.data.elasticsearch.core.query.ByQueryResponse;
+import org.springframework.data.elasticsearch.core.query.DeleteQuery;
 import org.springframework.data.elasticsearch.core.query.IndexQuery;
 import org.springframework.data.elasticsearch.core.query.MoreLikeThisQuery;
 import org.springframework.data.elasticsearch.core.query.Query;
@@ -177,6 +178,11 @@ public void bulkUpdate(List<UpdateQuery> queries, BulkOptions bulkOptions, Index
 		doBulkOperation(queries, bulkOptions, index);
 	}
 
+	@Override
+	public ByQueryResponse delete(DeleteQuery query, Class<?> clazz) {
+		return delete(query, clazz, getIndexCoordinatesFor(clazz));
+	}
+
 	@Override
 	public ByQueryResponse delete(Query query, Class<?> clazz, IndexCoordinates index) {
 
@@ -190,6 +196,18 @@ public ByQueryResponse delete(Query query, Class<?> clazz, IndexCoordinates inde
 		return responseConverter.byQueryResponse(response);
 	}
 
+	@Override
+	public ByQueryResponse delete(DeleteQuery query, Class<?> clazz, IndexCoordinates index) {
+		Assert.notNull(query, "query must not be null");
+
+		DeleteByQueryRequest request = requestConverter.documentDeleteByQueryRequest(query, routingResolver.getRouting(),
+				clazz, index, getRefreshPolicy());
+
+		DeleteByQueryResponse response = execute(client -> client.deleteByQuery(request));
+
+		return responseConverter.byQueryResponse(response);
+	}
+
 	@Override
 	public UpdateResponse update(UpdateQuery updateQuery, IndexCoordinates index) {
 

src/main/java/org/springframework/data/elasticsearch/client/elc/ReactiveElasticsearchTemplate.java

@@ -25,6 +25,7 @@
 import co.elastic.clients.json.JsonpMapper;
 import co.elastic.clients.transport.Version;
 import co.elastic.clients.transport.endpoints.BooleanResponse;
+import org.springframework.data.elasticsearch.core.query.DeleteQuery;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.util.function.Tuple2;
@@ -180,6 +181,15 @@ public Mono<ByQueryResponse> delete(Query query, Class<?> entityType, IndexCoord
         return Mono.from(execute(client -> client.deleteByQuery(request))).map(responseConverter::byQueryResponse);
     }
 
+    @Override
+    public Mono<ByQueryResponse> delete(DeleteQuery query, Class<?> entityType, IndexCoordinates index) {
+        Assert.notNull(query, "query must not be null");
+
+        DeleteByQueryRequest request = requestConverter.documentDeleteByQueryRequest(query, routingResolver.getRouting(),
+                entityType, index, getRefreshPolicy());
+        return Mono.from(execute(client -> client.deleteByQuery(request))).map(responseConverter::byQueryResponse);
+    }
+
     @Override
     public <T> Mono<T> get(String id, Class<T> entityType, IndexCoordinates index) {
 

src/main/java/org/springframework/data/elasticsearch/client/elc/RequestConverter.java

@@ -968,6 +968,60 @@ public DeleteByQueryRequest documentDeleteByQueryRequest(Query query, @Nullable
 		});
 	}
 
+	public DeleteByQueryRequest documentDeleteByQueryRequest(DeleteQuery query, @Nullable String routing, Class<?> clazz,
+															 IndexCoordinates index, @Nullable RefreshPolicy refreshPolicy) {
+		Assert.notNull(query, "query must not be null");
+		Assert.notNull(index, "index must not be null");
+
+		// Old documentDeleteByQueryRequest needs to be refactored.
+		return DeleteByQueryRequest.of(dqb -> {
+			dqb.index(Arrays.asList(index.getIndexNames())) //
+					.query(getQuery(query, clazz))//
+					.refresh(deleteByQueryRefresh(refreshPolicy));
+
+			if (query.isLimiting()) {
+				// noinspection ConstantConditions
+				dqb.maxDocs(Long.valueOf(query.getMaxResults()));
+			}
+
+			dqb.scroll(time(query.getScrollTime()))
+					.scrollSize(query.getScrollSize());
+
+			if (query.getRoute() != null) {
+				dqb.routing(query.getRoute());
+			} else if (StringUtils.hasText(routing)) {
+				dqb.routing(routing);
+			}
+
+			if (query.getQ() != null) {
+				dqb.q(query.getQ())
+						.analyzer(query.getAnalyzer())
+						.analyzeWildcard(query.getAnalyzeWildcard())
+						.defaultOperator(query.getDefaultOperator())
+						.df(query.getDf())
+						.lenient(query.getLenient());
+			}
+
+			if (query.getExpandWildcards() != null && !query.getExpandWildcards().isEmpty()) {
+				dqb.expandWildcards(expandWildcards(query.getExpandWildcards()));
+			}
+			dqb.allowNoIndices(query.getAllowNoIndices())
+					.conflicts(query.getConflicts())
+					.ignoreUnavailable(query.getIgnoreUnavailable())
+					.preference(query.getPreference())
+					.requestCache(query.getRequestCache())
+					.searchType(searchType(query.getSearchType()))
+					.searchTimeout(time(query.getSearchTimeout()))
+					.slices(query.getSlices())
+					.stats(query.getStats())
+					.terminateAfter(query.getTerminateAfter())
+					.timeout(time(query.getTimeout()))
+					.version(query.getVersion());
+
+			return dqb;
+		});
+	}
+
 	public UpdateRequest<Document, ?> documentUpdateRequest(UpdateQuery query, IndexCoordinates index,
 			@Nullable RefreshPolicy refreshPolicy, @Nullable String routing) {
 

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.elasticsearch.core;
 
+import org.springframework.data.elasticsearch.core.query.DeleteQuery;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
@@ -412,6 +413,11 @@ public Mono<String> delete(String id, IndexCoordinates index) {
 	public Mono<ByQueryResponse> delete(Query query, Class<?> entityType) {
 		return delete(query, entityType, getIndexCoordinatesFor(entityType));
 	}
+
+	@Override
+	public Mono<ByQueryResponse> delete(DeleteQuery query, Class<?> entityType) {
+		return delete(query, entityType, getIndexCoordinatesFor(entityType));
+	}
 	// endregion
 
 	// region SearchDocument

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

@@ -21,6 +21,7 @@
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.elasticsearch.core.query.BulkOptions;
 import org.springframework.data.elasticsearch.core.query.ByQueryResponse;
+import org.springframework.data.elasticsearch.core.query.DeleteQuery;
 import org.springframework.data.elasticsearch.core.query.IndexQuery;
 import org.springframework.data.elasticsearch.core.query.Query;
 import org.springframework.data.elasticsearch.core.query.UpdateQuery;
@@ -279,9 +280,21 @@ default void bulkUpdate(List<UpdateQuery> queries, IndexCoordinates index) {
 	 *          {@link org.springframework.data.elasticsearch.annotations.Document}
 	 * @return response with detailed information
 	 * @since 4.1
+	 * @deprecated since 5.3.0, use {@link #delete(DeleteQuery, Class)}
 	 */
 	ByQueryResponse delete(Query query, Class<?> clazz);
 
+	/**
+	 * Delete all records matching the query.
+	 *
+	 * @param query query defining the objects
+	 * @param clazz The entity class must be annotated with
+	 *              {@link org.springframework.data.elasticsearch.annotations.Document}
+	 * @return response with detailed information
+	 * @since 5.3
+	 */
+	ByQueryResponse delete(DeleteQuery query, Class<?> clazz);
+
 	/**
 	 * Delete all records matching the query.
 	 *
@@ -290,9 +303,22 @@ default void bulkUpdate(List<UpdateQuery> queries, IndexCoordinates index) {
 	 *          {@link org.springframework.data.elasticsearch.annotations.Document}
 	 * @param index the index from which to delete
 	 * @return response with detailed information
+	 * @deprecated since 5.3.0, use {@link #delete(DeleteQuery, Class, IndexCoordinates)}
 	 */
 	ByQueryResponse delete(Query query, Class<?> clazz, IndexCoordinates index);
 
+	/**
+	 * Delete all records matching the query.
+	 *
+	 * @param query query defining the objects
+	 * @param clazz The entity class must be annotated with
+	 *              {@link org.springframework.data.elasticsearch.annotations.Document}
+	 * @param index the index from which to delete
+	 * @return response with detailed information
+	 * @since 5.3
+	 */
+	ByQueryResponse delete(DeleteQuery query, Class<?> clazz, IndexCoordinates index);
+
 	/**
 	 * Partially update a document by the given entity.
 	 *

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.DeleteQuery;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -331,19 +332,42 @@ default Mono<Void> bulkUpdate(List<UpdateQuery> queries, IndexCoordinates index)
 	 * @param query must not be {@literal null}.
 	 * @param entityType must not be {@literal null}.
 	 * @return a {@link Mono} emitting the number of the removed documents.
+	 * @deprecated since 5.3.0, use {@link #delete(DeleteQuery, Class)}
 	 */
 	Mono<ByQueryResponse> delete(Query query, Class<?> entityType);
 
+	/**
+	 * Delete the documents matching the given {@link Query} extracting index from entity metadata.
+	 *
+	 * @param query must not be {@literal null}.
+	 * @param entityType must not be {@literal null}.
+	 * @return a {@link Mono} emitting the number of the removed documents.
+	 * @since 5.3
+	 */
+	Mono<ByQueryResponse> delete(DeleteQuery query, Class<?> entityType);
+
 	/**
 	 * Delete the documents matching the given {@link Query} extracting index from entity metadata.
 	 *
 	 * @param query must not be {@literal null}.
 	 * @param entityType must not be {@literal null}.
 	 * @param index the target index, must not be {@literal null}
 	 * @return a {@link Mono} emitting the number of the removed documents.
+	 * @deprecated since 5.3.0, use {@link #delete(DeleteQuery, Class, IndexCoordinates)}
 	 */
 	Mono<ByQueryResponse> delete(Query query, Class<?> entityType, IndexCoordinates index);
 
+	/**
+	 * Delete the documents matching the given {@link Query} extracting index from entity metadata.
+	 *
+	 * @param query must not be {@literal null}.
+	 * @param entityType must not be {@literal null}.
+	 * @param index the target index, must not be {@literal null}
+	 * @return a {@link Mono} emitting the number of the removed documents.
+	 * @since 5.3
+	 */
+	Mono<ByQueryResponse> delete(DeleteQuery query, Class<?> entityType, IndexCoordinates index);
+
 	/**
 	 * Partial update of the document.
 	 *