Fixes #27549: Respect from/size pagination in /search/fieldQuery (#27590)

* Add scheduled workflow to sync fork main with upstream

* Remove fork upstream sync workflow

* Fix fieldQuery pagination by propagating from and size

* Avoid truncating ingestion pipelines in DataInsights lookup

---------

Co-authored-by: Mohit Yadav <105265192+mohityadav766@users.noreply.github.com>
Co-authored-by: Pere Miquel Brull <peremiquelbrull@gmail.com>

Author: 3 people

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/DataInsightSystemChartRepository.java

@@ -156,6 +156,8 @@ public static String getLiveSearchIndex(String index) {
    */
   private List<Map> getIngestionPipelineStatus(String serviceName) {
     List<Map> combinedStatus = new ArrayList<>();
+    final int pageSize = 100;
+    final int maxResults = 5000;
 
     try {
       if (serviceName == null || serviceName.trim().isEmpty()) {
@@ -175,16 +177,27 @@ private List<Map> getIngestionPipelineStatus(String serviceName) {
       SearchClient searchClient = Entity.getSearchRepository().getSearchClient();
       if (searchClient != null) {
         try {
-          // Search for ingestion pipelines with the service name
-          var response =
-              searchClient.searchByField(
-                  "service.name.keyword", serviceName, INGESTION_PIPELINE, false);
+          // Search for ingestion pipelines with the service name, paging through all results.
+          for (int from = 0; from < maxResults; from += pageSize) {
+            var response =
+                searchClient.searchByField(
+                    "service.name.keyword", serviceName, INGESTION_PIPELINE, false, from, pageSize);
+
+            if (response == null || response.getStatus() != 200) {
+              break;
+            }
 
-          if (response != null && response.getStatus() == 200) {
-            // Parse the response to extract pipeline information
             String responseBody =
                 (String) ((OutboundJaxrsResponse) response).getContext().getEntity();
-            combinedStatus.addAll(parseIngestionPipelineResponse(responseBody));
+            List<Map> pageStatuses = parseIngestionPipelineResponse(responseBody);
+            if (pageStatuses.isEmpty()) {
+              break;
+            }
+
+            combinedStatus.addAll(pageStatuses);
+            if (pageStatuses.size() < pageSize) {
+              break;
+            }
           }
         } catch (Exception e) {
           LOG.error("Error searching for ingestion pipelines for service: {}", serviceName, e);

openmetadata-service/src/main/java/org/openmetadata/service/resources/search/SearchResource.java

@@ -584,10 +584,18 @@ public Response searchByField(
       @Parameter(description = "Filter documents by deleted param. By default deleted is false")
           @DefaultValue("false")
           @QueryParam("deleted")
-          boolean deleted)
+          boolean deleted,
+      @Parameter(description = "From field to paginate the results, defaults to 0")
+          @DefaultValue("0")
+          @QueryParam("from")
+          int from,
+      @Parameter(description = "Size field to limit the no.of results returned, defaults to 10")
+          @DefaultValue("10")
+          @QueryParam("size")
+          int size)
       throws IOException {
 
-    return searchRepository.searchByField(fieldName, fieldValue, index, deleted);
+    return searchRepository.searchByField(fieldName, fieldValue, index, deleted, from, size);
   }
 
   @GET

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

@@ -67,10 +67,13 @@ Response previewSearch(
    * @param fieldValue the value to match (supports wildcards)
    * @param index the index to search in
    * @param deleted whether to include deleted entities
+   * @param from starting position for pagination
+   * @param size maximum number of results to return
    * @return response containing matching entities
    * @throws IOException if search execution fails
    */
-  Response searchByField(String fieldName, String fieldValue, String index, Boolean deleted)
+  Response searchByField(
+      String fieldName, String fieldValue, String index, Boolean deleted, int from, int size)
       throws IOException;
 
   /**

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

@@ -2927,9 +2927,10 @@ public SearchLineageResult searchLineageForExport(
             .withIsConnectedVia(isConnectedVia(entityType)));
   }
 
-  public Response searchByField(String fieldName, String fieldValue, String index, Boolean deleted)
+  public Response searchByField(
+      String fieldName, String fieldValue, String index, Boolean deleted, int from, int size)
       throws IOException {
-    return searchClient.searchByField(fieldName, fieldValue, index, deleted);
+    return searchClient.searchByField(fieldName, fieldValue, index, deleted, from, size);
   }
 
   public Response aggregate(AggregationRequest request) throws IOException {

openmetadata-service/src/main/java/org/openmetadata/service/search/elasticsearch/ElasticSearchClient.java

@@ -461,9 +461,10 @@ public Response searchBySourceUrl(String sourceUrl) throws IOException {
   }
 
   @Override
-  public Response searchByField(String fieldName, String fieldValue, String index, Boolean deleted)
+  public Response searchByField(
+      String fieldName, String fieldValue, String index, Boolean deleted, int from, int size)
       throws IOException {
-    return searchManager.searchByField(fieldName, fieldValue, index, deleted);
+    return searchManager.searchByField(fieldName, fieldValue, index, deleted, from, size);
   }
 
   @Override

openmetadata-service/src/main/java/org/openmetadata/service/search/elasticsearch/ElasticSearchSearchManager.java

@@ -169,7 +169,8 @@ public Response searchBySourceUrl(String sourceUrl) throws IOException {
   }
 
   @Override
-  public Response searchByField(String fieldName, String fieldValue, String index, Boolean deleted)
+  public Response searchByField(
+      String fieldName, String fieldValue, String index, Boolean deleted, int from, int size)
       throws IOException {
     if (!isClientAvailable) {
       throw new IOException("Elasticsearch client is not available");
@@ -179,6 +180,8 @@ public Response searchByField(String fieldName, String fieldValue, String index,
         SearchRequest.of(
             s ->
                 s.index(Entity.getSearchRepository().getIndexOrAliasName(index))
+                    .from(from)
+                    .size(size)
                     .query(
                         q ->
                             q.bool(

openmetadata-service/src/main/java/org/openmetadata/service/search/opensearch/OpenSearchClient.java

@@ -447,9 +447,10 @@ public Response searchSchemaEntityRelationship(
   }
 
   @Override
-  public Response searchByField(String fieldName, String fieldValue, String index, Boolean deleted)
+  public Response searchByField(
+      String fieldName, String fieldValue, String index, Boolean deleted, int from, int size)
       throws IOException {
-    return searchManager.searchByField(fieldName, fieldValue, index, deleted);
+    return searchManager.searchByField(fieldName, fieldValue, index, deleted, from, size);
   }
 
   @Override

openmetadata-service/src/main/java/org/openmetadata/service/search/opensearch/OpenSearchSearchManager.java

@@ -189,7 +189,8 @@ public Response searchBySourceUrl(String sourceUrl) throws IOException {
   }
 
   @Override
-  public Response searchByField(String fieldName, String fieldValue, String index, Boolean deleted)
+  public Response searchByField(
+      String fieldName, String fieldValue, String index, Boolean deleted, int from, int size)
       throws IOException {
     if (!isClientAvailable) {
       throw new IOException("OpenSearch client is not available");
@@ -199,6 +200,8 @@ public Response searchByField(String fieldName, String fieldValue, String index,
         SearchRequest.of(
             s ->
                 s.index(Entity.getSearchRepository().getIndexOrAliasName(index))
+                    .from(from)
+                    .size(size)
                     .query(
                         q ->
                             q.bool(

openmetadata-service/src/test/java/org/openmetadata/service/search/SearchRepositoryBehaviorTest.java

@@ -2429,7 +2429,7 @@ void reportingWrappersDelegateToSearchClient() throws IOException {
                 .SearchSchemaEntityRelationshipResult();
 
     when(filter.getCondition(Entity.TABLE)).thenReturn("deleted = false");
-    when(searchClient.searchByField("name", "orders", "table", false)).thenReturn(response);
+    when(searchClient.searchByField("name", "orders", "table", false, 0, 10)).thenReturn(response);
     when(searchClient.aggregate("query", Entity.TABLE, searchAggregation, "deleted = false"))
         .thenReturn(aggregationResult);
     when(searchClient.genericAggregation("query", "table", searchAggregation)).thenReturn(report);
@@ -2449,7 +2449,7 @@ void reportingWrappersDelegateToSearchClient() throws IOException {
     when(searchClient.getSchemaEntityRelationship("svc.db.schema", "{}", "*", 1, 2, 3, 4, false))
         .thenReturn(schemaResult);
 
-    assertSame(response, repository.searchByField("name", "orders", "table", false));
+    assertSame(response, repository.searchByField("name", "orders", "table", false, 0, 10));
     assertSame(
         aggregationResult, repository.aggregate("query", Entity.TABLE, searchAggregation, filter));
     assertSame(report, repository.genericAggregation("query", "table", searchAggregation));