diff --git a/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/VectorEmbeddingIntegrationIT.java b/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/VectorEmbeddingIntegrationIT.java index 64c769c39e61..c08d39df2e65 100644 --- a/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/VectorEmbeddingIntegrationIT.java +++ b/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/VectorEmbeddingIntegrationIT.java @@ -16,6 +16,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.UUID; import org.apache.hc.core5.http.HttpHost; import org.junit.jupiter.api.AfterAll; @@ -38,6 +39,7 @@ import org.testcontainers.utility.DockerImageName; import os.org.opensearch.client.json.jackson.JacksonJsonpMapper; import os.org.opensearch.client.opensearch.OpenSearchClient; +import os.org.opensearch.client.opensearch.generic.Requests; import os.org.opensearch.client.transport.httpclient5.ApacheHttpClient5Transport; import os.org.opensearch.client.transport.httpclient5.ApacheHttpClient5TransportBuilder; @@ -293,6 +295,31 @@ void testEnsureHybridSearchPipelineCreatesAndUpdates() throws Exception { assertWeightsInPipeline(updatedPipeline, 0.3, 0.7); } + @Test + void testCheckHybridSearchPipelineNeedsNoStoredPipeline() throws Exception { + deleteSearchPipeline(OpenSearchVectorService.HYBRID_PIPELINE_NAME); + + assertTrue( + vectorService.checkHybridSearchPipeline(0.6, 0.4).isEmpty(), + "Hybrid search runs with the inline pipeline, so the check must pass without a stored one"); + + Optional rejected = vectorService.checkHybridSearchPipeline(0.7, 0.7); + assertTrue(rejected.isPresent(), "OpenSearch rejects RRF weights that do not sum to 1.0"); + assertTrue(rejected.get().contains("sum of weights"), rejected.get()); + } + + private void deleteSearchPipeline(String pipelineName) throws Exception { + // A 404 only means the pipeline was never created. + openSearchClient + .generic() + .execute( + Requests.builder() + .method("DELETE") + .endpoint("/_search/pipeline/" + pipelineName) + .build()) + .close(); + } + @SuppressWarnings("unchecked") private Map getSearchPipeline(String pipelineName) throws Exception { var genericClient = openSearchClient.generic(); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/search/SearchRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/search/SearchRepository.java index 5517d49c753c..a455ffd37fb1 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/search/SearchRepository.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/search/SearchRepository.java @@ -855,7 +855,8 @@ public void updateHybridSearchPipeline(double keywordWeight, double semanticWeig public Optional checkHybridSearchPipeline() { if (vectorIndexService instanceof OpenSearchVectorService openSearchVectorService) { - return openSearchVectorService.checkHybridSearchPipeline(); + double[] weights = resolveHybridWeights(); + return openSearchVectorService.checkHybridSearchPipeline(weights[0], weights[1]); } return Optional.empty(); } diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/search/vector/OpenSearchVectorService.java b/openmetadata-service/src/main/java/org/openmetadata/service/search/vector/OpenSearchVectorService.java index 536a2eed1a32..a2300dabfeeb 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/search/vector/OpenSearchVectorService.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/search/vector/OpenSearchVectorService.java @@ -2,6 +2,7 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; import java.io.IOException; import java.nio.charset.StandardCharsets; @@ -37,7 +38,9 @@ import os.org.opensearch.client.opensearch.core.mget.MultiGetResponseItem; import os.org.opensearch.client.opensearch.generic.Body; import os.org.opensearch.client.opensearch.generic.OpenSearchGenericClient; +import os.org.opensearch.client.opensearch.generic.Request; import os.org.opensearch.client.opensearch.generic.Requests; +import os.org.opensearch.client.opensearch.generic.Response; @Slf4j public class OpenSearchVectorService implements VectorIndexService { @@ -106,6 +109,10 @@ public void close() { */ public static String buildHybridRrfPipelineDefinition( double keywordWeight, double semanticWeight) { + return buildHybridRrfPipeline(keywordWeight, semanticWeight).toString(); + } + + private static ObjectNode buildHybridRrfPipeline(double keywordWeight, double semanticWeight) { var weights = MAPPER.createArrayNode().add(keywordWeight).add(semanticWeight); var combination = MAPPER @@ -122,7 +129,7 @@ public static String buildHybridRrfPipelineDefinition( var pipeline = MAPPER.createObjectNode(); pipeline.set("phase_results_processors", MAPPER.createArrayNode().add(scoreRanker)); - return pipeline.toString(); + return pipeline; } /** @@ -143,50 +150,50 @@ public void ensureHybridSearchPipeline(double keywordWeight, double semanticWeig semanticWeight); } - public Optional checkHybridSearchPipeline() { + /** + * Runs a minimal hybrid query carrying the RRF pipeline inline, as hybrid search does. Never + * reads the stored pipeline: search does not use it, and a prefix-scoped search role cannot read + * it. {@code ignore_unavailable} keeps the check valid before the chunk index exists. + */ + public Optional checkHybridSearchPipeline(double keywordWeight, double semanticWeight) { + Optional error; try { - OpenSearchGenericClient genericClient = client.generic(); - var request = + Request probe = Requests.builder() - .endpoint("/_search/pipeline/" + HYBRID_PIPELINE_NAME) - .method("GET") + .endpoint("/" + getChunkIndexName() + "/_search") + .query(Map.of("ignore_unavailable", "true")) + .method("POST") + .json(buildHybridSearchProbe(keywordWeight, semanticWeight)) .build(); - try (var response = genericClient.execute(request)) { - int status = response.getStatus(); - if (status < 400) { - return Optional.empty(); - } - if (status == 404) { - return Optional.of( - "Hybrid search pipeline '" - + HYBRID_PIPELINE_NAME - + "' not found. Run a reindex to create it."); - } - String detail = - response - .getBody() - .map( - b -> { - try { - String body = new String(b.bodyAsBytes(), StandardCharsets.UTF_8); - return body.length() > 200 ? body.substring(0, 200) : body; - } catch (Exception ignored) { - return ""; - } - }) - .orElse(""); - return Optional.of( - "Unexpected status " - + status - + " when checking hybrid search pipeline '" - + HYBRID_PIPELINE_NAME - + "'." - + (detail.isEmpty() ? "" : " Response: " + detail)); + try (Response response = client.generic().execute(probe)) { + error = + response.getStatus() < 400 + ? Optional.empty() + : Optional.of(describeRejectedProbe(response)); } } catch (Exception e) { - LOG.error("Failed to check hybrid search pipeline '{}'", HYBRID_PIPELINE_NAME, e); - return Optional.of("Failed to check hybrid search pipeline: " + e.toString()); + LOG.error("Failed to run a hybrid search with the RRF pipeline", e); + error = Optional.of("Failed to run a hybrid search with the RRF pipeline: " + e); } + return error; + } + + private static String buildHybridSearchProbe(double keywordWeight, double semanticWeight) { + ObjectNode probe = MAPPER.createObjectNode().put("size", 0); + ArrayNode subQueries = probe.putObject("query").putObject("hybrid").putArray("queries"); + subQueries.addObject().putObject("match_all"); + subQueries.addObject().putObject("match_all"); + probe.set("search_pipeline", buildHybridRrfPipeline(keywordWeight, semanticWeight)); + return probe.toString(); + } + + private static String describeRejectedProbe(Response response) { + String body = response.getBody().map(Body::bodyAsString).orElse(""); + String detail = body.length() > 200 ? body.substring(0, 200) : body; + return "OpenSearch rejected a hybrid search with the RRF pipeline (status " + + response.getStatus() + + ")." + + (detail.isEmpty() ? "" : " Response: " + detail); } @Override diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/search/SearchRepositoryBehaviorTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/search/SearchRepositoryBehaviorTest.java index 8e34fcad30e4..b2247ff33bac 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/search/SearchRepositoryBehaviorTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/search/SearchRepositoryBehaviorTest.java @@ -54,6 +54,7 @@ import org.openmetadata.schema.api.lineage.LineagePaginationInfo; import org.openmetadata.schema.api.lineage.SearchLineageRequest; import org.openmetadata.schema.api.lineage.SearchLineageResult; +import org.openmetadata.schema.api.search.GlobalSettings; import org.openmetadata.schema.api.search.SearchSettings; import org.openmetadata.schema.configuration.LLMConfiguration; import org.openmetadata.schema.configuration.LLMEmbeddingsConfig; @@ -3086,6 +3087,34 @@ void updateHybridSearchPipelineDelegatesOnlyForOpenSearchVectorService() throws repository.updateHybridSearchPipeline(0.8, 0.2); } + @Test + void checkHybridSearchPipelineChecksTheWeightsSearchWillSend() throws Exception { + SearchRepository openSearchRepository = + newRepository( + Map.of(Entity.TABLE, TABLE_MAPPING), + "cluster", + ElasticSearchConfiguration.SearchType.OPENSEARCH, + new NaturalLanguageSearchConfiguration().withSemanticSearchEnabled(true)); + OpenSearchVectorService vectorService = mock(OpenSearchVectorService.class); + when(vectorService.checkHybridSearchPipeline(0.7, 0.7)).thenReturn(Optional.of("rejected")); + setPrivateField(openSearchRepository, "vectorIndexService", vectorService); + SearchSettings adminWeights = + new SearchSettings() + .withGlobalSettings( + new GlobalSettings().withKeywordWeight(0.7).withSemanticWeight(0.7)); + + try (MockedStatic settingsCacheMock = mockStatic(SettingsCache.class)) { + settingsCacheMock + .when(() -> SettingsCache.getSetting(SettingsType.SEARCH_SETTINGS, SearchSettings.class)) + .thenReturn(adminWeights); + + assertEquals(Optional.of("rejected"), openSearchRepository.checkHybridSearchPipeline()); + } + + setPrivateField(openSearchRepository, "vectorIndexService", mock(VectorIndexService.class)); + assertEquals(Optional.empty(), openSearchRepository.checkHybridSearchPipeline()); + } + @Test void requestAndAggregationWrappersDelegateToSearchClient() throws IOException { SearchRequest request = new SearchRequest().withQuery("orders"); diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/search/vector/OpenSearchVectorServiceTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/search/vector/OpenSearchVectorServiceTest.java index 5d06c86ff268..e1d484bbcd47 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/search/vector/OpenSearchVectorServiceTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/search/vector/OpenSearchVectorServiceTest.java @@ -11,8 +11,11 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; import java.io.IOException; import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; import java.util.Optional; @@ -21,13 +24,18 @@ import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import org.openmetadata.schema.entity.data.Table; +import org.openmetadata.service.Entity; +import org.openmetadata.service.search.SearchRepository; import org.openmetadata.service.search.vector.client.EmbeddingClient; import org.openmetadata.service.search.vector.utils.DTOs; import os.org.opensearch.client.opensearch.OpenSearchClient; +import os.org.opensearch.client.opensearch.generic.Body; import os.org.opensearch.client.opensearch.generic.OpenSearchGenericClient; +import os.org.opensearch.client.opensearch.generic.Request; import os.org.opensearch.client.opensearch.generic.Response; class OpenSearchVectorServiceTest { + private static final ObjectMapper MAPPER = new ObjectMapper(); private OpenSearchVectorService vectorService; private OpenSearchClient mockClient; @@ -509,43 +517,67 @@ void testEnsureHybridSearchPipelineWithCustomWeights() throws IOException { } @Test - void testCheckHybridSearchPipelineReturnsEmptyWhenAvailable() throws IOException { - mockOpenSearchResponse("{\"hybrid-rrf\":{}}"); + void testCheckHybridSearchPipelineRunsAHybridQueryWithTheInlinePipeline() throws IOException { + mockOpenSearchResponse("{\"hits\":{\"total\":{\"value\":0},\"hits\":[]}}"); + ArgumentCaptor captor = ArgumentCaptor.forClass(Request.class); - assertTrue( - vectorService.checkHybridSearchPipeline().isEmpty(), - "Pipeline should be available when OpenSearch returns 200"); + Optional result = vectorService.checkHybridSearchPipeline(0.6, 0.4); + + verify(mockGenericClient).execute(captor.capture()); + Request probe = captor.getValue(); + assertTrue(result.isEmpty(), "A query OpenSearch accepts means hybrid search can run"); + assertEquals("POST", probe.getMethod()); + assertEquals("/" + vectorService.getChunkIndexName() + "/_search", probe.getEndpoint()); + assertEquals("true", probe.getParameters().get("ignore_unavailable")); + JsonNode body = MAPPER.readTree(probe.getBody().get().bodyAsString()); + assertEquals(2, body.path("query").path("hybrid").path("queries").size()); + assertEquals( + MAPPER.readTree(OpenSearchVectorService.buildHybridRrfPipelineDefinition(0.6, 0.4)), + body.path("search_pipeline")); } @Test - void testCheckHybridSearchPipelineReturnsErrorOn404() throws IOException { - Response mockResponse = mock(Response.class); - when(mockResponse.getStatus()).thenReturn(404); - when(mockResponse.getBody()).thenReturn(Optional.empty()); - when(mockGenericClient.execute(any())).thenReturn(mockResponse); + void testCheckHybridSearchPipelineStaysInsideTheClusterAliasPrefix() throws IOException { + SearchRepository searchRepository = mock(SearchRepository.class); + when(searchRepository.getClusterAlias()).thenReturn("bi4all-pt"); + Entity.setSearchRepository(searchRepository); + mockOpenSearchResponse("{\"hits\":{\"total\":{\"value\":0},\"hits\":[]}}"); + ArgumentCaptor captor = ArgumentCaptor.forClass(Request.class); + try { + vectorService.checkHybridSearchPipeline(0.6, 0.4); + } finally { + Entity.setSearchRepository(null); + } - Optional result = vectorService.checkHybridSearchPipeline(); - assertTrue(result.isPresent(), "Should return error when pipeline not found"); - assertTrue(result.get().contains("not found"), "Error should mention pipeline not found"); + verify(mockGenericClient).execute(captor.capture()); + assertEquals( + "/bi4all-pt_data_asset_embeddings_chunks/_search", captor.getValue().getEndpoint()); } @Test - void testCheckHybridSearchPipelineReturnsErrorOn5xx() throws IOException { + void testCheckHybridSearchPipelineReportsWhyOpenSearchRejectedTheQuery() throws IOException { + String rejection = + "{\"error\":{\"type\":\"illegal_argument_exception\",\"reason\":\"sum of weights for" + + " combination must be equal to 1.0\"},\"status\":400}"; Response mockResponse = mock(Response.class); - when(mockResponse.getStatus()).thenReturn(500); - when(mockResponse.getBody()).thenReturn(Optional.empty()); + when(mockResponse.getStatus()).thenReturn(400); + when(mockResponse.getBody()) + .thenReturn( + Optional.of(Body.from(rejection.getBytes(StandardCharsets.UTF_8), "application/json"))); when(mockGenericClient.execute(any())).thenReturn(mockResponse); - Optional result = vectorService.checkHybridSearchPipeline(); - assertTrue(result.isPresent(), "Should return error on server error"); - assertTrue(result.get().contains("Unexpected status 500"), "Error should mention status code"); + Optional result = vectorService.checkHybridSearchPipeline(0.7, 0.7); + + assertTrue(result.isPresent(), "A rejected query means hybrid search cannot run"); + assertTrue(result.get().contains("status 400"), "Error should mention the status code"); + assertTrue(result.get().contains("sum of weights"), "Error should carry OpenSearch's reason"); } @Test void testCheckHybridSearchPipelineReturnsErrorOnException() throws IOException { when(mockGenericClient.execute(any())).thenThrow(new IOException("Connection refused")); - Optional result = vectorService.checkHybridSearchPipeline(); + Optional result = vectorService.checkHybridSearchPipeline(0.6, 0.4); assertTrue(result.isPresent(), "Should return error when exception occurs"); assertTrue( result.get().contains("Connection refused"), "Error should include exception message");