fix(profiler): N+1 / missing-index regression on /tables/.../columns?fields=profile (#3488) (#27746)

sonika-shah · claude · web-flow · commit 17c3b8b9b2e7 · 2026-05-14T06:56:39.000-07:00
* fix(profiler): N+1 / missing-index regression on /tables/.../columns?fields=profile (#3488) Root cause ---------- The 1.9.9 migration introduced two separate index regressions on `profiler_data_time_series`: 1. **PostgreSQL**: `schemaChanges.sql` explicitly dropped the unique constraint `profiler_data_time_series_unique_hash_extension_ts` (entityFQNHash, extension, operation, timestamp) to allow altering the generated `operation` column expression, but never recreated it. After the migration the table kept only the `(extension, timestamp)` index, which is useless for queries filtering by `entityFQNHash`. 2. **MySQL/both**: `postDataMigrationSQLScript.sql` created temporary indexes (idx_pdts_entityFQNHash, idx_pdts_composite, etc.) for its bulk UPDATE pass and then dropped **all** of them, including the only index covering `entityFQNHash`. The batch query issued by `getLatestExtensionsBatch()` when `fields=profile` is requested: SELECT entityFQNHash, MAX(timestamp) FROM profiler_data_time_series WHERE entityFQNHash IN (...N hashes...) AND extension = 'table.columnProfile' GROUP BY entityFQNHash required an `(entityFQNHash, extension, timestamp)` index. Without it the database performs a full table scan. On production deployments with millions of profiler rows this caused 100+ second response times (Grafana: 106 770 ms; 99 % in DB; 93 dbOps). Without `profile` in the fields param the same endpoint returned in ~150-220 ms. A secondary N+1 bug existed independently of the index: `customMetrics` in fields called `getCustomMetrics(table, column)` once per paginated column, issuing up to N identical queries against `entity_extension` and then filtering in Java. Fix --- * **migration 2.0.2** (MySQL + PostgreSQL): `CREATE INDEX IF NOT EXISTS idx_pdts_fqnhash_ext_ts ON profiler_data_time_series(entityFQNHash, extension, timestamp)`. The `IF NOT EXISTS` guard makes the migration safe to re-run and handles both upgrade and fresh-install paths. * **`getTableColumnsInternal`** — `customMetrics` block: fetch all column custom metrics for the table in one query, group by column name in Java, then distribute. Reduces N queries to 1. * **`getTableColumnsInternal`** — `profile` block: skip the duplicate `populateEntityFieldTags` call when `tags` was already fetched earlier in the same request, saving one prefix-scan on `tag_usage` per request. Related: PR #26855 (fixed N+1 tag queries on the list-tables path but left the profiler-index and customMetrics N+1 untouched on the columns sub-path). * fix(profiler): restore unique constraint on profiler_data_time_series + batch column extension/customMetrics fetch Move the migration from 2.0.2/ to 1.12.8/ and switch from a non-unique covering index to restoring the original unique constraint dropped in 1.9.9. The two-phase CREATE UNIQUE INDEX CONCURRENTLY + ADD CONSTRAINT USING INDEX pattern avoids the ACCESS EXCLUSIVE lock on the hot profiler_data_time_series table during the upgrade. Closes the 1.9.9 regression and brings Postgres back in line with MySQL (which never lost the constraint). The leading (entityFQNHash, extension) prefix serves the column-profile batch query — same shape MySQL has been running without 504s. MySQL needs no migration. Java side, eliminates two more N+1 patterns that compound the latency at customer scale: * getTableColumnsInternal extension block: replaced per-column getColumnExtension() loop with a single getExtensionsByJsonSchema() call, grouped by column FQN-hash in Java. * searchTableColumnsInternal customMetrics block: applied the same batch-fetch pattern already used in getTableColumnsInternal, replacing per-column getCustomMetrics() with one getExtensions() call. New DAO method on EntityExtensionDAO: getExtensionsByJsonSchema(id, jsonSchema) — selects extensions for a table id filtered by the jsonschema discriminator. Required because column extensions are stored with MD5-hashed extension keys and have no shared prefix the existing getExtensions(id, prefix) could use. * chore(profiler): address review feedback — empty-list literal + accurate test comments * Replace `new ArrayList<>()` default in `metricsByColumn.getOrDefault(...)` with `List.of()` at both call sites in `TableRepository` (getTableColumnsInternal and searchTableColumnsInternal). `getOrDefault` evaluates its default eagerly, so the new ArrayList allocates per-column even when the key is found — unnecessary work on a hot path. * Reword two stale test comments in `test_getColumnsWithProfileField_correctnessAndNoBatchRegression`: - "all four field combinations" → "the three field combinations exercised below" - "(c) duplicate populateEntityFieldTags must not run twice" → describe the observable contract the assertions actually verify (tags + profile both present), not the internal call count. * fix(profiler): force outer index scan in getLatestExtensionsBatch by pushing IN list to the join The getLatestExtensionsBatch query was the right shape for correctness but the planner — on Postgres at customer scale, with the new unique constraint in place — was still choosing a parallel sequential scan over the full profiler_data_time_series table for the outer side of the JOIN, rather than a merge join with index scan on both sides. Inner subquery: filtered by `entityFQNHash IN (...)`, used the index. Outer: only filtered by `p.extension = :extension`, no IN list, planner couldn't infer the transitive constraint that p.entityFQNHash must equal one of the inner hashes (because it's enforced through the JOIN ON clause, not a WHERE predicate). Result: full table scan reading 6.7M+ rows even when the actual answer is 23 rows. Adding the redundant `AND p.entityFQNHash IN (<entityFQNHashes>)` to the outer WHERE makes the constraint explicit. The result set is unchanged (implied by the join condition), but the planner can now use the unique index for the outer access too. Verified on the AUT dump (6.94M-row pdts): EXPLAIN of the batch query: 7,234ms → 79ms (Hash Join + Parallel Seq Scan → Merge Join + Index Only Scan). Live API /columns?fields=profile&include=all: 6-36 seconds → 22-28ms (warm) / 1.9s (very first call). 250-1000x improvement, depending on cache state. Same SQL works on both engines; no @ConnectionAwareSqlQuery split needed. * test(profiler): shorten classification/tag fixture names in IT to fit varchar(256) The IT fixture for test_getColumnsWithProfileField_correctnessAndNoBatchRegression was building a tagFQN of `<classification>.<tag>` where each part went through TestNamespace.prefix(). With the descriptive method name (62 chars) + class name (15 chars) + namespace UUID (32 chars) plus the `profile_test_cls` / `profile_test_tag` base names (16 chars each), the resulting tagFQN was 263 characters — over the tag_usage.tagFQN VARCHAR(256) limit: ERROR: value too long for type character varying(256) Shorten the fixture base names from `profile_test_cls`/`profile_test_tag` to `cls`/`tag`. The namespace prefix already encodes test isolation (class + method + UUID), so the base name doesn't need to repeat that context. New tagFQN length: 237 chars (cls__<32>__TableResourceIT__<62>.tag__<32>__TableResourceIT__<62>), comfortably under 256. * fix(table): include extensionKey in column-extension deserialize warn log Addresses gitar-bot review on PR #27746: the warning log on failed column- extension deserialization only had table.getId(), so operators could not pinpoint which row was bad. Add record.extensionName() (the entity_extension row key) to the log. No extra iteration - record is already in scope inside the catch. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * chore(migration): move profiler unique-constraint migration to 1.12.9 1.12.8 was already published with the PII classification fix from #27910. Move the profiler_data_time_series unique-constraint restore (this PR's postgres migration) to 1.12.9 so customers upgrading past the published 1.12.8 still pick it up. Add a MySQL placeholder schemaChanges.sql for 1.12.9 consistent with the 1.12.7 convention — MySQL was unaffected by the 1.9.9 regression (MODIFY COLUMN re-evaluates generated expressions in place without touching the constraint, so MySQL still has the constraint from 1.1.5). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(table): extract batchFetchCustomMetricsByColumn helper Addresses PR #27746 Copilot review: - Dedupe custom-metric batch logic between getTableColumnsInternal and searchTableColumnsInternal. - Reword IT inline comment to reflect what the test actually validates (completes within timeout + correct profiles) instead of claiming it inspects query plans. --------- Co-authored-by: Claude <noreply@anthropic.com>
diff --git a/bootstrap/sql/migrations/native/1.12.9/mysql/schemaChanges.sql b/bootstrap/sql/migrations/native/1.12.9/mysql/schemaChanges.sql
@@ -0,0 +1,5 @@
+-- Placeholder for 1.12.9 MySQL schema changes
+-- The Postgres-side fix for collate#3488 has no MySQL counterpart: MySQL's
+-- 1.1.5 unique constraint on profiler_data_time_series was never dropped
+-- (MODIFY COLUMN re-evaluates generated expressions in place), so the
+-- regression that hit Postgres did not affect MySQL.
diff --git a/bootstrap/sql/migrations/native/1.12.9/postgres/schemaChanges.sql b/bootstrap/sql/migrations/native/1.12.9/postgres/schemaChanges.sql
@@ -0,0 +1,14 @@
+-- Restore the unique constraint dropped in 1.9.9. Closes the 1.9.9 regression that caused
+-- /columns?fields=profile 504s, and brings Postgres back in line with MySQL (which never
+-- lost it). The leading (entityFQNHash, extension) prefix serves the column-profile batch query.
+-- Two-phase: CONCURRENTLY build avoids ACCESS EXCLUSIVE lock; ADD CONSTRAINT USING INDEX
+-- promotes the built index without re-scanning.
+CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS
+    profiler_data_time_series_unique_hash_extension_ts
+    ON profiler_data_time_series (entityFQNHash, extension, operation, timestamp);
+
+ALTER TABLE profiler_data_time_series
+    ADD CONSTRAINT profiler_data_time_series_unique_hash_extension_ts
+    UNIQUE USING INDEX profiler_data_time_series_unique_hash_extension_ts;
+
+ANALYZE profiler_data_time_series;
diff --git a/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/TableResourceIT.java b/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/TableResourceIT.java
@@ -5861,4 +5861,141 @@ void test_listTablesWithColumnTags_performance(TestNamespace ns) {
       assertFalse(table.getTags().isEmpty(), "Table tags should not be empty");
     }
   }
+
+  // ===================================================================
+  // REGRESSION TEST - columns API with fields=profile (collate#3488)
+  // ===================================================================
+
+  @Test
+  @Execution(ExecutionMode.SAME_THREAD)
+  void test_getColumnsWithProfileField_correctnessAndNoBatchRegression(TestNamespace ns) {
+    OpenMetadataClient client = SdkClients.adminClient();
+
+    DatabaseService service = DatabaseServiceTestFactory.createPostgres(ns);
+    DatabaseSchema schema = DatabaseSchemaTestFactory.createSimple(ns, service);
+
+    CreateClassification createClassification =
+        new CreateClassification()
+            .withName(ns.prefix("cls"))
+            .withDescription("Classification for profile regression test");
+    Classification cls = client.classifications().create(createClassification);
+
+    CreateTag createTag =
+        new CreateTag()
+            .withName(ns.prefix("tag"))
+            .withDescription("Tag for profile regression test")
+            .withClassification(cls.getName());
+    Tag tag = client.tags().create(createTag);
+
+    TagLabel tagLabel =
+        new TagLabel()
+            .withTagFQN(tag.getFullyQualifiedName())
+            .withSource(TagLabel.TagSource.CLASSIFICATION);
+
+    Column idCol = ColumnBuilder.of("id", "BIGINT").primaryKey().notNull().build();
+    idCol.setTags(List.of(tagLabel));
+    Column emailCol = ColumnBuilder.of("email", "VARCHAR").dataLength(255).build();
+    emailCol.setTags(List.of(tagLabel));
+    Column nameCol = ColumnBuilder.of("name", "VARCHAR").dataLength(255).build();
+
+    CreateTable createRequest = createRequest(ns.prefix("profile_regression_table"), ns);
+    createRequest.setDatabaseSchema(schema.getFullyQualifiedName());
+    createRequest.setColumns(List.of(idCol, emailCol, nameCol));
+    Table table = client.tables().create(createRequest);
+
+    Long timestamp = System.currentTimeMillis();
+    ColumnProfile idProfile =
+        new ColumnProfile()
+            .withName("id")
+            .withMin(1.0)
+            .withMax(999.0)
+            .withUniqueCount(100.0)
+            .withTimestamp(timestamp);
+    ColumnProfile emailProfile =
+        new ColumnProfile()
+            .withName("email")
+            .withNullCount(5.0)
+            .withNullProportion(0.05)
+            .withTimestamp(timestamp);
+
+    TableProfile tableProfile =
+        new TableProfile().withRowCount(100.0).withColumnCount(3.0).withTimestamp(timestamp);
+
+    CreateTableProfile createProfile =
+        new CreateTableProfile()
+            .withTableProfile(tableProfile)
+            .withColumnProfile(List.of(idProfile, emailProfile));
+    client.tables().updateTableProfile(table.getId(), createProfile);
+
+    // Verify the three field combinations exercised below don't regress:
+    // (a) fields=profile — completes within 30s and returns the expected column profiles
+    TableColumnList withProfile =
+        assertTimeout(
+            Duration.ofSeconds(30),
+            () -> client.tables().getColumns(table.getId(), "profile"),
+            "columns?fields=profile should complete within 30s");
+
+    assertEquals(3, withProfile.getData().size());
+    Column returnedId =
+        withProfile.getData().stream()
+            .filter(c -> "id".equals(c.getName()))
+            .findFirst()
+            .orElse(null);
+    Column returnedName =
+        withProfile.getData().stream()
+            .filter(c -> "name".equals(c.getName()))
+            .findFirst()
+            .orElse(null);
+    assertNotNull(returnedId, "id column should be present");
+    assertNotNull(returnedId.getProfile(), "id column should have profile data");
+    assertEquals(1.0, returnedId.getProfile().getMin(), "id column min should match");
+    assertEquals(999.0, returnedId.getProfile().getMax(), "id column max should match");
+    assertNotNull(returnedName, "name column should be present");
+    assertNull(returnedName.getProfile(), "name column has no profile, should be null");
+
+    // (b) fields=tags,customMetrics,extension,profile — the exact production query
+    TableColumnList withAllFields =
+        assertTimeout(
+            Duration.ofSeconds(30),
+            () -> client.tables().getColumns(table.getId(), "tags,customMetrics,extension,profile"),
+            "columns?fields=tags,customMetrics,extension,profile should complete within 30s");
+
+    assertEquals(3, withAllFields.getData().size());
+
+    Column idResult =
+        withAllFields.getData().stream()
+            .filter(c -> "id".equals(c.getName()))
+            .findFirst()
+            .orElse(null);
+    assertNotNull(idResult, "id column must be present");
+    assertNotNull(idResult.getProfile(), "id column must have profile");
+    assertNotNull(idResult.getTags(), "id column must have tags");
+    assertFalse(idResult.getTags().isEmpty(), "id column tags must not be empty");
+    assertTrue(
+        idResult.getTags().stream()
+            .anyMatch(t -> tag.getFullyQualifiedName().equals(t.getTagFQN())),
+        "id column should carry the test tag");
+
+    // (c) fields=tags,profile — both tags and profile are populated correctly when requested
+    //     together (the dedup of populateEntityFieldTags is exercised here, but this test
+    //     verifies the observable contract — tags + profile both present on the result —
+    //     not the internal call count)
+    TableColumnList withTagsAndProfile =
+        assertTimeout(
+            Duration.ofSeconds(30),
+            () -> client.tables().getColumns(table.getId(), "tags,profile"),
+            "columns?fields=tags,profile should complete within 30s");
+
+    assertEquals(3, withTagsAndProfile.getData().size());
+    Column idTagsProfile =
+        withTagsAndProfile.getData().stream()
+            .filter(c -> "id".equals(c.getName()))
+            .findFirst()
+            .orElse(null);
+    assertNotNull(idTagsProfile);
+    assertNotNull(idTagsProfile.getTags());
+    assertFalse(
+        idTagsProfile.getTags().isEmpty(), "Tags must be present even when profile requested");
+    assertNotNull(idTagsProfile.getProfile(), "Profile must be present when profile requested");
+  }
 }
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/CollectionDAO.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/CollectionDAO.java
@@ -1545,6 +1545,13 @@ void bulkUpsertExtensions(
     List<ExtensionRecord> getExtensions(
         @BindUUID("id") UUID id, @Bind("extensionPrefix") String extensionPrefix);
 
+    @RegisterRowMapper(ExtensionMapper.class)
+    @SqlQuery(
+        "SELECT extension, json FROM entity_extension WHERE id = :id AND jsonschema = :jsonSchema "
+            + "ORDER BY extension")
+    List<ExtensionRecord> getExtensionsByJsonSchema(
+        @BindUUID("id") UUID id, @Bind("jsonSchema") String jsonSchema);
+
     @ConnectionAwareSqlQuery(
         value =
             "SELECT json FROM ("
@@ -9550,7 +9557,8 @@ default String getTimeSeriesTableName() {
             + "  GROUP BY entityFQNHash"
             + ") latest "
             + "ON p.entityFQNHash = latest.entityFQNHash AND p.timestamp = latest.latestTs "
-            + "WHERE p.extension = :extension")
+            + "WHERE p.extension = :extension "
+            + "AND p.entityFQNHash IN (<entityFQNHashes>)")
     @RegisterRowMapper(LatestExtensionRecordMapper.class)
     List<LatestExtensionRecord> getLatestExtensionsBatch(
         @Define("table") String table,
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TableRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TableRepository.java
@@ -154,6 +154,7 @@ public class TableRepository extends EntityRepository<Table> {
   public static final String TABLE_COLUMN_EXTENSION = "table.column";
   public static final String TABLE_EXTENSION = "table.table";
   public static final String CUSTOM_METRICS_EXTENSION = "customMetrics.";
+  public static final String COLUMN_EXTENSION_JSON_SCHEMA = "columnExtension";
   public static final String TABLE_PROFILER_CONFIG = "tableProfilerConfig";
   private static final ReadPrefetchKey PREFETCH_DEFAULT_FIELDS =
       ReadPrefetchKey.TABLE_DEFAULT_FIELDS;
@@ -2194,6 +2195,21 @@ private Object getColumnExtension(UUID tableId, String columnFQN) {
     return null;
   }
 
+  private Map<String, List<CustomMetric>> batchFetchCustomMetricsByColumn(UUID tableId) {
+    List<ExtensionRecord> records =
+        daoCollection
+            .entityExtensionDAO()
+            .getExtensions(tableId, CUSTOM_METRICS_EXTENSION + TABLE_COLUMN_EXTENSION);
+    Map<String, List<CustomMetric>> metricsByColumn = new HashMap<>();
+    for (ExtensionRecord record : records) {
+      CustomMetric metric = JsonUtils.readValue(record.extensionJson(), CustomMetric.class);
+      if (metric != null && metric.getColumnName() != null) {
+        metricsByColumn.computeIfAbsent(metric.getColumnName(), k -> new ArrayList<>()).add(metric);
+      }
+    }
+    return metricsByColumn;
+  }
+
   private List<CustomMetric> getCustomMetrics(Table table, String columnName) {
     String extension = columnName != null ? TABLE_COLUMN_EXTENSION : TABLE_EXTENSION;
     extension = CUSTOM_METRICS_EXTENSION + extension;
@@ -2891,20 +2907,43 @@ private ResultList<Column> getTableColumnsInternal(
     }
 
     if (fieldsParam != null && fieldsParam.contains("customMetrics")) {
+      Map<String, List<CustomMetric>> metricsByColumn =
+          batchFetchCustomMetricsByColumn(table.getId());
       for (Column column : paginatedColumns) {
-        column.setCustomMetrics(getCustomMetrics(table, column.getName()));
+        column.setCustomMetrics(metricsByColumn.getOrDefault(column.getName(), List.of()));
       }
     }
 
     if (fieldsParam != null && fieldsParam.contains("extension")) {
+      List<ExtensionRecord> allColumnExtensions =
+          daoCollection
+              .entityExtensionDAO()
+              .getExtensionsByJsonSchema(table.getId(), COLUMN_EXTENSION_JSON_SCHEMA);
+      Map<String, Object> extensionByColumnHash = new HashMap<>();
+      for (ExtensionRecord record : allColumnExtensions) {
+        try {
+          extensionByColumnHash.put(
+              record.extensionName(), JsonUtils.readValue(record.extensionJson(), Object.class));
+        } catch (Exception e) {
+          LOG.warn(
+              "Failed to deserialize column extension for table {} extensionKey {}: {}",
+              table.getId(),
+              record.extensionName(),
+              e.getMessage());
+        }
+      }
       for (Column column : paginatedColumns) {
-        column.setExtension(getColumnExtension(table.getId(), column.getFullyQualifiedName()));
+        column.setExtension(
+            extensionByColumnHash.get(
+                FullyQualifiedName.buildHash(column.getFullyQualifiedName())));
       }
     }
 
     if (fieldsParam != null && fieldsParam.contains("profile")) {
       setColumnProfile(paginatedColumns);
-      populateEntityFieldTags(entityType, paginatedColumns, table.getFullyQualifiedName(), true);
+      if (!fieldsParam.contains("tags")) {
+        populateEntityFieldTags(entityType, paginatedColumns, table.getFullyQualifiedName(), true);
+      }
       paginatedColumns =
           piiOwners != null
               ? PIIMasker.getTableProfile(piiOwners, paginatedColumns, authorizer, securityContext)
@@ -3227,8 +3266,10 @@ private ResultList<Column> searchTableColumnsInternal(
 
     Fields fields = getFields(fieldsParam);
     if (fields.contains("customMetrics") || fields.contains("*")) {
+      Map<String, List<CustomMetric>> metricsByColumn =
+          batchFetchCustomMetricsByColumn(table.getId());
       for (Column column : paginatedResults) {
-        column.setCustomMetrics(getCustomMetrics(table, column.getName()));
+        column.setCustomMetrics(metricsByColumn.getOrDefault(column.getName(), List.of()));
       }
     }
 
