Fixes #35317: check hybrid search with a query, not the stored pipeline (backport 2.0 #35321)

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<String> 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<String, Object> getSearchPipeline(String pipelineName) throws Exception {
     var genericClient = openSearchClient.generic();

openmetadata-service/src/main/java/org/openmetadata/service/search/SearchRepository.java

@@ -855,7 +855,8 @@ public void updateHybridSearchPipeline(double keywordWeight, double semanticWeig
 
   public Optional<String> checkHybridSearchPipeline() {
     if (vectorIndexService instanceof OpenSearchVectorService openSearchVectorService) {
-      return openSearchVectorService.checkHybridSearchPipeline();
+      double[] weights = resolveHybridWeights();
+      return openSearchVectorService.checkHybridSearchPipeline(weights[0], weights[1]);
     }
     return Optional.empty();
   }

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<String> 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<String> checkHybridSearchPipeline(double keywordWeight, double semanticWeight) {
+    Optional<String> 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

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<SettingsCache> 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");

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<Request> captor = ArgumentCaptor.forClass(Request.class);
 
-    assertTrue(
-        vectorService.checkHybridSearchPipeline().isEmpty(),
-        "Pipeline should be available when OpenSearch returns 200");
+    Optional<String> 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<Request> captor = ArgumentCaptor.forClass(Request.class);
+    try {
+      vectorService.checkHybridSearchPipeline(0.6, 0.4);
+    } finally {
+      Entity.setSearchRepository(null);
+    }
 
-    Optional<String> 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<String> 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<String> 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<String> result = vectorService.checkHybridSearchPipeline();
+    Optional<String> 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");