From 903dfb07a8c22a8ada197e231b7881cdd3d5e08e Mon Sep 17 00:00:00 2001 From: Dayana Jean Date: Sat, 1 Aug 2026 06:38:10 -0700 Subject: [PATCH 1/2] Add gRPC supports for Search foundation; match-all query (#2071) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: add transparent gRPC transport as separate java-client-grpc module Adds a transparent gRPC transport layer that routes bulk operations over gRPC for improved performance while falling back to REST for all other operations. Isolated in a separate java-client-grpc module to prevent classpath conflicts. Includes: - GrpcTransport + HybridTransport (automatic routing and fallback) - Translation layer (BulkRequest/Response <-> protobuf conversion) - TLS support (trust cert, trust store, mTLS, insecure, hostname override) - Basic auth, AWS SigV4, and JWT authentication interceptors - Channel health monitoring via gRPC connectivity state machine - Integration tests (framework-compliant, version-gated to 3.5.0+) - Sample code and CI configuration Signed-off-by: Dayana Jean * ci: retrigger CI Signed-off-by: Dayana Jean * Making the FieldMappingUtil package private Co-authored-by: Andriy Redko Signed-off-by: Dayana * Making this package private Apply suggestion from @reta Co-authored-by: Andriy Redko Signed-off-by: Dayana * Making BasicAuthInterceptor package private Apply suggestion from @reta Co-authored-by: Andriy Redko Signed-off-by: Dayana * Remove unused toHttpStatus method and fix test package for package-private classes - Removed GrpcStatusConverter.toHttpStatus() which was unused anywhere in the codebase (per maintainer feedback) - Moved TranslationTest to the translation package so it can access package-private FieldMappingUtil and GrpcStatusConverter.convert() Signed-off-by: Dayana Jean * refactor: split AWS SigV4 into AwsGrpcTransport per maintainer feedback Follows the same pattern as ApacheHttpClient5Transport (general) vs AwsSdk2Transport (AWS-specific) in the existing codebase. Changes: - Created AwsGrpcTransport: extends GrpcTransport, adds SigV4 signing - AwsGrpcTransport.awsBuilder(host, port).sigV4(...).tls(...).build() - Requires sigV4Config and TLS - Overrides preProcessBulk() for payload hash computation - Removed SigV4 from GrpcTransport: - Removed .sigV4() builder method - Removed sigV4Interceptor field - Added protected preProcessBulk() hook for subclasses - GrpcTransport is now purely general-purpose (basic auth, JWT, TLS) - Updated tests to use AwsGrpcTransport for SigV4 tests Signed-off-by: Dayana Jean * refactor: remove silent REST fallback from HybridTransport Per maintainer feedback: if the intent is to use gRPC for a supported endpoint, errors should propagate to the user rather than silently falling back to REST. Changes: - HybridTransport no longer catches gRPC errors and retries via REST - Removed fallbackOnError constructor parameter and field - Routing behavior preserved: unsupported endpoints → REST directly - gRPC-supported endpoints: errors propagate to caller - Simplified performRequest/performRequestAsync (no try/catch) - Updated tests: testGrpcErrorPropagatesForSupportedEndpoint Behavior: client.bulk(req) → gRPC (errors propagate if gRPC fails) client.search(req) → REST (not supported by gRPC, routed directly) Signed-off-by: Dayana Jean * docs: update comments to reflect routing instead of fallback Stale references to 'automatic REST fallback' replaced with accurate 'REST routing for unsupported endpoints' language throughout. - GrpcTransport: updated javadoc and error message - GrpcDemo: rewrote Demo 3 to show REST routing (not fallback on error) - GrpcAwsSigV4: updated to use AwsGrpcTransport.awsBuilder() - GrpcBulkIT: renamed testRestFallback → testRestRouting Signed-off-by: Dayana Jean * fix: make integration test base classes self-contained Per maintainer feedback: OpenSearchJavaClientTestCase and TestcontainersThreadFilter are internal test classes not exposed for external module use. Changes: - AbstractGrpcIT: now extends nothing, uses standard JUnit 4 + Assume - Removed dependency on OpenSearchJavaClientTestCase - Removed @ThreadLeakFilters(TestcontainersThreadFilter) - Uses Assume.assumeTrue() with manual version parsing - Reads cluster config from system properties directly - GrpcTransportSupport: converted from interface (extending OpenSearchTransportSupport) to utility class - GrpcBulkIT: no longer implements GrpcTransportSupport interface Signed-off-by: Dayana Jean * fix: lower java-client-grpc baseline from JDK 11 to JDK 8 Per maintainer feedback: opensearch-java baseline is JDK 8 and gRPC-Java supports Java 8. No technical reason to require JDK 11. Changes: - build.gradle.kts: targetCompatibility/sourceCompatibility → 1.8 - GrpcSigV4Test: replaced 'var' (Java 10+) with explicit types Signed-off-by: Dayana Jean * fix: remove explicit jackson test deps (transitive via java-client) Per maintainer feedback: opensearch-java has a hard dependency on Jackson 3.x, so it comes transitively. No need to declare it again. Signed-off-by: Dayana Jean * fix: wire up integration tests with java21 source set Per maintainer feedback: integration tests were not being compiled or run. Added the standard java21 source set configuration used by java-client to java-client-grpc. Changes: - Added unitTest/integrationTest task definitions - Added java21 source set (src/test/java11) gated on JDK 21+ - Added test framework, testcontainers, opensearch-testcontainers deps - Added static import for JUnit Assert in GrpcBulkIT - Integration tests now compile and will run with: ./gradlew :java-client-grpc:integrationTest -Dtests.opensearch.version=3.5.0 Signed-off-by: Dayana Jean * ci: add OpenSearch 3.5.0 to integration test matrix Adds the first gRPC-capable version to the test matrix so the gRPC integration tests run in CI. Includes both Java 21 and Java 25. Signed-off-by: Dayana Jean * fix: replace wildcard import with explicit imports (spotless) Spotless rejects wildcard imports. Replaced 'import static org.junit.Assert.*' with explicit imports for assertEquals, assertFalse, assertNotNull, assertTrue. Signed-off-by: Dayana Jean * fix: remove java21 classes from unitTest task The java21 source set only contains integration tests (integTest package). Adding it to unitTest causes 'No tests found' failure since the filter excludes integTest classes. Only integrationTest needs the java21 classes. Signed-off-by: Dayana Jean * fix: spotless formatting for samples (license header, import order) - GrpcDemo.java: replaced custom header with standard Apache-2.0 license - GrpcAwsSigV4.java: fixed import ordering (AwsGrpcTransport alphabetical) - Removed unused imports Signed-off-by: Dayana Jean * fix: skip gRPC integration tests when port is unreachable assumeGrpcSupported() now verifies both: 1. Server version is 3.5.0+ (via REST info endpoint) 2. gRPC port is actually reachable (TCP socket check) This prevents test failures when running integrationTest against OpenSearch versions that don't have gRPC enabled or when the gRPC port isn't exposed by testcontainers. Signed-off-by: Dayana Jean * fix: only enable gRPC in testcontainer for OpenSearch 3.5.0+ GrpcTestContainerRule now checks the tests.opensearch.version property before adding gRPC configuration. Older versions (1.x, 2.x) don't support aux.transport.types and would fail to start. On pre-3.5.0 versions: - Container starts without gRPC config (REST only) - gRPC port not exposed - Tests skip via assumeGrpcSupported() (version check + port check) Signed-off-by: Dayana Jean * ci: exclude grpc integration tests until OpenSearch 3.5.0 is released OpenSearch 3.5.0 Docker image is not yet published, so the gRPC integration tests cannot run in CI. Changes: - Exclude :java-client-grpc:integrationTest from the main CI run (all matrix versions are pre-3.5.0) - Comment out 3.5.0 entries in test matrix (uncomment when released) The gRPC integration tests can still be run locally against a container started manually or once 3.5.0 is published. Signed-off-by: Dayana Jean * fix: graceful skip on container failure + re-enable 3.5.0 in CI OpenSearch 3.5.0 Docker image is published. Re-enabled in CI matrix. Changes: - GrpcTestContainerRule.before(): catches ContainerLaunchException and calls Assume.assumeTrue(false) to skip tests gracefully instead of failing the build - Added 5-minute startup timeout for CI environments - Removed duplicate disk watermark env var - Re-enabled 3.5.0 in test-integration.yml matrix - Removed -x :java-client-grpc:integrationTest exclusion On pre-3.5.0 versions: container starts without gRPC, tests skip via assumeGrpcSupported(). On 3.5.0+: if container fails for any reason, tests skip instead of failing the build. Signed-off-by: Dayana Jean * fix: spotless formatting on GrpcTestContainerRule Signed-off-by: Dayana Jean * fix: remove grpcStatusToHttpStatus from FieldMappingUtil Per maintainer feedback: method was only used in tests, not in production code. Removed the method and its 7 associated tests. Signed-off-by: Dayana Jean * fix: skip gRPC tests early with assumeTrue for unsupported versions Per maintainer feedback: use assumeTrue in the class rule to skip tests immediately for older versions, rather than conditionally configuring the container. Changes: - GrpcTestContainerRule.before(): assumeTrue("OpenSearch should support gRPC", supportsGrpc(version)) skips for pre-3.5.0 - createContainer(): always configures gRPC (only reached for 3.5.0+) - Removed conditional gRPC config logic Signed-off-by: Dayana Jean * fix: make GrpcChannelFactory methods package-private Signed-off-by: Dayana Jean * fix: apply maintainer suggestions on visibility and naming 1. BasicAuthInterceptor: constructor now package-private 2. GrpcChannelFactory: class now package-private (final class, no public) 3. GrpcTestContainerRule renamed to OpenSearchGrpcTestContainerRule 4. Updated references in AbstractGrpcIT Signed-off-by: Dayana Jean * feat: implement search over gRPC with match_all query support Adds search-over-gRPC support to the Java client transport, starting with match_all query. Follows the same pattern as Bulk. New files: - SearchRequestConverter: converts SearchRequest → protobuf - match_all query fully implemented - Unsupported query types throw UnsupportedOperationException with message directing to REST or contributing the converter - SearchResponseConverter: converts protobuf SearchResponse → java - Full response: took, timed_out, _shards, hits (total, max_score) - Per-hit: _index, _id, _score, _version, _seq_no, _primary_term - _source deserialization: bytes → JSON → TDocument via JsonpMapper - SearchRequestConverterTest: 7 unit tests - GrpcSearchIT: 3 integration tests (match_all, pagination, empty) Modified files: - GrpcTransport: added SearchServiceBlockingStub, registered SearchRequest._ENDPOINT, added performSearch() handler Query types not yet implemented (server supports them, client converter needs extending): - term, terms, match, match_phrase, bool, range, prefix, wildcard, regexp, fuzzy, exists, ids, nested, geo_distance, knn, hybrid Signed-off-by: Dayana Jean * Enhanced the endpoint support with request argument as suggested from feedback Signed-off-by: Dayana Jean --------- Signed-off-by: Dayana Jean Signed-off-by: Dayana Co-authored-by: Andriy Redko (cherry picked from commit f131e6d8acc0dece9383e417cf28616a6dea3808) Signed-off-by: opensearch-ci-bot --- CHANGELOG.md | 1 + .../client/transport/grpc/GrpcTransport.java | 62 +++++- .../transport/grpc/HybridTransport.java | 8 +- .../translation/SearchRequestConverter.java | 108 ++++++++++ .../translation/SearchResponseConverter.java | 174 ++++++++++++++++ .../transport/grpc/GrpcTransportTest.java | 35 ++++ .../SearchRequestConverterTest.java | 101 +++++++++ .../integTest/grpc/GrpcSearchIT.java | 196 ++++++++++++++++++ 8 files changed, 679 insertions(+), 6 deletions(-) create mode 100644 java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/translation/SearchRequestConverter.java create mode 100644 java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/translation/SearchResponseConverter.java create mode 100644 java-client-grpc/src/test/java/org/opensearch/client/transport/grpc/translation/SearchRequestConverterTest.java create mode 100644 java-client-grpc/src/test/java11/org/opensearch/client/opensearch/integTest/grpc/GrpcSearchIT.java diff --git a/CHANGELOG.md b/CHANGELOG.md index d13515922..0775ce9e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) - Added `equals()` and `hashCode()` implementations to `FieldValue` ([#1998](https://github.com/opensearch-project/opensearch-java/pull/1998)) - Add document lifecycle guide and runnable sample ([#2017](https://github.com/opensearch-project/opensearch-java/pull/2017)) - Add transparent gRPC transport with HybridTransport (bulk over gRPC, REST fallback), translation layer, TLS, basic auth, AWS SigV4, and JWT support ([#2062](https://github.com/opensearch-project/opensearch-java/pull/2062)) +- Add search over gRPC with match_all query support, SearchRequestConverter, SearchResponseConverter, and _source deserialization ([#2071](https://github.com/opensearch-project/opensearch-java/pull/2071)) ### Dependencies - Bump `org.apache.httpcomponents.client5:httpclient5` from 5.6 to 5.6.1 ([#1967](https://github.com/opensearch-project/opensearch-java/pull/1967)) diff --git a/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/GrpcTransport.java b/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/GrpcTransport.java index 83f09cf54..814b122a9 100644 --- a/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/GrpcTransport.java +++ b/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/GrpcTransport.java @@ -17,6 +17,8 @@ import org.opensearch.client.json.JsonpMapper; import org.opensearch.client.opensearch.core.BulkRequest; import org.opensearch.client.opensearch.core.BulkResponse; +import org.opensearch.client.opensearch.core.SearchRequest; +import org.opensearch.client.opensearch.core.SearchResponse; import org.opensearch.client.transport.Endpoint; import org.opensearch.client.transport.OpenSearchTransport; import org.opensearch.client.transport.TransportException; @@ -25,6 +27,7 @@ import org.opensearch.client.transport.grpc.translation.BulkResponseConverter; import org.opensearch.client.transport.grpc.translation.GrpcStatusConverter; import org.opensearch.protobufs.services.DocumentServiceGrpc; +import org.opensearch.protobufs.services.SearchServiceGrpc; /** * Pure gRPC transport for OpenSearch. Implements {@link OpenSearchTransport} and routes @@ -52,7 +55,8 @@ public class GrpcTransport implements OpenSearchTransport { static { java.util.Set> endpoints = new java.util.HashSet<>(); endpoints.add(BulkRequest._ENDPOINT); - // Future: SearchRequest._ENDPOINT, KnnSearchRequest._ENDPOINT + endpoints.add(SearchRequest._ENDPOINT); + // Future: KnnSearchRequest._ENDPOINT SUPPORTED_ENDPOINTS = java.util.Collections.unmodifiableSet(endpoints); } @@ -63,10 +67,35 @@ public static boolean isEndpointSupported(Endpoint endpoint) { return SUPPORTED_ENDPOINTS.contains(endpoint); } + /** + * Returns true if the given endpoint and request can be handled by gRPC transport. + * For endpoints that only partially support gRPC (e.g., search with limited query types), + * this method inspects the request to determine if gRPC can handle it. + */ + public static boolean isEndpointSupported(Endpoint endpoint, RequestT request) { + if (!SUPPORTED_ENDPOINTS.contains(endpoint)) { + return false; + } + // Bulk: all operations supported + if (endpoint == BulkRequest._ENDPOINT) { + return true; + } + // Search: only match_all is currently supported + if (endpoint == SearchRequest._ENDPOINT && request instanceof SearchRequest) { + SearchRequest searchRequest = (SearchRequest) request; + if (searchRequest.query() == null) { + return true; // No query = match_all by default + } + return searchRequest.query().isMatchAll(); + } + return true; + } + // ─── Instance Fields ───────────────────────────────────────────────────────── private final ManagedChannel channel; private final DocumentServiceGrpc.DocumentServiceBlockingStub documentStub; + private final SearchServiceGrpc.SearchServiceBlockingStub searchStub; private final JsonpMapper jsonpMapper; private final GrpcTransportOptions grpcOptions; private final TransportOptions transportOptions; @@ -81,6 +110,7 @@ public static boolean isEndpointSupported(Endpoint endpoint) { ) { this.channel = channel; this.documentStub = channel != null ? DocumentServiceGrpc.newBlockingStub(channel) : null; + this.searchStub = channel != null ? SearchServiceGrpc.newBlockingStub(channel) : null; this.jsonpMapper = jsonpMapper; this.grpcOptions = grpcOptions; this.transportOptions = transportOptions; @@ -101,7 +131,7 @@ public ResponseT performRequest( @Nullable TransportOptions options ) throws IOException { - if (!GrpcTransport.isEndpointSupported(endpoint)) { + if (!GrpcTransport.isEndpointSupported(endpoint, request)) { throw new UnsupportedOperationException( "Endpoint not supported by gRPC transport: " + endpoint.requestUrl(request) @@ -113,6 +143,9 @@ public ResponseT performRequest( if (endpoint == BulkRequest._ENDPOINT) { return (ResponseT) performBulk((BulkRequest) request); } + if (endpoint == SearchRequest._ENDPOINT) { + return (ResponseT) performSearch((SearchRequest) request); + } throw new UnsupportedOperationException("Endpoint registered but no handler: " + endpoint.requestUrl(request)); } @@ -238,6 +271,31 @@ private BulkResponse performBulk(BulkRequest request) throws TransportException } } + @SuppressWarnings("unchecked") + private SearchResponse performSearch(SearchRequest request) throws TransportException { + // Convert client request to protobuf + org.opensearch.protobufs.SearchRequest protoRequest = org.opensearch.client.transport.grpc.translation.SearchRequestConverter + .toProto(request, jsonpMapper); + + // Execute gRPC call + try { + org.opensearch.protobufs.SearchResponse protoResponse = searchStub.search(protoRequest); + + // Convert response — use Object.class as default; the actual deserialization + // is handled by the endpoint's response deserializer in the transport layer + return (SearchResponse) org.opensearch.client.transport.grpc.translation.SearchResponseConverter.fromProto( + protoResponse, + jsonpMapper, + (Class) Object.class + ); + } catch (StatusRuntimeException e) { + throw new TransportException( + "gRPC search request failed: " + e.getStatus().getDescription(), + new org.opensearch.client.transport.TransportException(e.getMessage(), e) + ); + } + } + // ─── Builder ───────────────────────────────────────────────────────────────── /** diff --git a/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/HybridTransport.java b/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/HybridTransport.java index 219bf7b3c..2cfabd56b 100644 --- a/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/HybridTransport.java +++ b/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/HybridTransport.java @@ -58,8 +58,8 @@ public ResponseT performRequest( @Nullable TransportOptions options ) throws IOException { - // Route unsupported endpoints directly to REST - if (!GrpcTransport.isEndpointSupported(endpoint)) { + // Route unsupported endpoints/requests directly to REST + if (!GrpcTransport.isEndpointSupported(endpoint, request)) { return restTransport.performRequest(request, endpoint, options); } @@ -73,8 +73,8 @@ public CompletableFuture performRequest Endpoint endpoint, @Nullable TransportOptions options ) { - // Route unsupported endpoints directly to REST - if (!GrpcTransport.isEndpointSupported(endpoint)) { + // Route unsupported endpoints/requests directly to REST + if (!GrpcTransport.isEndpointSupported(endpoint, request)) { return restTransport.performRequestAsync(request, endpoint, options); } diff --git a/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/translation/SearchRequestConverter.java b/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/translation/SearchRequestConverter.java new file mode 100644 index 000000000..3ee553c99 --- /dev/null +++ b/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/translation/SearchRequestConverter.java @@ -0,0 +1,108 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.client.transport.grpc.translation; + +import org.opensearch.client.json.JsonpMapper; +import org.opensearch.client.opensearch._types.query_dsl.MatchAllQuery; +import org.opensearch.client.opensearch._types.query_dsl.Query; +import org.opensearch.client.opensearch.core.SearchRequest; + +/** + * Converts opensearch-java SearchRequest to protobuf SearchRequest. + * + * Currently supported query types: + * - match_all — matches all documents + * + * Query types not yet implemented (these can be executed over gRPC once + * the converter is extended — the server already supports them): + * - term, terms, terms_set + * - match, match_phrase, match_phrase_prefix, match_bool_prefix, multi_match + * - bool, constant_score, function_score + * - range, prefix, wildcard, regexp, fuzzy, exists, ids + * - nested, geo_distance, geo_bounding_box + * - knn, hybrid, script + * + * To add a new query type, implement a convertXxx() method and add a case + * to convertQuery(Query). + */ +public class SearchRequestConverter { + + /** + * Convert an opensearch-java SearchRequest to a protobuf SearchRequest. + * + * @param request the client SearchRequest + * @param jsonpMapper the JSON mapper (for future use with script queries) + * @return the protobuf SearchRequest + */ + public static org.opensearch.protobufs.SearchRequest toProto(SearchRequest request, JsonpMapper jsonpMapper) { + org.opensearch.protobufs.SearchRequest.Builder protoBuilder = org.opensearch.protobufs.SearchRequest.newBuilder(); + + // Set index(es) + if (request.index() != null && !request.index().isEmpty()) { + protoBuilder.addAllIndex(request.index()); + } + + // Build SearchRequestBody + org.opensearch.protobufs.SearchRequestBody.Builder bodyBuilder = org.opensearch.protobufs.SearchRequestBody.newBuilder(); + + // Set size + if (request.size() != null) { + bodyBuilder.setSize(request.size()); + } + + // Set from (pagination offset) + if (request.from() != null) { + bodyBuilder.setFrom(request.from()); + } + + // Convert query + if (request.query() != null) { + bodyBuilder.setQuery(convertQuery(request.query())); + } + + protoBuilder.setSearchRequestBody(bodyBuilder.build()); + return protoBuilder.build(); + } + + /** + * Convert an opensearch-java Query to a protobuf QueryContainer. + * + * @throws UnsupportedOperationException if the query type is not yet supported + */ + private static org.opensearch.protobufs.QueryContainer convertQuery(Query query) { + org.opensearch.protobufs.QueryContainer.Builder containerBuilder = org.opensearch.protobufs.QueryContainer.newBuilder(); + + if (query.isMatchAll()) { + containerBuilder.setMatchAll(convertMatchAll(query.matchAll())); + } else { + throw new UnsupportedOperationException( + "Query type '" + + query._kind() + + "' is not yet supported for gRPC transport in this version. " + + "Use REST transport for this query type." + ); + } + + return containerBuilder.build(); + } + + /** + * Convert MatchAllQuery to protobuf. + * MatchAllQuery has only optional boost and _name fields. + */ + private static org.opensearch.protobufs.MatchAllQuery convertMatchAll(MatchAllQuery matchAll) { + org.opensearch.protobufs.MatchAllQuery.Builder builder = org.opensearch.protobufs.MatchAllQuery.newBuilder(); + + if (matchAll.boost() != null) { + builder.setBoost(matchAll.boost().floatValue()); + } + + return builder.build(); + } +} diff --git a/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/translation/SearchResponseConverter.java b/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/translation/SearchResponseConverter.java new file mode 100644 index 000000000..6de681d5c --- /dev/null +++ b/java-client-grpc/src/main/java/org/opensearch/client/transport/grpc/translation/SearchResponseConverter.java @@ -0,0 +1,174 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.client.transport.grpc.translation; + +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; +import org.opensearch.client.json.JsonpMapper; +import org.opensearch.client.opensearch._types.ShardStatistics; +import org.opensearch.client.opensearch.core.SearchResponse; +import org.opensearch.client.opensearch.core.search.Hit; +import org.opensearch.client.opensearch.core.search.HitsMetadata; +import org.opensearch.client.opensearch.core.search.TotalHits; +import org.opensearch.client.opensearch.core.search.TotalHitsRelation; + +/** + * Converts protobuf SearchResponse to opensearch-java SearchResponse format. + *

+ * Currently supports match_all queries. Other query types (term, match, bool, etc.) + * are not yet implemented in the client-side converter but can be executed over gRPC + * once the request converter supports them — the response format is the same regardless + * of query type. + *

+ * Note: Unsupported query types will throw UnsupportedOperationException at the request + * conversion stage. The response converter handles all response formats since the + * SearchResponse structure is query-agnostic. + */ +public class SearchResponseConverter { + + /** + * Convert a protobuf SearchResponse to an opensearch-java SearchResponse. + * + * @param protoResponse the protobuf SearchResponse from the server + * @param jsonpMapper the JSON mapper for _source deserialization + * @param tDocumentClass the target class for hit _source deserialization + * @param the document type + * @return the opensearch-java SearchResponse + */ + public static SearchResponse fromProto( + org.opensearch.protobufs.SearchResponse protoResponse, + JsonpMapper jsonpMapper, + Class tDocumentClass + ) { + SearchResponse.Builder builder = new SearchResponse.Builder(); + + // took + builder.took(protoResponse.getTook()); + + // timed_out + builder.timedOut(protoResponse.getTimedOut()); + + // _shards + if (protoResponse.hasXShards()) { + org.opensearch.protobufs.ShardStatistics protoShards = protoResponse.getXShards(); + builder.shards( + new ShardStatistics.Builder().total(protoShards.getTotal()) + .successful(protoShards.getSuccessful()) + .failed(protoShards.getFailed()) + .build() + ); + } else { + builder.shards(new ShardStatistics.Builder().total(0).successful(0).failed(0).build()); + } + + // hits + if (protoResponse.hasHits()) { + builder.hits(convertHitsMetadata(protoResponse.getHits(), jsonpMapper, tDocumentClass)); + } else { + builder.hits(new HitsMetadata.Builder().hits(new ArrayList<>()).build()); + } + + return builder.build(); + } + + private static HitsMetadata convertHitsMetadata( + org.opensearch.protobufs.HitsMetadata protoHits, + JsonpMapper jsonpMapper, + Class tDocumentClass + ) { + HitsMetadata.Builder builder = new HitsMetadata.Builder(); + + // total hits + if (protoHits.hasTotal()) { + org.opensearch.protobufs.HitsMetadataTotal protoTotal = protoHits.getTotal(); + if (protoTotal.hasTotalHits()) { + org.opensearch.protobufs.TotalHits totalHits = protoTotal.getTotalHits(); + TotalHitsRelation relation = totalHits.getRelation() == org.opensearch.protobufs.TotalHitsRelation.TOTAL_HITS_RELATION_EQ + ? TotalHitsRelation.Eq + : TotalHitsRelation.Gte; + builder.total(new TotalHits.Builder().value(totalHits.getValue()).relation(relation).build()); + } + } + + // max_score + if (protoHits.hasMaxScore()) { + org.opensearch.protobufs.HitsMetadataMaxScore maxScore = protoHits.getMaxScore(); + if (maxScore.hasFloat()) { + builder.maxScore(maxScore.getFloat()); + } + } + + // individual hits + List> hits = new ArrayList<>(); + for (org.opensearch.protobufs.HitsMetadataHitsInner protoHit : protoHits.getHitsList()) { + hits.add(convertHit(protoHit, jsonpMapper, tDocumentClass)); + } + builder.hits(hits); + + return builder.build(); + } + + private static Hit convertHit( + org.opensearch.protobufs.HitsMetadataHitsInner protoHit, + JsonpMapper jsonpMapper, + Class tDocumentClass + ) { + Hit.Builder builder = new Hit.Builder(); + + // _index + builder.index(protoHit.getXIndex()); + + // _id + builder.id(protoHit.getXId()); + + // _score + if (protoHit.hasXScore()) { + org.opensearch.protobufs.HitXScore score = protoHit.getXScore(); + if (score.hasDouble()) { + builder.score(score.getDouble()); + } + } + + // _version + if (protoHit.hasXVersion()) { + builder.version(protoHit.getXVersion()); + } + + // _seq_no + if (protoHit.hasXSeqNo()) { + builder.seqNo(protoHit.getXSeqNo()); + } + + // _primary_term + if (protoHit.hasXPrimaryTerm()) { + builder.primaryTerm(protoHit.getXPrimaryTerm()); + } + + // _source — decode bytes and deserialize to TDocument + if (!protoHit.getXSource().isEmpty()) { + TDocument source = deserializeSource(protoHit.getXSource().toByteArray(), jsonpMapper, tDocumentClass); + builder.source(source); + } + + return builder.build(); + } + + /** + * Decode _source bytes from protobuf hit to a Java object. + * The server returns _source as UTF-8 JSON bytes. + */ + static TDocument deserializeSource(byte[] sourceBytes, JsonpMapper jsonpMapper, Class tDocumentClass) { + InputStream stream = new ByteArrayInputStream(sourceBytes); + jakarta.json.stream.JsonParser parser = jsonpMapper.jsonProvider().createParser(stream); + parser.next(); // advance to first token + return jsonpMapper.deserialize(parser, tDocumentClass); + } +} diff --git a/java-client-grpc/src/test/java/org/opensearch/client/transport/grpc/GrpcTransportTest.java b/java-client-grpc/src/test/java/org/opensearch/client/transport/grpc/GrpcTransportTest.java index 1ed6c102d..0f4c0d88f 100644 --- a/java-client-grpc/src/test/java/org/opensearch/client/transport/grpc/GrpcTransportTest.java +++ b/java-client-grpc/src/test/java/org/opensearch/client/transport/grpc/GrpcTransportTest.java @@ -24,6 +24,7 @@ import org.opensearch.client.json.jackson.JacksonJsonpMapper; import org.opensearch.client.opensearch.core.BulkRequest; import org.opensearch.client.opensearch.core.BulkResponse; +import org.opensearch.client.opensearch.core.SearchRequest; import org.opensearch.client.transport.Endpoint; import org.opensearch.client.transport.OpenSearchTransport; import org.opensearch.client.transport.TransportException; @@ -74,6 +75,40 @@ public void testBulkSupported() { assertTrue(GrpcTransport.isEndpointSupported(BulkRequest._ENDPOINT)); } + @Test + public void testBulkSupportedWithRequest() { + BulkRequest request = new BulkRequest.Builder().operations(op -> op.delete(d -> d.id("1").index("t"))).build(); + assertTrue(GrpcTransport.isEndpointSupported(BulkRequest._ENDPOINT, request)); + } + + @Test + public void testSearchMatchAllSupportedWithRequest() { + SearchRequest request = new SearchRequest.Builder().index("test").query(q -> q.matchAll(m -> m)).build(); + assertTrue(GrpcTransport.isEndpointSupported(SearchRequest._ENDPOINT, request)); + } + + @Test + public void testSearchNoQuerySupportedWithRequest() { + SearchRequest request = new SearchRequest.Builder().index("test").build(); + assertTrue(GrpcTransport.isEndpointSupported(SearchRequest._ENDPOINT, request)); + } + + @Test + public void testSearchTermQueryNotSupportedWithRequest() { + SearchRequest request = new SearchRequest.Builder().index("test") + .query(q -> q.term(t -> t.field("status").value(v -> v.stringValue("active")))) + .build(); + assertFalse(GrpcTransport.isEndpointSupported(SearchRequest._ENDPOINT, request)); + } + + @Test + public void testSearchMatchQueryNotSupportedWithRequest() { + SearchRequest request = new SearchRequest.Builder().index("test") + .query(q -> q.match(m -> m.field("title").query(fv -> fv.stringValue("hello")))) + .build(); + assertFalse(GrpcTransport.isEndpointSupported(SearchRequest._ENDPOINT, request)); + } + @Test public void testUnsupportedEndpoint() { Endpoint fake = new Endpoint() { diff --git a/java-client-grpc/src/test/java/org/opensearch/client/transport/grpc/translation/SearchRequestConverterTest.java b/java-client-grpc/src/test/java/org/opensearch/client/transport/grpc/translation/SearchRequestConverterTest.java new file mode 100644 index 000000000..0de757a9e --- /dev/null +++ b/java-client-grpc/src/test/java/org/opensearch/client/transport/grpc/translation/SearchRequestConverterTest.java @@ -0,0 +1,101 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.client.transport.grpc.translation; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; +import org.opensearch.client.json.jackson3.JacksonJsonpMapper; +import org.opensearch.client.opensearch.core.SearchRequest; + +/** + * Unit tests for SearchRequestConverter. + */ +public class SearchRequestConverterTest { + + private final JacksonJsonpMapper mapper = new JacksonJsonpMapper(); + + @Test + public void testMatchAllQueryBasic() { + SearchRequest request = new SearchRequest.Builder().index("test-index").query(q -> q.matchAll(m -> m)).build(); + + org.opensearch.protobufs.SearchRequest proto = SearchRequestConverter.toProto(request, mapper); + + assertNotNull(proto); + assertEquals(1, proto.getIndexCount()); + assertEquals("test-index", proto.getIndex(0)); + assertTrue(proto.hasSearchRequestBody()); + assertTrue(proto.getSearchRequestBody().hasQuery()); + assertTrue(proto.getSearchRequestBody().getQuery().hasMatchAll()); + } + + @Test + public void testMatchAllQueryWithBoost() { + SearchRequest request = new SearchRequest.Builder().index("test-index").query(q -> q.matchAll(m -> m.boost(2.0f))).build(); + + org.opensearch.protobufs.SearchRequest proto = SearchRequestConverter.toProto(request, mapper); + + assertTrue(proto.getSearchRequestBody().getQuery().hasMatchAll()); + assertEquals(2.0f, proto.getSearchRequestBody().getQuery().getMatchAll().getBoost(), 0.001f); + } + + @Test + public void testMatchAllWithSize() { + SearchRequest request = new SearchRequest.Builder().index("test-index").query(q -> q.matchAll(m -> m)).size(50).build(); + + org.opensearch.protobufs.SearchRequest proto = SearchRequestConverter.toProto(request, mapper); + + assertEquals(50, proto.getSearchRequestBody().getSize()); + } + + @Test + public void testMatchAllWithFrom() { + SearchRequest request = new SearchRequest.Builder().index("test-index").query(q -> q.matchAll(m -> m)).from(10).size(20).build(); + + org.opensearch.protobufs.SearchRequest proto = SearchRequestConverter.toProto(request, mapper); + + assertEquals(10, proto.getSearchRequestBody().getFrom()); + assertEquals(20, proto.getSearchRequestBody().getSize()); + } + + @Test + public void testMultipleIndexes() { + SearchRequest request = new SearchRequest.Builder().index("index-1", "index-2", "index-3").query(q -> q.matchAll(m -> m)).build(); + + org.opensearch.protobufs.SearchRequest proto = SearchRequestConverter.toProto(request, mapper); + + assertEquals(3, proto.getIndexCount()); + assertEquals("index-1", proto.getIndex(0)); + assertEquals("index-2", proto.getIndex(1)); + assertEquals("index-3", proto.getIndex(2)); + } + + @Test + public void testNoQuery() { + SearchRequest request = new SearchRequest.Builder().index("test-index").size(10).build(); + + org.opensearch.protobufs.SearchRequest proto = SearchRequestConverter.toProto(request, mapper); + + assertNotNull(proto); + assertTrue(proto.hasSearchRequestBody()); + // No query set — server returns all docs by default + } + + @Test(expected = UnsupportedOperationException.class) + public void testUnsupportedQueryTypeThrows() { + SearchRequest request = new SearchRequest.Builder().index("test-index") + .query(q -> q.term(t -> t.field("status").value(v -> v.stringValue("active")))) + .build(); + + // Should throw UnsupportedOperationException for term query + SearchRequestConverter.toProto(request, mapper); + } +} diff --git a/java-client-grpc/src/test/java11/org/opensearch/client/opensearch/integTest/grpc/GrpcSearchIT.java b/java-client-grpc/src/test/java11/org/opensearch/client/opensearch/integTest/grpc/GrpcSearchIT.java new file mode 100644 index 000000000..a0ac4a6bf --- /dev/null +++ b/java-client-grpc/src/test/java11/org/opensearch/client/opensearch/integTest/grpc/GrpcSearchIT.java @@ -0,0 +1,196 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.client.opensearch.integTest.grpc; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import org.junit.Test; +import org.opensearch.client.opensearch._types.Refresh; +import org.opensearch.client.opensearch.core.BulkRequest; +import org.opensearch.client.opensearch.core.BulkResponse; +import org.opensearch.client.opensearch.core.SearchResponse; +import org.opensearch.client.opensearch.core.bulk.BulkOperation; +import org.opensearch.client.opensearch.core.bulk.IndexOperation; +import org.opensearch.client.opensearch.core.search.Hit; + +/** + * Integration tests for Search over gRPC transport. + * + * Verifies the complete search pipeline: + * 1. Index documents via gRPC bulk + * 2. Search with match_all via gRPC + * 3. Verify response structure and _source deserialization + * + * Skips automatically on OpenSearch versions below 3.5.0. + * + * Run: + * ./gradlew integrationTest --tests "org.opensearch.client.opensearch.integTest.grpc.GrpcSearchIT" \ + * -Dtests.opensearch.version=3.5.0 + */ +public class GrpcSearchIT extends AbstractGrpcIT { + + private static final String INDEX = "grpc-search-it"; + + // ─── Test Document ─────────────────────────────────────────────────────────── + + public static class Movie { + public String title; + public int year; + public String director; + + public Movie() {} + + public Movie(String title, int year, String director) { + this.title = title; + this.year = year; + this.director = director; + } + } + + // ─── Tests ─────────────────────────────────────────────────────────────────── + + @Test + public void testMatchAllSearch() throws IOException { + assumeGrpcSupported(); + + try { + // Setup: create index and index documents via gRPC bulk + grpcClient().indices().create(c -> c.index(INDEX)); + + List ops = new ArrayList<>(); + ops.add( + new BulkOperation.Builder().index( + new IndexOperation.Builder().index(INDEX) + .id("1") + .document(new Movie("The Dark Knight", 2008, "Christopher Nolan")) + .build() + ).build() + ); + ops.add( + new BulkOperation.Builder().index( + new IndexOperation.Builder().index(INDEX) + .id("2") + .document(new Movie("Inception", 2010, "Christopher Nolan")) + .build() + ).build() + ); + ops.add( + new BulkOperation.Builder().index( + new IndexOperation.Builder().index(INDEX) + .id("3") + .document(new Movie("Interstellar", 2014, "Christopher Nolan")) + .build() + ).build() + ); + + BulkResponse bulkResponse = grpcClient().bulk( + new BulkRequest.Builder().index(INDEX).operations(ops).refresh(Refresh.True).build() + ); + assertFalse("Bulk should have no errors", bulkResponse.errors()); + + // Search: match_all via gRPC + SearchResponse searchResponse = grpcClient().search(s -> s.index(INDEX).query(q -> q.matchAll(m -> m)), Movie.class); + + // Verify response structure + assertNotNull("Search response should not be null", searchResponse); + assertNotNull("Hits should not be null", searchResponse.hits()); + assertNotNull("Total should not be null", searchResponse.hits().total()); + assertEquals("Should find 3 documents", 3L, searchResponse.hits().total().value()); + assertTrue("Took should be >= 0", searchResponse.took() >= 0); + assertFalse("Should not time out", searchResponse.timedOut()); + + // Verify hits + List> hits = searchResponse.hits().hits(); + assertEquals("Should have 3 hits", 3, hits.size()); + + // Verify _source deserialization + for (Hit hit : hits) { + assertNotNull("Hit should have _index", hit.index()); + assertNotNull("Hit should have _id", hit.id()); + assertNotNull("Hit should have _source", hit.source()); + assertNotNull("Movie should have title", hit.source().title); + assertTrue("Movie year should be > 0", hit.source().year > 0); + assertNotNull("Movie should have director", hit.source().director); + } + + } finally { + // Cleanup + grpcClient().indices().delete(d -> d.index(INDEX).ignoreUnavailable(true)); + } + } + + @Test + public void testMatchAllWithSizeAndFrom() throws IOException { + assumeGrpcSupported(); + + try { + // Setup: index 5 documents + grpcClient().indices().create(c -> c.index(INDEX)); + + List ops = new ArrayList<>(); + for (int i = 1; i <= 5; i++) { + final int id = i; + ops.add( + new BulkOperation.Builder().index( + new IndexOperation.Builder().index(INDEX) + .id(String.valueOf(id)) + .document(new Movie("Movie " + id, 2000 + id, "Director " + id)) + .build() + ).build() + ); + } + + grpcClient().bulk(new BulkRequest.Builder().index(INDEX).operations(ops).refresh(Refresh.True).build()); + + // Search with size=2 + SearchResponse response = grpcClient().search(s -> s.index(INDEX).query(q -> q.matchAll(m -> m)).size(2), Movie.class); + + assertEquals("Total should be 5", 5L, response.hits().total().value()); + assertEquals("Should return 2 hits (size=2)", 2, response.hits().hits().size()); + + // Search with from=2, size=2 (pagination) + SearchResponse page2 = grpcClient().search( + s -> s.index(INDEX).query(q -> q.matchAll(m -> m)).from(2).size(2), + Movie.class + ); + + assertEquals("Total should still be 5", 5L, page2.hits().total().value()); + assertEquals("Should return 2 hits (page 2)", 2, page2.hits().hits().size()); + + } finally { + grpcClient().indices().delete(d -> d.index(INDEX).ignoreUnavailable(true)); + } + } + + @Test + public void testSearchEmptyIndex() throws IOException { + assumeGrpcSupported(); + + try { + // Create empty index + grpcClient().indices().create(c -> c.index(INDEX)); + + // Search empty index + SearchResponse response = grpcClient().search(s -> s.index(INDEX).query(q -> q.matchAll(m -> m)), Movie.class); + + assertNotNull("Response should not be null", response); + assertEquals("Should find 0 documents", 0L, response.hits().total().value()); + assertTrue("Hits list should be empty", response.hits().hits().isEmpty()); + + } finally { + grpcClient().indices().delete(d -> d.index(INDEX).ignoreUnavailable(true)); + } + } +} From e30ea3687112b35d8198a84e06adf9466107d519 Mon Sep 17 00:00:00 2001 From: Andriy Redko Date: Sun, 2 Aug 2026 12:54:28 -0400 Subject: [PATCH 2/2] Update java-client-grpc/src/test/java/org/opensearch/client/transport/grpc/translation/SearchRequestConverterTest.java Signed-off-by: Andriy Redko --- .../transport/grpc/translation/SearchRequestConverterTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/java-client-grpc/src/test/java/org/opensearch/client/transport/grpc/translation/SearchRequestConverterTest.java b/java-client-grpc/src/test/java/org/opensearch/client/transport/grpc/translation/SearchRequestConverterTest.java index 0de757a9e..48263d843 100644 --- a/java-client-grpc/src/test/java/org/opensearch/client/transport/grpc/translation/SearchRequestConverterTest.java +++ b/java-client-grpc/src/test/java/org/opensearch/client/transport/grpc/translation/SearchRequestConverterTest.java @@ -13,7 +13,7 @@ import static org.junit.Assert.assertTrue; import org.junit.Test; -import org.opensearch.client.json.jackson3.JacksonJsonpMapper; +import org.opensearch.client.json.jackson.JacksonJsonpMapper; import org.opensearch.client.opensearch.core.SearchRequest; /**