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

* Fixes #26824: filter column-grid metadata status on the aggregate row, not per document

The "Has / Missing Metadata" filter in Column Bulk Operations did not hide
rows of other statuses, and page counts drifted per page.

metadataStatus (MISSING/INCOMPLETE/COMPLETE) was pushed down as a per-document
search query, while INCONSISTENT/hasConflicts/hasMissingMetadata were applied
after the aggregator had already paginated and computed totals. But a row's
status is an aggregate over all of a column's occurrences (INCONSISTENT when
they disagree), so a per-document filter re-grouped into rows whose status
differed from the request, and the post-pagination filter shrank the page while
the total stayed unfiltered.

Move every row-level filter (metadataStatus, hasConflicts, hasMissingMetadata)
onto the fully-grouped items, before pagination, with totals derived from the
filtered set. Shared pure helpers (hasRowLevelFilter, matchesRowFilters,
paginateFilteredItems) live on the ColumnAggregator interface; both aggregators
route through a materialized name-enumeration path when such a filter is active,
and the untouched composite/pattern paths still serve unfiltered browsing.

Removing the per-document status query also removes the wildcard/exists query on
flat-object columns.description/columns.tags that crashed ES/OS with
search_phase_execution_exception and had ColumnGridResourceIT disabled; the IT
is re-enabled and its status assertions strengthened to check every returned
row carries the requested status, plus a status+pagination consistency test.

* perf(search): back column-grid status filter with a field-filtered _source scan

Replace the row-filter path's names-agg + per-name top_hits fan-out (~1+N
queries/page) with a single _source scan per field-path group — the same
mechanism the tag/glossary filter already uses — restricting _source to the
column tree and entity-identity fields.

This cuts queries per page from ~1+N to ~1 per field-path group, reuses the
already-verified in-memory pagination, and reads every occurrence of a column
instead of a 100-doc top_hits sample, so the aggregate status can no longer be
misclassified (e.g. COMPLETE vs INCONSISTENT) by under-sampling. Restricting
_source to columns + identity fields keeps the per-entity payload small,
dropping the heavy entity-level derived fields (columnNames/columnNamesFuzzy).

extractMatchingColumnsFromHit gains an includeAllColumns flag (the tag path
passes false; the status scan passes true); the now-unused name-enumeration
constants are removed.

* fix(search): honor columnNamePattern per column in the status-filter scan

The _source-scan row-filter path returned every column of each matching entity
(includeAllColumns=true). Combined with a columnNamePattern the entity-level
name wildcard only decides which entities are scanned — flat-object mapping
can't isolate the matching column — so non-matching columns leaked into the grid
and inflated totalUniqueColumns. Drop columns whose name doesn't contain the
pattern after the scan (mirrors the tag path), in both ES and OS aggregators.

Also correct the row-filter Javadocs: OS still described the old
name-enumeration path, and both overstated "reads all occurrences" — it reads
every occurrence of each scanned entity, up to the 10K-entity scan cap.

Adds an IT for the columnNamePattern + metadataStatus combination.

* refactor(search): extract shared applyColumnNamePattern helper

The per-column name-pattern drop-loop was duplicated verbatim in the ES and OS
row-filter scans. Move it to a static ColumnAggregator.applyColumnNamePattern
helper (alongside the other shared row-filter helpers) so the two backends
can't drift in pattern semantics.

Author: sonika-shah

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) {}