Skip to content

Commit be34ff8

Browse files
hobbutxhaggi
hobbut
authored and
xhaggi
committed
DATAES-459 - Return ScrolledPage instead of Page while using startScroll and continueScroll methods.
Original pull request: #274
1 parent 9abe0f2 commit be34ff8

File tree

5 files changed

+63
-62
lines changed

5 files changed

+63
-62
lines changed

‎src/main/asciidoc/reference/elasticsearch-misc.adoc

+4-4
Original file line numberDiff line numberDiff line change
@@ -40,13 +40,13 @@ SearchQuery searchQuery = new NativeSearchQueryBuilder()
4040
.withPageable(PageRequest.of(0, 10))
4141
.build();
4242
43-
Page<SampleEntity> scroll = elasticsearchTemplate.startScroll(1000, searchQuery, SampleEntity.class);
43+
ScrolledPage<SampleEntity> scroll = elasticsearchTemplate.startScroll(1000, searchQuery, SampleEntity.class);
4444
45-
String scrollId = ((ScrolledPage) scroll).getScrollId();
45+
String scrollId = scroll.getScrollId();
4646
List<SampleEntity> sampleEntities = new ArrayList<>();
4747
while (scroll.hasContent()) {
4848
sampleEntities.addAll(scroll.getContent());
49-
scrollId = ((ScrolledPage) scroll).getScrollId();
49+
scrollId = scroll.getScrollId();
5050
scroll = elasticsearchTemplate.continueScroll(scrollId, 1000, SampleEntity.class);
5151
}
5252
elasticsearchTemplate.clearScroll(scrollId);
@@ -74,4 +74,4 @@ while (stream.hasNext()) {
7474
sampleEntities.add(stream.next());
7575
}
7676
----
77-
====
77+
====

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

+7-6
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
* @author Mohsin Husen
3737
* @author Kevin Leturc
3838
* @author Zetang Zeng
39+
* @author Dmitriy Yakovlev
3940
*/
4041
public interface ElasticsearchOperations {
4142

@@ -544,7 +545,7 @@ default List<List<?>> queryForList(List<SearchQuery> queries, List<Class<?>> cla
544545
* @param clazz The class of entity to retrieve.
545546
* @return The scan id for input query.
546547
*/
547-
<T> Page<T> startScroll(long scrollTimeInMillis, SearchQuery query, Class<T> clazz);
548+
<T> ScrolledPage<T> startScroll(long scrollTimeInMillis, SearchQuery query, Class<T> clazz);
548549

549550
/**
550551
* Returns scrolled page for given query
@@ -555,7 +556,7 @@ default List<List<?>> queryForList(List<SearchQuery> queries, List<Class<?>> cla
555556
* @param mapper Custom impl to map result to entities
556557
* @return The scan id for input query.
557558
*/
558-
<T> Page<T> startScroll(long scrollTimeInMillis, SearchQuery query, Class<T> clazz, SearchResultMapper mapper);
559+
<T> ScrolledPage<T> startScroll(long scrollTimeInMillis, SearchQuery query, Class<T> clazz, SearchResultMapper mapper);
559560

560561
/**
561562
* Returns scrolled page for given query
@@ -566,7 +567,7 @@ default List<List<?>> queryForList(List<SearchQuery> queries, List<Class<?>> cla
566567
* @param clazz The class of entity to retrieve.
567568
* @return The scan id for input query.
568569
*/
569-
<T> Page<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz);
570+
<T> ScrolledPage<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz);
570571

571572
/**
572573
* Returns scrolled page for given query
@@ -577,11 +578,11 @@ default List<List<?>> queryForList(List<SearchQuery> queries, List<Class<?>> cla
577578
* @param mapper Custom impl to map result to entities
578579
* @return The scan id for input query.
579580
*/
580-
<T> Page<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz, SearchResultMapper mapper);
581+
<T> ScrolledPage<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz, SearchResultMapper mapper);
581582

582583

583-
<T> Page<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz);
584-
<T> Page<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz, SearchResultMapper mapper);
584+
<T> ScrolledPage<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz);
585+
<T> ScrolledPage<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz, SearchResultMapper mapper);
585586
/**
586587
* Clears the search contexts associated with specified scroll ids.
587588
*

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

+12-12
Original file line numberDiff line numberDiff line change
@@ -496,7 +496,7 @@ public <T> Page<T> queryForPage(StringQuery query, Class<T> clazz, SearchResultM
496496
@Override
497497
public <T> CloseableIterator<T> stream(CriteriaQuery query, Class<T> clazz) {
498498
final long scrollTimeInMillis = TimeValue.timeValueMinutes(1).millis();
499-
return doStream(scrollTimeInMillis, (ScrolledPage<T>) startScroll(scrollTimeInMillis, query, clazz), clazz,
499+
return doStream(scrollTimeInMillis, startScroll(scrollTimeInMillis, query, clazz), clazz,
500500
resultsMapper);
501501
}
502502

@@ -508,7 +508,7 @@ public <T> CloseableIterator<T> stream(SearchQuery query, Class<T> clazz) {
508508
@Override
509509
public <T> CloseableIterator<T> stream(SearchQuery query, final Class<T> clazz, final SearchResultMapper mapper) {
510510
final long scrollTimeInMillis = TimeValue.timeValueMinutes(1).millis();
511-
return doStream(scrollTimeInMillis, (ScrolledPage<T>) startScroll(scrollTimeInMillis, query, clazz, mapper), clazz,
511+
return doStream(scrollTimeInMillis, startScroll(scrollTimeInMillis, query, clazz, mapper), clazz,
512512
mapper);
513513
}
514514

@@ -547,7 +547,7 @@ public boolean hasNext() {
547547
// Test if it remains hits
548548
if (currentHits == null || !currentHits.hasNext()) {
549549
// Do a new request
550-
final ScrolledPage<T> scroll = (ScrolledPage<T>) continueScroll(scrollId, scrollTimeInMillis, clazz, mapper);
550+
final ScrolledPage<T> scroll = continueScroll(scrollId, scrollTimeInMillis, clazz, mapper);
551551
// Save hits and scroll id
552552
currentHits = scroll.iterator();
553553
finished = !currentHits.hasNext();
@@ -878,14 +878,14 @@ public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz,
878878
}
879879
};
880880

881-
Page<SearchHit> scrolledResult = startScroll(scrollTimeInMillis, searchQuery, SearchHit.class,
881+
ScrolledPage<SearchHit> scrolledResult = startScroll(scrollTimeInMillis, searchQuery, SearchHit.class,
882882
deleteEntryResultMapper);
883883
BulkRequest request = new BulkRequest();
884884
List<SearchHit> documentsToDelete = new ArrayList<>();
885885

886886
do {
887887
documentsToDelete.addAll(scrolledResult.getContent());
888-
scrolledResult = continueScroll(((ScrolledPage<T>) scrolledResult).getScrollId(), scrollTimeInMillis,
888+
scrolledResult = continueScroll(scrolledResult.getScrollId(), scrollTimeInMillis,
889889
SearchHit.class, deleteEntryResultMapper);
890890
} while (scrolledResult.getContent().size() != 0);
891891

@@ -903,7 +903,7 @@ public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz,
903903
}
904904
}
905905

906-
clearScroll(((ScrolledPage<T>) scrolledResult).getScrollId());
906+
clearScroll(scrolledResult.getScrollId());
907907
}
908908

909909
@Override
@@ -1010,29 +1010,29 @@ private SearchResponse doScroll(SearchRequest request, SearchQuery searchQuery)
10101010
}
10111011
}
10121012

1013-
public <T> Page<T> startScroll(long scrollTimeInMillis, SearchQuery searchQuery, Class<T> clazz) {
1013+
public <T> ScrolledPage<T> startScroll(long scrollTimeInMillis, SearchQuery searchQuery, Class<T> clazz) {
10141014
SearchResponse response = doScroll(prepareScroll(searchQuery, scrollTimeInMillis, clazz), searchQuery);
10151015
return resultsMapper.mapResults(response, clazz, null);
10161016
}
10171017

1018-
public <T> Page<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz) {
1018+
public <T> ScrolledPage<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz) {
10191019
SearchResponse response = doScroll(prepareScroll(criteriaQuery, scrollTimeInMillis, clazz), criteriaQuery);
10201020
return resultsMapper.mapResults(response, clazz, null);
10211021
}
10221022

1023-
public <T> Page<T> startScroll(long scrollTimeInMillis, SearchQuery searchQuery, Class<T> clazz,
1023+
public <T> ScrolledPage<T> startScroll(long scrollTimeInMillis, SearchQuery searchQuery, Class<T> clazz,
10241024
SearchResultMapper mapper) {
10251025
SearchResponse response = doScroll(prepareScroll(searchQuery, scrollTimeInMillis, clazz), searchQuery);
10261026
return mapper.mapResults(response, clazz, null);
10271027
}
10281028

1029-
public <T> Page<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz,
1029+
public <T> ScrolledPage<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz,
10301030
SearchResultMapper mapper) {
10311031
SearchResponse response = doScroll(prepareScroll(criteriaQuery, scrollTimeInMillis, clazz), criteriaQuery);
10321032
return mapper.mapResults(response, clazz, null);
10331033
}
10341034

1035-
public <T> Page<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz) {
1035+
public <T> ScrolledPage<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz) {
10361036
SearchScrollRequest request = new SearchScrollRequest(scrollId);
10371037
request.scroll(TimeValue.timeValueMillis(scrollTimeInMillis));
10381038
SearchResponse response;
@@ -1044,7 +1044,7 @@ public <T> Page<T> continueScroll(@Nullable String scrollId, long scrollTimeInMi
10441044
return resultsMapper.mapResults(response, clazz, Pageable.unpaged());
10451045
}
10461046

1047-
public <T> Page<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz,
1047+
public <T> ScrolledPage<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz,
10481048
SearchResultMapper mapper) {
10491049
SearchScrollRequest request = new SearchScrollRequest(scrollId);
10501050
request.scroll(TimeValue.timeValueMillis(scrollTimeInMillis));

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

+12-12
Original file line numberDiff line numberDiff line change
@@ -437,7 +437,7 @@ public <T> Page<T> queryForPage(StringQuery query, Class<T> clazz, SearchResultM
437437
@Override
438438
public <T> CloseableIterator<T> stream(CriteriaQuery query, Class<T> clazz) {
439439
final long scrollTimeInMillis = TimeValue.timeValueMinutes(1).millis();
440-
return doStream(scrollTimeInMillis, (ScrolledPage<T>) startScroll(scrollTimeInMillis, query, clazz), clazz,
440+
return doStream(scrollTimeInMillis, startScroll(scrollTimeInMillis, query, clazz), clazz,
441441
resultsMapper);
442442
}
443443

@@ -449,7 +449,7 @@ public <T> CloseableIterator<T> stream(SearchQuery query, Class<T> clazz) {
449449
@Override
450450
public <T> CloseableIterator<T> stream(SearchQuery query, final Class<T> clazz, final SearchResultMapper mapper) {
451451
final long scrollTimeInMillis = TimeValue.timeValueMinutes(1).millis();
452-
return doStream(scrollTimeInMillis, (ScrolledPage<T>) startScroll(scrollTimeInMillis, query, clazz, mapper), clazz,
452+
return doStream(scrollTimeInMillis, startScroll(scrollTimeInMillis, query, clazz, mapper), clazz,
453453
mapper);
454454
}
455455

@@ -488,7 +488,7 @@ public boolean hasNext() {
488488
// Test if it remains hits
489489
if (currentHits == null || !currentHits.hasNext()) {
490490
// Do a new request
491-
final ScrolledPage<T> scroll = (ScrolledPage<T>) continueScroll(scrollId, scrollTimeInMillis, clazz, mapper);
491+
final ScrolledPage<T> scroll = continueScroll(scrollId, scrollTimeInMillis, clazz, mapper);
492492
// Save hits and scroll id
493493
currentHits = scroll.iterator();
494494
finished = !currentHits.hasNext();
@@ -765,14 +765,14 @@ public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz,
765765
}
766766
};
767767

768-
Page<SearchHit> scrolledResult = startScroll(scrollTimeInMillis, searchQuery, SearchHit.class,
768+
ScrolledPage<SearchHit> scrolledResult = startScroll(scrollTimeInMillis, searchQuery, SearchHit.class,
769769
deleteEntryResultMapper);
770770
BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
771771
List<SearchHit> documentsToDelete = new ArrayList<>();
772772

773773
do {
774774
documentsToDelete.addAll(scrolledResult.getContent());
775-
scrolledResult = continueScroll(((ScrolledPage<T>) scrolledResult).getScrollId(), scrollTimeInMillis,
775+
scrolledResult = continueScroll(scrolledResult.getScrollId(), scrollTimeInMillis,
776776
SearchHit.class, deleteEntryResultMapper);
777777
} while (scrolledResult.getContent().size() != 0);
778778

@@ -784,7 +784,7 @@ public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz,
784784
bulkRequestBuilder.execute().actionGet();
785785
}
786786

787-
clearScroll(((ScrolledPage<T>) scrolledResult).getScrollId());
787+
clearScroll(scrolledResult.getScrollId());
788788
}
789789

790790
@Override
@@ -873,35 +873,35 @@ private SearchResponse doScroll(SearchRequestBuilder requestBuilder, SearchQuery
873873
return getSearchResponse(requestBuilder.setQuery(searchQuery.getQuery()));
874874
}
875875

876-
public <T> Page<T> startScroll(long scrollTimeInMillis, SearchQuery searchQuery, Class<T> clazz) {
876+
public <T> ScrolledPage<T> startScroll(long scrollTimeInMillis, SearchQuery searchQuery, Class<T> clazz) {
877877
SearchResponse response = doScroll(prepareScroll(searchQuery, scrollTimeInMillis, clazz), searchQuery);
878878
return resultsMapper.mapResults(response, clazz, null);
879879
}
880880

881-
public <T> Page<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz) {
881+
public <T> ScrolledPage<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz) {
882882
SearchResponse response = doScroll(prepareScroll(criteriaQuery, scrollTimeInMillis, clazz), criteriaQuery);
883883
return resultsMapper.mapResults(response, clazz, null);
884884
}
885885

886-
public <T> Page<T> startScroll(long scrollTimeInMillis, SearchQuery searchQuery, Class<T> clazz,
886+
public <T> ScrolledPage<T> startScroll(long scrollTimeInMillis, SearchQuery searchQuery, Class<T> clazz,
887887
SearchResultMapper mapper) {
888888
SearchResponse response = doScroll(prepareScroll(searchQuery, scrollTimeInMillis, clazz), searchQuery);
889889
return mapper.mapResults(response, clazz, null);
890890
}
891891

892-
public <T> Page<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz,
892+
public <T> ScrolledPage<T> startScroll(long scrollTimeInMillis, CriteriaQuery criteriaQuery, Class<T> clazz,
893893
SearchResultMapper mapper) {
894894
SearchResponse response = doScroll(prepareScroll(criteriaQuery, scrollTimeInMillis, clazz), criteriaQuery);
895895
return mapper.mapResults(response, clazz, null);
896896
}
897897

898-
public <T> Page<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz) {
898+
public <T> ScrolledPage<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz) {
899899
SearchResponse response = getSearchResponse(
900900
client.prepareSearchScroll(scrollId).setScroll(TimeValue.timeValueMillis(scrollTimeInMillis)).execute());
901901
return resultsMapper.mapResults(response, clazz, Pageable.unpaged());
902902
}
903903

904-
public <T> Page<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz,
904+
public <T> ScrolledPage<T> continueScroll(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz,
905905
SearchResultMapper mapper) {
906906
SearchResponse response = getSearchResponse(
907907
client.prepareSearchScroll(scrollId).setScroll(TimeValue.timeValueMillis(scrollTimeInMillis)).execute());

0 commit comments

Comments
 (0)