Fixes #26824: filter column bulk operations metadata status on the aggregate row

openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/ColumnGridResourceIT.java

@@ -13,7 +13,6 @@
 import java.time.Duration;
 import java.util.List;
 import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.parallel.Execution;
@@ -53,17 +52,10 @@
 import org.openmetadata.sdk.fluent.Tables;
 import org.openmetadata.sdk.network.HttpMethod;
 
-// TEMPORARILY DISABLED — the metadataStatus aggregation on this endpoint reproducibly fails
-// with [search_phase_execution_exception] all shards failed on both postgres+ES+redis (single
-// failure on test_getColumnGrid_withMetadataStatusIncomplete) AND postgres+OpenSearch (the same
-// query crashes the OS container, then 15 follow-up tests in the class fail with Connection
-// refused). Same behavior on PR #28100 with and without the cache changes, so it is a
-// pre-existing aggregator bug, not a cache regression. The ES Java client swallows the
-// underlying `caused_by`, so root-causing the actual ES-side error requires response-body
-// logging that is not wired up yet. Re-enable once the underlying aggregator/index-mapping
-// issue is fixed in a follow-up. See PR #28100 history and CI run 25940411417 for context.
-@Disabled(
-    "ColumnGrid metadataStatus aggregation crashes ES/OS — pre-existing flake, follow-up needed")
+// Re-enabled with #26824: the metadataStatus crash came from the per-document filter query
+// (wildcard/exists on flat-object columns.description/columns.tags), which ES 7.17 and OpenSearch
+// rejected with `search_phase_execution_exception ... all shards failed`. That push-down is gone —
+// status is now filtered on the aggregate grouped item — so the crashing query no longer runs.
 @Execution(ExecutionMode.CONCURRENT)
 @ExtendWith(TestNamespaceExtension.class)
 public class ColumnGridResourceIT {
@@ -314,6 +306,11 @@ void test_getColumnGrid_withMetadataStatusMissing(TestNamespace ns) throws Excep
 
     assertNotNull(response);
     assertNotNull(response.getColumns());
+    assertAllRowsHaveStatus(response, MetadataStatus.MISSING);
+    assertEquals(
+        response.getColumns().size(),
+        response.getTotalUniqueColumns(),
+        "totalUniqueColumns must reflect the filtered set, not the unfiltered total");
   }
 
   @Test
@@ -328,6 +325,9 @@ void test_getColumnGrid_withMetadataStatusComplete(TestNamespace ns) throws Exce
 
     assertNotNull(response);
     assertNotNull(response.getColumns());
+    assertFalse(response.getColumns().isEmpty(), "the COMPLETE column should be returned");
+    // The reported bug (#26824): COMPLETE must not surface MISSING/INCOMPLETE/INCONSISTENT rows.
+    assertAllRowsHaveStatus(response, MetadataStatus.COMPLETE);
   }
 
   @Test
@@ -342,6 +342,8 @@ void test_getColumnGrid_withMetadataStatusIncomplete(TestNamespace ns) throws Ex
 
     assertNotNull(response);
     assertNotNull(response.getColumns());
+    assertFalse(response.getColumns().isEmpty(), "the INCOMPLETE column should be returned");
+    assertAllRowsHaveStatus(response, MetadataStatus.INCOMPLETE);
   }
 
   @Test
@@ -422,6 +424,125 @@ void test_getColumnGrid_withMetadataStatusInconsistent(TestNamespace ns) throws
 
     assertNotNull(response);
     assertNotNull(response.getColumns());
+    assertFalse(response.getColumns().isEmpty(), "the INCONSISTENT column should be returned");
+    assertAllRowsHaveStatus(response, MetadataStatus.INCONSISTENT);
+    assertTrue(
+        response.getColumns().stream().allMatch(ColumnGridItem::getHasVariations),
+        "INCONSISTENT rows have metadata variations across occurrences");
+  }
+
+  @Test
+  void test_getColumnGrid_metadataStatusPaginationCountsAreConsistent(TestNamespace ns)
+      throws Exception {
+    OpenMetadataClient client = SdkClients.adminClient();
+    DatabaseService service = DatabaseServiceTestFactory.createPostgres(ns);
+    DatabaseSchema schema = DatabaseSchemaTestFactory.createSimple(ns, service);
+
+    // Three COMPLETE columns and one MISSING column in the same service.
+    for (int i = 0; i < 3; i++) {
+      Column complete =
+          Columns.build(ns.prefix("paged_complete_" + i))
+              .withType(ColumnDataType.BIGINT)
+              .withDescription("has description")
+              .withTags(List.of(new TagLabel().withTagFQN("PII.Sensitive")))
+              .create();
+      Tables.create()
+          .name(ns.prefix("paged_table_" + i))
+          .inSchema(schema.getFullyQualifiedName())
+          .withColumns(List.of(complete))
+          .execute();
+    }
+    Column missing =
+        Columns.build(ns.prefix("paged_missing")).withType(ColumnDataType.BIGINT).create();
+    Tables.create()
+        .name(ns.prefix("paged_table_missing"))
+        .inSchema(schema.getFullyQualifiedName())
+        .withColumns(List.of(missing))
+        .execute();
+
+    waitForSearchIndexRefresh(ns);
+
+    ColumnGridResponse page1 =
+        getColumnGrid(
+            client,
+            "size=2&entityTypes=table&metadataStatus=COMPLETE&serviceName=" + service.getName());
+
+    // totalUniqueColumns must count only the 3 COMPLETE columns (not 4), and the page must respect
+    // the requested size — the pagination half of #26824.
+    assertEquals(3, page1.getTotalUniqueColumns());
+    assertEquals(2, page1.getColumns().size());
+    assertAllRowsHaveStatus(page1, MetadataStatus.COMPLETE);
+    assertNotNull(page1.getCursor(), "a second page of COMPLETE columns remains");
+
+    ColumnGridResponse page2 =
+        getColumnGrid(
+            client,
+            "size=2&entityTypes=table&metadataStatus=COMPLETE&serviceName="
+                + service.getName()
+                + "&cursor="
+                + URLEncoder.encode(page1.getCursor(), StandardCharsets.UTF_8));
+
+    assertEquals(3, page2.getTotalUniqueColumns());
+    assertEquals(1, page2.getColumns().size(), "the last page holds the remaining COMPLETE column");
+    assertAllRowsHaveStatus(page2, MetadataStatus.COMPLETE);
+  }
+
+  @Test
+  void test_getColumnGrid_metadataStatusWithColumnNamePattern(TestNamespace ns) throws Exception {
+    OpenMetadataClient client = SdkClients.adminClient();
+    DatabaseService service = DatabaseServiceTestFactory.createPostgres(ns);
+    DatabaseSchema schema = DatabaseSchemaTestFactory.createSimple(ns, service);
+
+    // Two COMPLETE columns; only one name contains "alpha".
+    String matchName = ns.prefix("alpha_amount");
+    String otherName = ns.prefix("zzz_other");
+    for (String colName : List.of(matchName, otherName)) {
+      Column col =
+          Columns.build(colName)
+              .withType(ColumnDataType.BIGINT)
+              .withDescription("has description")
+              .withTags(List.of(new TagLabel().withTagFQN("PII.Sensitive")))
+              .create();
+      Tables.create()
+          .name(ns.prefix("pat_" + colName))
+          .inSchema(schema.getFullyQualifiedName())
+          .withColumns(List.of(col))
+          .execute();
+    }
+    waitForSearchIndexRefresh(ns);
+
+    ColumnGridResponse response =
+        getColumnGrid(
+            client,
+            "entityTypes=table&metadataStatus=COMPLETE&columnNamePattern=alpha&serviceName="
+                + service.getName());
+
+    // Combining columnNamePattern with a status filter must honor the pattern per column:
+    // the non-matching "zzz_other" column must not leak in (regression for the _source-scan path).
+    assertNotNull(response);
+    assertFalse(response.getColumns().isEmpty(), "the matching COMPLETE column should be returned");
+    assertAllRowsHaveStatus(response, MetadataStatus.COMPLETE);
+    assertTrue(
+        response.getColumns().stream()
+            .allMatch(c -> c.getColumnName().toLowerCase().contains("alpha")),
+        "only columns whose name matches the pattern should be returned");
+    assertEquals(
+        response.getColumns().size(),
+        response.getTotalUniqueColumns(),
+        "totalUniqueColumns must not include pattern-mismatched columns");
+  }
+
+  /**
+   * Every returned row must carry the requested aggregate status — the core guarantee of #26824
+   * (before the fix a COMPLETE/INCOMPLETE filter leaked rows of other statuses).
+   */
+  private void assertAllRowsHaveStatus(ColumnGridResponse response, MetadataStatus expected) {
+    for (ColumnGridItem item : response.getColumns()) {
+      assertEquals(
+          expected,
+          item.getMetadataStatus(),
+          "column '" + item.getColumnName() + "' should have status " + expected);
+    }
   }
 
   @Test
@@ -1424,6 +1545,7 @@ private DatabaseService createTableWithFullMetadata(TestNamespace ns) {
         Columns.build("full_metadata_id")
             .withType(ColumnDataType.BIGINT)
             .withDescription("Primary key with description")
+            .withTags(List.of(new TagLabel().withTagFQN("PII.Sensitive")))
             .create();
 
     Tables.create()

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

@@ -40,7 +40,6 @@
 import org.openmetadata.schema.EntityInterface;
 import org.openmetadata.schema.api.data.BulkColumnUpdatePreview;
 import org.openmetadata.schema.api.data.BulkColumnUpdateRequest;
-import org.openmetadata.schema.api.data.ColumnGridItem;
 import org.openmetadata.schema.api.data.ColumnGridResponse;
 import org.openmetadata.schema.api.data.ColumnMetadata;
 import org.openmetadata.schema.api.data.ColumnOccurrence;
@@ -97,39 +96,10 @@ public ColumnRepository(Authorizer authorizer, SearchClient searchClient) {
   public ColumnGridResponse getColumnGridPaginated(
       SecurityContext securityContext, ColumnAggregator.ColumnAggregationRequest request)
       throws IOException {
-    ColumnGridResponse response = columnAggregator.aggregateColumns(request);
-
-    if (Boolean.TRUE.equals(request.getHasConflicts())) {
-      response.setColumns(
-          response.getColumns().stream()
-              .filter(ColumnGridItem::getHasVariations)
-              .collect(Collectors.toList()));
-    }
-
-    if (Boolean.TRUE.equals(request.getHasMissingMetadata())) {
-      response.setColumns(
-          response.getColumns().stream()
-              .filter(this::hasMissingMetadata)
-              .collect(Collectors.toList()));
-    }
-
-    // Filter by INCONSISTENT status (requires post-aggregation filtering)
-    if ("INCONSISTENT".equalsIgnoreCase(request.getMetadataStatus())) {
-      response.setColumns(
-          response.getColumns().stream()
-              .filter(ColumnGridItem::getHasVariations)
-              .collect(Collectors.toList()));
-    }
-
-    return response;
-  }
-
-  private boolean hasMissingMetadata(ColumnGridItem item) {
-    return item.getGroups().stream()
-        .anyMatch(
-            group ->
-                (group.getDescription() == null || group.getDescription().isEmpty())
-                    || (group.getTags() == null || group.getTags().isEmpty()));
+    // Row-level filters (metadataStatus / hasConflicts / hasMissingMetadata) are applied inside the
+    // aggregator over the fully-grouped columns, before pagination, so page counts and per-page
+    // size stay correct (#26824). Nothing to post-process here.
+    return columnAggregator.aggregateColumns(request);
   }
 
   public Column getColumnByFQN(

openmetadata-service/src/main/java/org/openmetadata/service/resources/columns/ColumnResource.java

@@ -319,13 +319,15 @@ public Response getColumnGrid(
           boolean hasMissingMetadata,
       @Parameter(
               description =
-                  "Filter by metadata status: MISSING (no description AND no tags), "
+                  "Filter by aggregate metadata status of a column across all its occurrences: "
+                      + "MISSING (no description AND no tags), "
                       + "INCOMPLETE (has description OR tags, but not both), "
-                      + "COMPLETE (has both description AND tags)",
+                      + "COMPLETE (has both description AND tags), "
+                      + "INCONSISTENT (occurrences disagree on description/tags)",
               schema =
                   @Schema(
                       type = "string",
-                      allowableValues = {"MISSING", "INCOMPLETE", "COMPLETE"}))
+                      allowableValues = {"MISSING", "INCOMPLETE", "COMPLETE", "INCONSISTENT"}))
           @QueryParam("metadataStatus")
           String metadataStatus,
       @Parameter(

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

@@ -16,9 +16,13 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Base64;
+import java.util.Comparator;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
+import org.openmetadata.schema.api.data.ColumnGridItem;
 import org.openmetadata.schema.api.data.ColumnGridResponse;
 import org.openmetadata.schema.utils.JsonUtils;
 import org.slf4j.Logger;
@@ -115,6 +119,107 @@ static int toIntSaturating(long value) {
     return (int) value;
   }
 
+  /**
+   * True if a request carries a row-level filter — one that acts on the aggregate status of a
+   * grouped column (metadataStatus, hasConflicts, hasMissingMetadata) rather than on individual
+   * documents. These cannot be pushed into the search query (the aggregate status is only known
+   * after grouping occurrences), so they are applied via {@link #paginateFilteredItems}.
+   */
+  static boolean hasRowLevelFilter(ColumnAggregationRequest request) {
+    return !nullOrEmptyStr(request.getMetadataStatus())
+        || Boolean.TRUE.equals(request.getHasConflicts())
+        || Boolean.TRUE.equals(request.getHasMissingMetadata());
+  }
+
+  /** True if the grouped column satisfies every active row-level filter on the request. */
+  static boolean matchesRowFilters(ColumnGridItem item, ColumnAggregationRequest request) {
+    if (Boolean.TRUE.equals(request.getHasConflicts())
+        && !Boolean.TRUE.equals(item.getHasVariations())) {
+      return false;
+    }
+    if (Boolean.TRUE.equals(request.getHasMissingMetadata()) && !itemHasMissingMetadata(item)) {
+      return false;
+    }
+    String status = request.getMetadataStatus();
+    if (!nullOrEmptyStr(status)) {
+      String itemStatus = item.getMetadataStatus() != null ? item.getMetadataStatus().value() : "";
+      return status.trim().equalsIgnoreCase(itemStatus);
+    }
+    return true;
+  }
+
+  /**
+   * Drop columns whose name doesn't contain the request's {@code columnNamePattern}
+   * (case-insensitive). The name wildcard in the search query only scopes which entities are
+   * scanned; flat-object mapping can't isolate the matching column, so the pattern is enforced per
+   * column here. Shared by the ES and OS row-filter scans so their pattern semantics can't drift.
+   */
+  static void applyColumnNamePattern(
+      Map<String, ?> columnsByName, ColumnAggregationRequest request) {
+    if (nullOrEmptyStr(request.getColumnNamePattern())) {
+      return;
+    }
+    String pattern = request.getColumnNamePattern().toLowerCase(Locale.ROOT);
+    columnsByName.keySet().removeIf(name -> !name.toLowerCase(Locale.ROOT).contains(pattern));
+  }
+
+  /** A column has missing metadata if any of its groups lacks a description or tags. */
+  static boolean itemHasMissingMetadata(ColumnGridItem item) {
+    if (item.getGroups() == null) {
+      return true;
+    }
+    return item.getGroups().stream()
+        .anyMatch(
+            group ->
+                (group.getDescription() == null || group.getDescription().isEmpty())
+                    || (group.getTags() == null || group.getTags().isEmpty()));
+  }
+
+  /**
+   * Apply row-level filters to the fully-grouped column list and paginate the result in memory.
+   * Filtering the aggregate items (not documents) is what makes a "Complete"/"Incomplete"/… filter
+   * return only rows whose displayed status matches, and computing totals from the filtered set is
+   * what keeps the page count and per-page size correct (issue #26824). Ordering is by column name
+   * (case-insensitive) so the offset cursor is stable across pages.
+   */
+  static ColumnGridResponse paginateFilteredItems(
+      List<ColumnGridItem> allItems, ColumnAggregationRequest request) {
+    List<ColumnGridItem> filtered =
+        allItems.stream()
+            .filter(item -> matchesRowFilters(item, request))
+            .sorted(
+                Comparator.comparing(
+                    ColumnGridItem::getColumnName,
+                    Comparator.nullsLast(String.CASE_INSENSITIVE_ORDER)))
+            .toList();
+
+    int totalUniqueColumns = filtered.size();
+    int totalOccurrences =
+        filtered.stream()
+            .mapToInt(item -> item.getTotalOccurrences() != null ? item.getTotalOccurrences() : 0)
+            .sum();
+
+    int offset = decodeSearchOffset(request.getCursor());
+    int pageSize = request.getSize();
+    int fromIndex = Math.min(offset, totalUniqueColumns);
+    int toIndex = Math.min(offset + pageSize, totalUniqueColumns);
+
+    List<ColumnGridItem> page = new ArrayList<>(filtered.subList(fromIndex, toIndex));
+    boolean hasMore = toIndex < totalUniqueColumns;
+    String cursor = hasMore ? encodeSearchOffset(toIndex) : null;
+
+    ColumnGridResponse response = new ColumnGridResponse();
+    response.setColumns(page);
+    response.setTotalUniqueColumns(totalUniqueColumns);
+    response.setTotalOccurrences(totalOccurrences);
+    response.setCursor(cursor);
+    return response;
+  }
+
+  private static boolean nullOrEmptyStr(String s) {
+    return s == null || s.isBlank();
+  }
+
   /** Phase 1 result: matching column names and the total doc_count summed across buckets. */
   record NamesWithCount(List<String> names, long totalDocCount) {}