fix(migration): heal stuck PG certification stranded by v1125 (v11210) (#28635)

manerow · pmbrull · commit 26d5ae1609f3 · 2026-06-02T18:06:53.000+02:00
* fix(migration): heal stuck PG certification stranded by v1125 in v11210

* test(migration): assert v11210 heal is no-op on 1.13 reprocess (idempotent insert)
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/migration/mysql/v11210/Migration.java b/openmetadata-service/src/main/java/org/openmetadata/service/migration/mysql/v11210/Migration.java
@@ -14,6 +14,7 @@
 
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import org.openmetadata.service.jdbi3.locator.ConnectionType;
 import org.openmetadata.service.migration.api.MigrationProcessImpl;
 import org.openmetadata.service.migration.utils.MigrationFile;
 import org.openmetadata.service.migration.utils.v11210.MigrationUtil;
@@ -30,5 +31,10 @@ public Migration(MigrationFile migrationFile) {
   public void runDataMigration() {
     MigrationUtil.removeFlattenedChildrenSearchSettings();
     MigrationUtil.removeStaleFileExtensionAggregation();
+    try {
+      MigrationUtil.healStuckCertificationOnEntityJson(handle, ConnectionType.MYSQL);
+    } catch (Exception e) {
+      LOG.error("v11210 heal of stuck certification on entity json failed", e);
+    }
   }
 }
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/migration/postgres/v11210/Migration.java b/openmetadata-service/src/main/java/org/openmetadata/service/migration/postgres/v11210/Migration.java
@@ -14,6 +14,7 @@
 
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import org.openmetadata.service.jdbi3.locator.ConnectionType;
 import org.openmetadata.service.migration.api.MigrationProcessImpl;
 import org.openmetadata.service.migration.utils.MigrationFile;
 import org.openmetadata.service.migration.utils.v11210.MigrationUtil;
@@ -30,5 +31,10 @@ public Migration(MigrationFile migrationFile) {
   public void runDataMigration() {
     MigrationUtil.removeFlattenedChildrenSearchSettings();
     MigrationUtil.removeStaleFileExtensionAggregation();
+    try {
+      MigrationUtil.healStuckCertificationOnEntityJson(handle, ConnectionType.POSTGRES);
+    } catch (Exception e) {
+      LOG.error("v11210 heal of stuck certification on entity json failed", e);
+    }
   }
 }
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/migration/utils/v11210/MigrationUtil.java b/openmetadata-service/src/main/java/org/openmetadata/service/migration/utils/v11210/MigrationUtil.java
@@ -15,16 +15,26 @@
 import static org.openmetadata.common.utils.CommonUtil.listOrEmpty;
 import static org.openmetadata.common.utils.CommonUtil.nullOrEmpty;
 
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.sql.Timestamp;
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import lombok.extern.slf4j.Slf4j;
+import org.jdbi.v3.core.Handle;
+import org.jdbi.v3.core.statement.PreparedBatch;
 import org.openmetadata.schema.api.search.Aggregation;
 import org.openmetadata.schema.api.search.AllowedSearchFields;
 import org.openmetadata.schema.api.search.AssetTypeConfiguration;
 import org.openmetadata.schema.api.search.FieldBoost;
 import org.openmetadata.schema.api.search.SearchSettings;
 import org.openmetadata.schema.settings.Settings;
+import org.openmetadata.schema.type.TagLabel;
+import org.openmetadata.schema.utils.JsonUtils;
+import org.openmetadata.service.jdbi3.locator.ConnectionType;
 import org.openmetadata.service.migration.utils.SearchSettingsMergeUtil;
+import org.openmetadata.service.util.FullyQualifiedName;
 
 @Slf4j
 public class MigrationUtil {
@@ -164,4 +174,209 @@ private static boolean removeStaleSearchField(List<FieldBoost> searchFields) {
     }
     return removed;
   }
+
+  // Entity tables that v1125 attempted to drain of inline certification into tag_usage.
+  private static final String[] CERTIFIED_ENTITY_TABLES = {
+    "table_entity",
+    "dashboard_entity",
+    "topic_entity",
+    "pipeline_entity",
+    "storage_container_entity",
+    "search_index_entity",
+    "ml_model_entity",
+    "stored_procedure_entity",
+    "dashboard_data_model_entity",
+    "api_endpoint_entity",
+    "api_collection_entity",
+    "database_entity",
+    "database_schema_entity",
+    "data_product_entity",
+    "domain_entity",
+    "chart_entity",
+    "metric_entity",
+    "file_entity",
+    "directory_entity",
+    "spreadsheet_entity",
+    "worksheet_entity",
+    "llm_model_entity",
+    "ai_application_entity"
+  };
+
+  private static final int HEAL_BATCH_SIZE = 500;
+  private static final int CERT_SOURCE = TagLabel.TagSource.CLASSIFICATION.ordinal();
+  private static final int CERT_LABEL_TYPE = TagLabel.LabelType.AUTOMATED.ordinal();
+  private static final int CERT_STATE = TagLabel.State.CONFIRMED.ordinal();
+
+  /**
+   * Heals installs where certification is still inlined in entity JSON instead of being in
+   * tag_usage. v1125 on PostgreSQL silently failed on a missing varchar-&gt;json cast for the
+   * metadata column: the insert threw, the per-table catch swallowed it, and v1125 was still
+   * recorded as DONE — stranding certification in entity JSON with no tag_usage row. v1125 never
+   * re-runs, so this heal (self-contained with the correct cast, independent of the v1125 SQL)
+   * recovers those clusters. Idempotent: the select only matches rows that still carry
+   * certification in JSON, so fresh installs, healthy MySQL, and already-healed PG databases exit
+   * at zero rows.
+   */
+  public static void healStuckCertificationOnEntityJson(Handle handle, ConnectionType connType) {
+    int totalHealed = 0;
+    for (String table : CERTIFIED_ENTITY_TABLES) {
+      try {
+        int healed = healCertificationForTable(handle, connType, table);
+        totalHealed += healed;
+        if (healed > 0) {
+          LOG.info("v11210 heal: moved {} stuck certification rows from {}", healed, table);
+        }
+      } catch (Exception e) {
+        // Per-table failure shouldn't stop the rest; logged at ERROR so it isn't lost.
+        LOG.error("v11210 heal failed for table '{}': {}", table, e.getMessage(), e);
+      }
+    }
+    LOG.info("v11210 heal: total stuck certification rows healed: {}", totalHealed);
+  }
+
+  private record BatchResult(int rowsFetched, int healed) {}
+
+  private static int healCertificationForTable(
+      Handle handle, ConnectionType connType, String table) {
+    int totalHealed = 0;
+    boolean morePages = true;
+    while (morePages) {
+      BatchResult batch = healCertificationBatch(handle, connType, table);
+      totalHealed += batch.healed();
+      morePages = batch.rowsFetched() == HEAL_BATCH_SIZE;
+    }
+    return totalHealed;
+  }
+
+  private static BatchResult healCertificationBatch(
+      Handle handle, ConnectionType connType, String table) {
+    boolean isPostgres = connType == ConnectionType.POSTGRES;
+    int healed = 0;
+    List<Map<String, Object>> rows =
+        handle.createQuery(buildSelectStuckCertificationSql(table, isPostgres)).mapToMap().list();
+    if (!nullOrEmpty(rows)) {
+      List<String> selectedIds = new ArrayList<>();
+      PreparedBatch batch = handle.prepareBatch(buildInsertTagUsageSql(isPostgres));
+      for (Map<String, Object> row : rows) {
+        String tagFQN = (String) row.get("tagfqn");
+        // SELECT already filters tagFQN NOT NULL, but stay defensive against concurrent edits.
+        if (tagFQN != null) {
+          selectedIds.add(row.get("id").toString());
+          batch
+              .bind("source", CERT_SOURCE)
+              .bind("tagFQN", tagFQN)
+              .bind("tagFQNHash", FullyQualifiedName.buildHash(tagFQN))
+              .bind("targetFQNHash", row.get("fqnhash").toString())
+              .bind("labelType", CERT_LABEL_TYPE)
+              .bind("state", CERT_STATE)
+              .bind("appliedAt", parseEpochMillisToTimestamp(row.get("applieddate")))
+              .bind("metadata", buildCertMetadataJson(row.get("expirydate")))
+              .add();
+        }
+      }
+      if (!nullOrEmpty(selectedIds)) {
+        batch.execute();
+        handle
+            .createUpdate(buildStripCertificationFromJsonSql(table, isPostgres))
+            .bindList("ids", selectedIds)
+            .execute();
+        healed = selectedIds.size();
+      }
+    }
+    return new BatchResult(rows.size(), healed);
+  }
+
+  private static String buildSelectStuckCertificationSql(String table, boolean isPostgres) {
+    String sql;
+    if (isPostgres) {
+      sql =
+          String.format(
+              "SELECT id, fqnHash, "
+                  + "json::json -> 'certification' -> 'tagLabel' ->> 'tagFQN' AS tagFQN, "
+                  + "json::json -> 'certification' ->> 'expiryDate' AS expiryDate, "
+                  + "json::json -> 'certification' ->> 'appliedDate' AS appliedDate "
+                  + "FROM %s WHERE json::jsonb ?? 'certification' "
+                  + "AND json::json -> 'certification' -> 'tagLabel' ->> 'tagFQN' IS NOT NULL "
+                  + "LIMIT %d",
+              table, HEAL_BATCH_SIZE);
+    } else {
+      sql =
+          String.format(
+              "SELECT id, fqnHash, "
+                  + "JSON_UNQUOTE(JSON_EXTRACT(json, '$.certification.tagLabel.tagFQN')) AS tagFQN, "
+                  + "JSON_UNQUOTE(JSON_EXTRACT(json, '$.certification.expiryDate')) AS expiryDate, "
+                  + "JSON_UNQUOTE(JSON_EXTRACT(json, '$.certification.appliedDate')) AS appliedDate "
+                  + "FROM %s WHERE JSON_CONTAINS_PATH(json, 'one', '$.certification') = 1 "
+                  + "AND JSON_UNQUOTE(JSON_EXTRACT(json, '$.certification.tagLabel.tagFQN')) IS NOT NULL "
+                  + "LIMIT %d",
+              table, HEAL_BATCH_SIZE);
+    }
+    return sql;
+  }
+
+  private static String buildInsertTagUsageSql(boolean isPostgres) {
+    // PG won't auto-cast varchar -> json on the metadata column; MySQL JSON accepts a string.
+    String sql;
+    if (isPostgres) {
+      sql =
+          "INSERT INTO tag_usage "
+              + "(source, tagFQN, tagFQNHash, targetFQNHash, labelType, state, appliedBy, appliedAt, metadata) "
+              + "VALUES (:source, :tagFQN, :tagFQNHash, :targetFQNHash, :labelType, :state, 'admin', :appliedAt, :metadata::json) "
+              + "ON CONFLICT (source, tagFQNHash, targetFQNHash) DO NOTHING";
+    } else {
+      sql =
+          "INSERT IGNORE INTO tag_usage "
+              + "(source, tagFQN, tagFQNHash, targetFQNHash, labelType, state, appliedBy, appliedAt, metadata) "
+              + "VALUES (:source, :tagFQN, :tagFQNHash, :targetFQNHash, :labelType, :state, 'admin', :appliedAt, :metadata)";
+    }
+    return sql;
+  }
+
+  private static String buildStripCertificationFromJsonSql(String table, boolean isPostgres) {
+    String sql;
+    if (isPostgres) {
+      sql =
+          "UPDATE "
+              + table
+              + " SET json = (json::jsonb - 'certification')::json WHERE id IN (<ids>)";
+    } else {
+      sql =
+          "UPDATE "
+              + table
+              + " SET json = JSON_REMOVE(json, '$.certification') WHERE id IN (<ids>)";
+    }
+    return sql;
+  }
+
+  private static Timestamp parseEpochMillisToTimestamp(Object val) {
+    Timestamp result = null;
+    if (val != null) {
+      try {
+        long epochMillis =
+            val instanceof Number num ? num.longValue() : Long.parseLong(val.toString());
+        result = new Timestamp(epochMillis);
+      } catch (NumberFormatException e) {
+        LOG.warn("Unparseable appliedDate '{}' on heal; binding null", val);
+      }
+    }
+    return result;
+  }
+
+  private static String buildCertMetadataJson(Object expiryDateVal) {
+    ObjectNode node = JsonUtils.getObjectNode();
+    if (expiryDateVal != null) {
+      if (expiryDateVal instanceof Long longVal) {
+        node.put("expiryDate", longVal);
+      } else if (expiryDateVal instanceof Number numberVal) {
+        node.put("expiryDate", numberVal.longValue());
+      } else {
+        try {
+          node.put("expiryDate", Long.parseLong(expiryDateVal.toString()));
+        } catch (NumberFormatException e) {
+          LOG.warn("Unparseable expiryDate '{}' on heal; omitting from metadata", expiryDateVal);
+        }
+      }
+    }
+    return node.toString();
+  }
 }
diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/migration/utils/v11210/MigrationUtilTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/migration/utils/v11210/MigrationUtilTest.java
@@ -0,0 +1,100 @@
+/*
+ *  Copyright 2024 Collate.
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package org.openmetadata.service.migration.utils.v11210;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Answers.RETURNS_DEEP_STUBS;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.Map;
+import org.jdbi.v3.core.Handle;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.openmetadata.service.jdbi3.locator.ConnectionType;
+
+class MigrationUtilTest {
+
+  private static Map<String, Object> stuckCertificationRow() {
+    return Map.of(
+        "id", "11111111-1111-1111-1111-111111111111",
+        "fqnhash", "abcdef",
+        "tagfqn", "Certification.Gold",
+        "applieddate", 1700000000000L,
+        "expirydate", 1800000000000L);
+  }
+
+  @Test
+  void postgresHealInsertCastsMetadataToJson() {
+    // The #2B regression: on PG the metadata (json) column rejects an uncast varchar bind. The
+    // heal's insert must cast :metadata to json or it fails the same way v1125 did.
+    Handle handle = mock(Handle.class, RETURNS_DEEP_STUBS);
+    when(handle.createQuery(anyString()).mapToMap().list())
+        .thenReturn(List.of(stuckCertificationRow()));
+    ArgumentCaptor<String> insertSql = ArgumentCaptor.forClass(String.class);
+
+    MigrationUtil.healStuckCertificationOnEntityJson(handle, ConnectionType.POSTGRES);
+
+    verify(handle, atLeastOnce()).prepareBatch(insertSql.capture());
+    assertTrue(
+        insertSql.getAllValues().stream().allMatch(sql -> sql.contains(":metadata::json")),
+        "PostgreSQL heal insert must cast :metadata to json");
+    // Backstop for the 1.13 reprocess: a duplicate insert must be a no-op against the unique key.
+    assertTrue(
+        insertSql.getAllValues().stream()
+            .allMatch(sql -> sql.contains("ON CONFLICT") && sql.contains("DO NOTHING")),
+        "PostgreSQL heal insert must be idempotent via ON CONFLICT DO NOTHING");
+  }
+
+  @Test
+  void mysqlHealInsertDoesNotCastMetadata() {
+    // MySQL JSON columns accept a string literal; the ::json cast is PG-only syntax.
+    Handle handle = mock(Handle.class, RETURNS_DEEP_STUBS);
+    when(handle.createQuery(anyString()).mapToMap().list())
+        .thenReturn(List.of(stuckCertificationRow()));
+    ArgumentCaptor<String> insertSql = ArgumentCaptor.forClass(String.class);
+
+    MigrationUtil.healStuckCertificationOnEntityJson(handle, ConnectionType.MYSQL);
+
+    verify(handle, atLeastOnce()).prepareBatch(insertSql.capture());
+    assertFalse(
+        insertSql.getAllValues().stream().anyMatch(sql -> sql.contains("::json")),
+        "MySQL heal insert must not use the PG-only ::json cast");
+    // Backstop for the 1.13 reprocess: a duplicate insert must be a no-op against the unique key.
+    assertTrue(
+        insertSql.getAllValues().stream().allMatch(sql -> sql.contains("INSERT IGNORE")),
+        "MySQL heal insert must be idempotent via INSERT IGNORE");
+  }
+
+  @Test
+  void healWithNoStuckRowsRunsNoInsertOrStrip() {
+    // This is the 1.13-upgrade case: v11210 already healed (and STRIPPED certification from the
+    // entity JSON) during the 1.12.10 run, so the SELECT now matches zero rows. The framework
+    // reprocesses v11210 and also runs v1130's heal on that upgrade, but both must issue NO insert
+    // and NO strip query — the same insert queries from 1.12.10 do not run again.
+    Handle handle = mock(Handle.class, RETURNS_DEEP_STUBS);
+    when(handle.createQuery(anyString()).mapToMap().list()).thenReturn(List.of());
+
+    assertDoesNotThrow(
+        () -> MigrationUtil.healStuckCertificationOnEntityJson(handle, ConnectionType.POSTGRES));
+    verify(handle, never()).prepareBatch(anyString());
+    verify(handle, never()).createUpdate(anyString());
+  }
+}
