Skip to content

Commit 496b8d6

Browse files
youssef3wiyoussef3wi
authored
Support Delete by query with es parameters.
Original Pull Request #2875 Closes #2865
1 parent d2b3ba9 commit 496b8d6

File tree

12 files changed

+1009
-0
lines changed

12 files changed

+1009
-0
lines changed

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

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@
5454
import org.springframework.data.elasticsearch.core.query.BaseQueryBuilder;
5555
import org.springframework.data.elasticsearch.core.query.BulkOptions;
5656
import org.springframework.data.elasticsearch.core.query.ByQueryResponse;
57+
import org.springframework.data.elasticsearch.core.query.DeleteQuery;
5758
import org.springframework.data.elasticsearch.core.query.IndexQuery;
5859
import org.springframework.data.elasticsearch.core.query.MoreLikeThisQuery;
5960
import org.springframework.data.elasticsearch.core.query.Query;
@@ -177,6 +178,11 @@ public void bulkUpdate(List<UpdateQuery> queries, BulkOptions bulkOptions, Index
177178
doBulkOperation(queries, bulkOptions, index);
178179
}
179180

181+
@Override
182+
public ByQueryResponse delete(DeleteQuery query, Class<?> clazz) {
183+
return delete(query, clazz, getIndexCoordinatesFor(clazz));
184+
}
185+
180186
@Override
181187
public ByQueryResponse delete(Query query, Class<?> clazz, IndexCoordinates index) {
182188

@@ -190,6 +196,18 @@ public ByQueryResponse delete(Query query, Class<?> clazz, IndexCoordinates inde
190196
return responseConverter.byQueryResponse(response);
191197
}
192198

199+
@Override
200+
public ByQueryResponse delete(DeleteQuery query, Class<?> clazz, IndexCoordinates index) {
201+
Assert.notNull(query, "query must not be null");
202+
203+
DeleteByQueryRequest request = requestConverter.documentDeleteByQueryRequest(query, routingResolver.getRouting(),
204+
clazz, index, getRefreshPolicy());
205+
206+
DeleteByQueryResponse response = execute(client -> client.deleteByQuery(request));
207+
208+
return responseConverter.byQueryResponse(response);
209+
}
210+
193211
@Override
194212
public UpdateResponse update(UpdateQuery updateQuery, IndexCoordinates index) {
195213

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import co.elastic.clients.json.JsonpMapper;
2626
import co.elastic.clients.transport.Version;
2727
import co.elastic.clients.transport.endpoints.BooleanResponse;
28+
import org.springframework.data.elasticsearch.core.query.DeleteQuery;
2829
import reactor.core.publisher.Flux;
2930
import reactor.core.publisher.Mono;
3031
import reactor.util.function.Tuple2;
@@ -180,6 +181,15 @@ public Mono<ByQueryResponse> delete(Query query, Class<?> entityType, IndexCoord
180181
return Mono.from(execute(client -> client.deleteByQuery(request))).map(responseConverter::byQueryResponse);
181182
}
182183

184+
@Override
185+
public Mono<ByQueryResponse> delete(DeleteQuery query, Class<?> entityType, IndexCoordinates index) {
186+
Assert.notNull(query, "query must not be null");
187+
188+
DeleteByQueryRequest request = requestConverter.documentDeleteByQueryRequest(query, routingResolver.getRouting(),
189+
entityType, index, getRefreshPolicy());
190+
return Mono.from(execute(client -> client.deleteByQuery(request))).map(responseConverter::byQueryResponse);
191+
}
192+
183193
@Override
184194
public <T> Mono<T> get(String id, Class<T> entityType, IndexCoordinates index) {
185195

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

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -968,6 +968,79 @@ public DeleteByQueryRequest documentDeleteByQueryRequest(Query query, @Nullable
968968
});
969969
}
970970

971+
public DeleteByQueryRequest documentDeleteByQueryRequest(DeleteQuery query, @Nullable String routing, Class<?> clazz,
972+
IndexCoordinates index, @Nullable RefreshPolicy refreshPolicy) {
973+
Assert.notNull(query, "query must not be null");
974+
Assert.notNull(index, "index must not be null");
975+
976+
return DeleteByQueryRequest.of(dqb -> {
977+
dqb.index(Arrays.asList(index.getIndexNames())) //
978+
.query(getQuery(query.getQuery(), clazz))//
979+
.refresh(deleteByQueryRefresh(refreshPolicy))
980+
.requestsPerSecond(query.getRequestsPerSecond())
981+
.maxDocs(query.getMaxDocs())
982+
.scroll(time(query.getScroll()))
983+
.scrollSize(query.getScrollSize());
984+
985+
if (query.getRouting() != null) {
986+
dqb.routing(query.getRouting());
987+
} else if (StringUtils.hasText(routing)) {
988+
dqb.routing(routing);
989+
}
990+
991+
if (query.getQ() != null) {
992+
dqb.q(query.getQ())
993+
.analyzer(query.getAnalyzer())
994+
.analyzeWildcard(query.getAnalyzeWildcard())
995+
.defaultOperator(operator(query.getDefaultOperator()))
996+
.df(query.getDf())
997+
.lenient(query.getLenient());
998+
}
999+
1000+
if (query.getExpandWildcards() != null && !query.getExpandWildcards().isEmpty()) {
1001+
dqb.expandWildcards(expandWildcards(query.getExpandWildcards()));
1002+
}
1003+
if (query.getStats() != null && !query.getStats().isEmpty()) {
1004+
dqb.stats(query.getStats());
1005+
}
1006+
if (query.getSlices() != null) {
1007+
dqb.slices(sb -> sb.value(query.getSlices()));
1008+
}
1009+
if (query.getSort() != null) {
1010+
ElasticsearchPersistentEntity<?> persistentEntity = getPersistentEntity(clazz);
1011+
List<SortOptions> sortOptions = getSortOptions(query.getSort(), persistentEntity);
1012+
1013+
if (!sortOptions.isEmpty()) {
1014+
dqb.sort(
1015+
sortOptions.stream()
1016+
.map(sortOption -> {
1017+
String order = "asc";
1018+
var sortField = sortOption.field();
1019+
if (sortField.order() != null) {
1020+
order = sortField.order().jsonValue();
1021+
}
1022+
1023+
return sortField.field() + ":" + order;
1024+
})
1025+
.collect(Collectors.toList())
1026+
);
1027+
}
1028+
}
1029+
dqb.allowNoIndices(query.getAllowNoIndices())
1030+
.conflicts(conflicts(query.getConflicts()))
1031+
.ignoreUnavailable(query.getIgnoreUnavailable())
1032+
.preference(query.getPreference())
1033+
.requestCache(query.getRequestCache())
1034+
.searchType(searchType(query.getSearchType()))
1035+
.searchTimeout(time(query.getSearchTimeout()))
1036+
.terminateAfter(query.getTerminateAfter())
1037+
.timeout(time(query.getTimeout()))
1038+
.version(query.getVersion());
1039+
1040+
return dqb;
1041+
});
1042+
}
1043+
9711044
public UpdateRequest<Document, ?> documentUpdateRequest(UpdateQuery query, IndexCoordinates index,
9721045
@Nullable RefreshPolicy refreshPolicy, @Nullable String routing) {
9731046

‎src/main/java/org/springframework/data/elasticsearch/client/elc/TypeUtils.java

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import co.elastic.clients.elasticsearch._types.*;
1919
import co.elastic.clients.elasticsearch._types.mapping.FieldType;
2020
import co.elastic.clients.elasticsearch._types.mapping.TypeMapping;
21+
import co.elastic.clients.elasticsearch._types.query_dsl.Operator;
2122
import co.elastic.clients.elasticsearch.core.search.BoundaryScanner;
2223
import co.elastic.clients.elasticsearch.core.search.HighlighterEncoder;
2324
import co.elastic.clients.elasticsearch.core.search.HighlighterFragmenter;
@@ -46,6 +47,8 @@
4647
import org.springframework.data.elasticsearch.core.query.Query;
4748
import org.springframework.data.elasticsearch.core.query.RescorerQuery;
4849
import org.springframework.data.elasticsearch.core.query.UpdateResponse;
50+
import org.springframework.data.elasticsearch.core.query.types.ConflictsType;
51+
import org.springframework.data.elasticsearch.core.query.types.OperatorType;
4952
import org.springframework.data.elasticsearch.core.reindex.ReindexRequest;
5053
import org.springframework.lang.Nullable;
5154
import org.springframework.util.Assert;
@@ -500,4 +503,26 @@ static Map<String, JsonData> paramsMap(Map<String, Object> params) {
500503
});
501504
return mappedParams;
502505
}
506+
507+
/**
508+
* Convert a spring-data-elasticsearch operator to an Elasticsearch operator.
509+
*
510+
* @param operator spring-data-elasticsearch operator.
511+
* @return an Elasticsearch Operator.
512+
*/
513+
@Nullable
514+
static Operator operator(@Nullable OperatorType operator) {
515+
return operator != null ? Operator.valueOf(operator.name()) : null;
516+
}
517+
518+
/**
519+
* Convert a spring-data-elasticsearch {@literal conflicts} to an Elasticsearch {@literal conflicts}.
520+
*
521+
* @param conflicts spring-data-elasticsearch {@literal conflicts}.
522+
* @return an Elasticsearch {@literal conflicts}.
523+
*/
524+
@Nullable
525+
static Conflicts conflicts(@Nullable ConflictsType conflicts) {
526+
return conflicts != null ? Conflicts.valueOf(conflicts.name()) : null;
527+
}
503528
}

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
*/
1616
package org.springframework.data.elasticsearch.core;
1717

18+
import org.springframework.data.elasticsearch.core.query.DeleteQuery;
1819
import reactor.core.publisher.Flux;
1920
import reactor.core.publisher.Mono;
2021
import reactor.core.publisher.Sinks;
@@ -411,6 +412,11 @@ public Mono<String> delete(String id, IndexCoordinates index) {
411412
public Mono<ByQueryResponse> delete(Query query, Class<?> entityType) {
412413
return delete(query, entityType, getIndexCoordinatesFor(entityType));
413414
}
415+
416+
@Override
417+
public Mono<ByQueryResponse> delete(DeleteQuery query, Class<?> entityType) {
418+
return delete(query, entityType, getIndexCoordinatesFor(entityType));
419+
}
414420
// endregion
415421

416422
// region SearchDocument

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

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
2222
import org.springframework.data.elasticsearch.core.query.BulkOptions;
2323
import org.springframework.data.elasticsearch.core.query.ByQueryResponse;
24+
import org.springframework.data.elasticsearch.core.query.DeleteQuery;
2425
import org.springframework.data.elasticsearch.core.query.IndexQuery;
2526
import org.springframework.data.elasticsearch.core.query.Query;
2627
import org.springframework.data.elasticsearch.core.query.UpdateQuery;
@@ -279,9 +280,21 @@ default void bulkUpdate(List<UpdateQuery> queries, IndexCoordinates index) {
279280
* {@link org.springframework.data.elasticsearch.annotations.Document}
280281
* @return response with detailed information
281282
* @since 4.1
283+
* @deprecated since 5.3.0, use {@link #delete(DeleteQuery, Class)}
282284
*/
283285
ByQueryResponse delete(Query query, Class<?> clazz);
284286

287+
/**
288+
* Delete all records matching the query.
289+
*
290+
* @param query query defining the objects
291+
* @param clazz The entity class must be annotated with
292+
* {@link org.springframework.data.elasticsearch.annotations.Document}
293+
* @return response with detailed information
294+
* @since 5.3
295+
*/
296+
ByQueryResponse delete(DeleteQuery query, Class<?> clazz);
297+
285298
/**
286299
* Delete all records matching the query.
287300
*
@@ -290,9 +303,22 @@ default void bulkUpdate(List<UpdateQuery> queries, IndexCoordinates index) {
290303
* {@link org.springframework.data.elasticsearch.annotations.Document}
291304
* @param index the index from which to delete
292305
* @return response with detailed information
306+
* @deprecated since 5.3.0, use {@link #delete(DeleteQuery, Class, IndexCoordinates)}
293307
*/
294308
ByQueryResponse delete(Query query, Class<?> clazz, IndexCoordinates index);
295309

310+
/**
311+
* Delete all records matching the query.
312+
*
313+
* @param query query defining the objects
314+
* @param clazz The entity class must be annotated with
315+
* {@link org.springframework.data.elasticsearch.annotations.Document}
316+
* @param index the index from which to delete
317+
* @return response with detailed information
318+
* @since 5.3
319+
*/
320+
ByQueryResponse delete(DeleteQuery query, Class<?> clazz, IndexCoordinates index);
321+
296322
/**
297323
* Partially update a document by the given entity.
298324
*

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

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
*/
1616
package org.springframework.data.elasticsearch.core;
1717

18+
import org.springframework.data.elasticsearch.core.query.DeleteQuery;
1819
import reactor.core.publisher.Flux;
1920
import reactor.core.publisher.Mono;
2021

@@ -331,19 +332,42 @@ default Mono<Void> bulkUpdate(List<UpdateQuery> queries, IndexCoordinates index)
331332
* @param query must not be {@literal null}.
332333
* @param entityType must not be {@literal null}.
333334
* @return a {@link Mono} emitting the number of the removed documents.
335+
* @deprecated since 5.3.0, use {@link #delete(DeleteQuery, Class)}
334336
*/
335337
Mono<ByQueryResponse> delete(Query query, Class<?> entityType);
336338

339+
/**
340+
* Delete the documents matching the given {@link Query} extracting index from entity metadata.
341+
*
342+
* @param query must not be {@literal null}.
343+
* @param entityType must not be {@literal null}.
344+
* @return a {@link Mono} emitting the number of the removed documents.
345+
* @since 5.3
346+
*/
347+
Mono<ByQueryResponse> delete(DeleteQuery query, Class<?> entityType);
348+
337349
/**
338350
* Delete the documents matching the given {@link Query} extracting index from entity metadata.
339351
*
340352
* @param query must not be {@literal null}.
341353
* @param entityType must not be {@literal null}.
342354
* @param index the target index, must not be {@literal null}
343355
* @return a {@link Mono} emitting the number of the removed documents.
356+
* @deprecated since 5.3.0, use {@link #delete(DeleteQuery, Class, IndexCoordinates)}
344357
*/
345358
Mono<ByQueryResponse> delete(Query query, Class<?> entityType, IndexCoordinates index);
346359

360+
/**
361+
* Delete the documents matching the given {@link Query} extracting index from entity metadata.
362+
*
363+
* @param query must not be {@literal null}.
364+
* @param entityType must not be {@literal null}.
365+
* @param index the target index, must not be {@literal null}
366+
* @return a {@link Mono} emitting the number of the removed documents.
367+
* @since 5.3
368+
*/
369+
Mono<ByQueryResponse> delete(DeleteQuery query, Class<?> entityType, IndexCoordinates index);
370+
347371
/**
348372
* Partial update of the document.
349373
*

0 commit comments

Comments
 (0)