forked from opensearch-project/OpenSearch
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
7 changed files
with
282 additions
and
458 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
136 changes: 136 additions & 0 deletions
136
server/src/main/java/org/opensearch/grpc/services/search/GprcSearchAction.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,136 @@ | ||
| package org.opensearch.grpc.services.search; | ||
|
|
||
| import io.grpc.stub.StreamObserver; | ||
| import opensearch.proto.*; | ||
| import org.opensearch.action.search.SearchResponse; | ||
| import org.opensearch.action.search.ShardSearchFailure; | ||
| import org.opensearch.core.action.ActionListener; | ||
|
|
||
| import java.util.Arrays; | ||
| import java.util.List; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import static org.opensearch.rest.action.search.RestSearchAction.TOTAL_HITS_AS_INT_PARAM; | ||
|
|
||
|
|
||
| public class GprcSearchAction { | ||
|
|
||
| public static opensearch.proto.SearchResponse searchResponseToProto(org.opensearch.action.search.SearchResponse transportResponse) { | ||
| ResponseBody.Builder responseBodyBuilder = ResponseBody.newBuilder(); | ||
|
|
||
| // [required] Milliseconds it took Elasticsearch to execute the request. | ||
| // optional int64 took = 1; | ||
| responseBodyBuilder.setTook(transportResponse.getTook().getMillis()); | ||
| // | ||
| // // [required] If true, the request timed out before completion; returned results may be partial or empty. | ||
| // optional bool timed_out = 2; | ||
| responseBodyBuilder.setTimedOut(transportResponse.isTimedOut()); | ||
| // | ||
| // // [required] Contains a count of shards used for the request. | ||
| // ShardStatistics shards = 3; | ||
| responseBodyBuilder.setShards(ShardStatistics.newBuilder() | ||
| .setFailed(transportResponse.getFailedShards()) | ||
| .setSuccessful(transportResponse.getSuccessfulShards()) | ||
| .setTotal(transportResponse.getTotalShards()) | ||
| .addAllFailures(getShardFails(transportResponse)) | ||
| .build()); | ||
| // | ||
| // // [optional] Phase-level took time values in the response. | ||
| // optional PhaseTook phase_took = 4; | ||
| // | ||
| // // [required] Contains returned documents and metadata. | ||
| // HitsMetadata hits = 5; | ||
| TotalHits.TotalHitsRelation totalHitsRelation = TotalHits.TotalHitsRelation.TOTAL_HITS_RELATION_INVALID; | ||
| switch (transportResponse.getHits().getTotalHits().relation.name()) { | ||
| case "EQUAL_TO": | ||
| totalHitsRelation = TotalHits.TotalHitsRelation.TOTAL_HITS_RELATION_EQ; | ||
| break; | ||
| case "GREATER_THAN_OR_EQUAL_TO": | ||
| totalHitsRelation = TotalHits.TotalHitsRelation.TOTAL_HITS_RELATION_GTE; | ||
| break; | ||
| } | ||
| // transportResponse.getHits().getHits(); | ||
| responseBodyBuilder.setMaxScore(transportResponse.getHits().getMaxScore()).setHits(HitsMetadata.newBuilder() | ||
| .setTotal(opensearch.proto.HitsMetadata.Total.newBuilder() | ||
| .setTotalHits(TotalHits.newBuilder() | ||
| .setValue(transportResponse.getHits().getTotalHits().value) | ||
| .setRelation(totalHitsRelation) | ||
| .build()) | ||
| .build()) | ||
| .build()) | ||
| .build(); | ||
|
|
||
| // | ||
| // // [optional] When you search one or more remote clusters, a `_clusters` section is included to provide information about the search on each cluster. | ||
| // optional ClusterStatistics clusters = 6; | ||
| // | ||
| // // [optional] Retrieved specific fields in the search response | ||
| // optional .google.protobuf.Struct fields = 7; | ||
| // | ||
| // // [optional] Highest returned document _score. | ||
| // optional float max_score = 8; | ||
| // | ||
| // // [optional] The number of times that the coordinating node aggregates results from batches of shard responses | ||
| // optional int32 num_reduce_phases = 9; | ||
| // | ||
| // // [optional] Contains profiling information. | ||
| // Profile profile = 10; | ||
| // | ||
| // // [optional] The PIT ID. | ||
| // optional string pit_id = 11; | ||
| // | ||
| // // [optional] Identifier for the search and its search context. | ||
| // optional string scroll_id = 12; | ||
| // | ||
| // // [optional] If the query was terminated early, the terminated_early flag will be set to true in the response | ||
| // optional bool terminated_early = 13; | ||
|
|
||
| opensearch.proto.SearchResponse response = opensearch.proto.SearchResponse.newBuilder() | ||
| .setResponseBody(responseBodyBuilder.build()) | ||
| .build(); | ||
| return response; | ||
| } | ||
|
|
||
| public static List<ShardFailure> getShardFails(org.opensearch.action.search.SearchResponse transportResponse) { | ||
| ShardSearchFailure[] failures = transportResponse.getShardFailures(); | ||
| return Arrays.stream(failures) | ||
| .map(failure -> ShardFailure.newBuilder() | ||
| .setIndex(failure.index()) | ||
| .setShard(failure.shardId()) | ||
| .setStatus(failure.status().name()) | ||
| .build()) | ||
| .collect(Collectors.toList()); | ||
|
|
||
| } | ||
|
|
||
| static class SearchRequestActionListener implements ActionListener<SearchResponse> { | ||
| private StreamObserver<opensearch.proto.SearchResponse> respObserver = null; | ||
|
|
||
| public SearchRequestActionListener(StreamObserver<opensearch.proto.SearchResponse> responseObserver){ | ||
| super(); | ||
| respObserver = responseObserver; | ||
| } | ||
|
|
||
| @Override | ||
| public void onResponse(org.opensearch.action.search.SearchResponse response) { | ||
| try { | ||
| opensearch.proto.SearchResponse protoResponse = searchResponseToProto(response); | ||
| respObserver.onNext(protoResponse); | ||
| respObserver.onCompleted(); | ||
| } catch (Exception e) { | ||
| respObserver.onError( | ||
| new RuntimeException("Failed to process SearchResponse", e) | ||
| ); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(Exception e) { | ||
| respObserver.onError( | ||
| new RuntimeException("SearchRequest task failed", e) | ||
| ); | ||
| } | ||
| }; | ||
|
|
||
| } | ||
|
|
34 changes: 0 additions & 34 deletions
34
server/src/main/java/org/opensearch/grpc/services/search/SearchRequestBodyProtoHelper.java
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.