Provide new method to return SearchHits in ReactiveElasticsearchClient [DATAES-796]

[spring-projects-issues]

Dmitriy Sulimchuk opened DATAES-796 and commented

It is useful to get full information with the response (for example to get total hits).

Currently, I have to make 2 requests (1 to get data and another for total)

 

org.springframework.data.elasticsearch.client.reactive.DefaultReactiveElasticsearchClient#search

 

@Override
public Flux<SearchHit> search(HttpHeaders headers, SearchRequest searchRequest) {

   return sendRequest(searchRequest, RequestCreator.search(), SearchResponse.class, headers) //
         .map(SearchResponse::getHits)
         .flatMap(Flux::fromIterable); //here we lose a lot of usefull information
}

 

 

I think we should make another method with this signature:

 

Mono<SearchHits> searchHits(HttpHeaders headers, SearchRequest searchRequest)

p.s. I'm not sure about method name;

 

---

Affects: 3.2.6 (Moore SR6)

Attachments:

• image-2020-04-23-23-36-04-848.png (56.87 kB)
• image-2020-04-23-23-36-56-802.png (57.09 kB)

Issue Links:

• DATAES-696 reactiveElasticsearchClient search can't get total match
  ("duplicates")
• DATAES-879 Retrieve SearchResponse using ReactiveElasticsearchTemplate
  ("is duplicated by")

Referenced from: pull request #540

3 votes, 6 watchers

[spring-projects-issues]

sothawo commented

It does not really make sense in a reactive flow to return a Mono of an object that contains a whole list of objects. What would you do with the SearchHit objects that are contained in that?

The count methods from the ReactiveElasticsearcOperations interface don't retrieve the data itself from Elasticsearch. only the count

[spring-projects-issues]

Dmitriy Sulimchuk commented

I'd like to return just page (for example first 20 items) with information about total elements (to show available page links on UI)

 

class Page<T>  {
  long total;
  List<T> page; 
)

Currently, I have to make 2 requests, but in reality, this data comes in the first call.

 

 

2020-04-23 23:27:39.951 TRACE 12900 --- [ctor-http-nio-3] o.s.data.elasticsearch.client.WIRE       : [5b1adc63] Sending request POST /person12/_search with parameters: {typed_keys=true, ignore_unavailable=false, expand_wildcards=open, allow_no_indices=true, search_type=dfs_query_then_fetch, batched_reduce_size=512}
Request body: {"from":0,"size":5,"query":{"wrapper":{"query":"ewogICJtYXRjaF9hbGwiIDogewogICAgImJvb3N0IiA6IDEuMAogIH0KfQ=="}},"version":false}

2020-04-23 23:27:39.999 TRACE 12900 --- [ctor-http-nio-3] o.s.data.elasticsearch.client.WIRE       : [5b1adc63] Received response: 200 OK
Response body: {"took":1,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":9,"relation":"eq"},"max_score":1.0,"hits":[{"_index":"person12","_type":"person","_id":"78a71368-ade2-454c-97ee-33f14bfe8df8","_score":1.0,"_source":{"id":"78a71368-ade2-454c-97ee-33f14bfe8df8","firstname":"FirstNam1","lastname":"LastName1","veryLongString":"dskfjsalkjf","d":"20200423T190136.062+0000"}},{"_index":"person12","_type":"person","_id":"acaf553f-27f8-4280-ace6-7f7004290659","_score":1.0,"_source":{"id":"acaf553f-27f8-4280-ace6-7f7004290659","firstname":"FirstNam1","lastname":"LastName1","veryLongString":"dskfjsalkjf","d":"20200423T190236.820+0000"}},{"_index":"person12","_type":"person","_id":"c1cf020f-8b31-4a7c-ba7b-279e3a29a232","_score":1.0,"_source":{"id":"c1cf020f-8b31-4a7c-ba7b-279e3a29a232","firstname":"FirstNam2","lastname":"LastName2","veryLongString":"dskfjsalkjf","d":"20200423T190238.422+0000"}},{"_index":"person12","_type":"person","_id":"e50052db-8bbe-4e8e-96bf-ff087c6b2a75","_score":1.0,"_source":{"id":"e50052db-8bbe-4e8e-96bf-ff087c6b2a75","firstname":"FirstNam1","lastname":"LastName1","veryLongString":"dskfjsalkjf","d":"20200423T191524.474+0000"}},{"_index":"person12","_type":"person","_id":"dcd7c945-e229-4a3c-a6d6-3e4fee6d082c","_score":1.0,"_source":{"id":"dcd7c945-e229-4a3c-a6d6-3e4fee6d082c","firstname":"FirstNam2","lastname":"LastName2","veryLongString":"dskfjsalkjf","d":"20200423T191525.913+0000"}}]}}

Here in logs, you can see that the elastic search response already contains this data. 

!image-2020-04-23-23-36-56-802.png!

 

I think that for my use case (get just 1 page) Flux doesnt's add any value

[spring-projects-issues]

sothawo commented

If you want to return a page with the entities in it, why do you use the reactive setup here to access Elasticsearch? And how do you count the data for your query?

[spring-projects-issues]

Dmitriy Sulimchuk commented

If you want to return a page with the entities in it, why do you use the reactive setup here to access Elasticsearch

For scrolling results it's effective and convenient to use reactive stack, and not just wrapping Java High Level REST Client into Mono.

 And how do you count the data for your query?

I don't understand the question. As I showed elastic search response return hits count, and it is enough for me.

 

As I understood you're proposing me to use different implementation for different use cases inside one application?

 

[spring-projects-issues]

gitterhh commented

In ReactiveElasticsearchTemplate class there is already the method which receive a SearchRequest object

protected Flux<SearchDocument> doFind(SearchRequest request) {

 if (QUERY_LOGGER.isDebugEnabled()) {
 QUERY_LOGGER.debug("Executing doFind: {}", request);
 }

 return Flux.from(execute(client -> client.search(request))).map(DocumentAdapters::from) //
 .onErrorResume(NoSuchIndexException.class, it -> Mono.empty());
}

Why not to expuse it as public?

[spring-projects-issues]

gitterhh commented

It will be nice to have a general purpose method which can be used if some methods are missing or not implemented yet

public Flux<SearchResponse> search(SearchRequest) {

    // ...

}

[spring-projects-issues]

sothawo commented

why expose this method? This would mean to pull Elasticsearch classes into out API which we don't. And how would this help with regard to have a method that returns a Mono<SearchHits>?

[spring-projects-issues]

gitterhh commented

It will be possible to retrieve Mono<SearchHits> from Fux<SearchResponse>.

Well, you can have own SearchRequest and SearchResponse classes

[spring-projects-issues]

sothawo commented

??It will be possible to retrieve Mono<SearchHits> from Fux<SearchResponse>.??

 

The method you refer to returns a Flux<SearchDocument>. A SearchDocument is a single search hit. A SearchHits as known from the non-reactive part contains information like aggregations and maximum score, this is not available in a Flux<SearchDocument>

[spring-projects-issues]

sothawo commented

If you want to send custom requests to the WebClient and parse the response, there already is the method <T> Publisher<T> ReactiveElasticsearchOperationsReactiveElasticsearchOperations.execute(ClientCallback<Publisher<T>> callback);<T> Publisher<T> execute(ClientCallback<Publisher<T>> callback);

That's the same method that is called in the doFInd() method

[spring-projects-issues]

Takaaki Iida commented

It does not really make sense in a reactive flow to return a Mono of an object that contains a whole list of objects.

I understood that this suggestion does not follow reactive way.

If you want to return a page with the entities in it, why do you use the reactive setup here to access Elasticsearch? And how do you count the data for your query?

For my case, simply I want to get merits of non-blocking that's way I want to use ReactiveElasticsearchTemplate.

If ReactiveElasticsearchTemplate returns SearchResponse, I can get count from SearchResponse.hits.totalHits.

 

ReactiveElasticsearchOperations.execute(ClientCallback<Publisher<T>> callback);

Hmm, it seems I need to implement much code to invoke ES by WebClient.

Code might be something like below (This is Kotlin) then probably I will use RestHighLevelClient..

operations.execute { client: ReactiveElasticsearchClient ->
    client.execute { webClient: WebClient ->
        webClient.post()
            .uri(...)
            .body(...)
            .exchange()
    }
}

 
I thought "showing search results with total counts" is a common requirements in real world and I want to achieve it based on non-blocking but now I will use RestHighLevelClient..

[spring-projects-issues]

rgmz commented

+1 to this.

If you want to return a page with the entities in it, why do you use the reactive setup here to access Elasticsearch?

Pagination is a common use case in the real world, and it would be nice to have something equivalent in the reactive setup.

Having non-blocking code is desirable, however, ReactiveX isn't always the best way to express your logic. For example, Kotlin Coroutines are quite popular and allow you to write non-blocking code imperatively. 

 

[spring-projects-issues]

sothawo commented

Method will be Mono<SearchPage<T>> ReactiveSearchOperations.searchForPage(Query, java.lang.Class<T>), plus variants for the different parameters