Add source filter and use indexed hash prefix in cert tag batch query

The certification tag batch query (TagUsageDAO.getCertTagsInternalBatch)
was hitting ~12 seconds per call on instances with deep classification
hierarchies — fired ~5,800 times per Data Insights run, contributing
~19 hrs of cumulative DB time per DI run.

Two missing index-friendly predicates caused the slowness:
1. No `source = ?` filter — couldn't use idx_tag_usage_target_exact
   (source, targetFQNHash, state) INCLUDE (tagFQN, labelType) whose
   covering INCLUDE has tagFQN.
2. `tagFQN LIKE 'Certification.%'` on the raw column — there's no
   LIKE-friendly index on raw tagFQN, only on tagfqn_lower text_pattern_ops
   and tagFQNHash. The LIKE always ran as a post-filter on every row the
   IN clause returned.

Fix:
- Add `source = :source` filter (Certifications are always Classification
  source = 0).
- Switch `tagFQN LIKE :tagFQNPrefix` → `tagFQNHash LIKE :tagFQNHashPrefix`,
  with the hash prefix pre-computed via FullyQualifiedName.buildHash so the
  query hits the indexed hash column.

Same SQL on MySQL and Postgres — no @ConnectionAwareSqlQuery split needed.

Also a correctness improvement: the `source = 0` filter excludes glossary
terms (source = 1) that happen to have FQNs starting with "Certification.".
Previously such glossary terms could be incorrectly returned as
certifications; now they're excluded as expected.

Test:
- Added test_certBatch_bulkFetchReturnsCorrectCertsPerEntity in
  TagResourceIT — exercises the bulk fetch path with three schemas
  (cert-tagged / untagged / non-cert-tagged) and asserts each gets
  the right certification (or null) in the listed response. Locks in
  source-filter correctness and prevents future regressions where a
  non-cert tag could leak into the certification field.

Author: sonika-shah

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

@@ -1589,6 +1589,114 @@ void test_certificationTagNotLeakingIntoTagsField(TestNamespace ns) {
         "LIST (batch): regular tag must still be present in tags field");
   }
 
+  /**
+   * Verifies the batched cert fetch path — the underlying query uses
+   * {@code WHERE source = 0 AND targetFQNHash IN (...) AND tagFQNHash LIKE 'md5(Certification).%'}.
+   *
+   * <p>Covers three correctness properties of that query:
+   * <ul>
+   *   <li>Entities with a cert get the correct certification populated</li>
+   *   <li>Entities without any cert get a null certification (no false positives from the IN list)</li>
+   *   <li>Entities with a non-cert tag from a different classification do NOT get that tag treated
+   *       as a certification (regression test for the {@code source} filter + hash prefix).</li>
+   * </ul>
+   */
+  @Test
+  void test_certBatch_bulkFetchReturnsCorrectCertsPerEntity(TestNamespace ns) {
+    OpenMetadataClient client = SdkClients.adminClient();
+
+    org.openmetadata.schema.entity.classification.Classification certClassification =
+        client.classifications().getByName("Certification", null);
+    assertNotNull(certClassification, "Certification classification must exist");
+
+    String certTagName = ns.shortPrefix("cert_bulk_tag");
+    CreateTag createCertTag = new CreateTag();
+    createCertTag.setName(certTagName);
+    createCertTag.setClassification(certClassification.getFullyQualifiedName());
+    createCertTag.setDescription("Cert tag for bulk fetch test");
+    Tag certTag = SdkClients.adminClient().tags().create(createCertTag);
+
+    org.openmetadata.schema.entity.classification.Classification regularClassification =
+        createClassification(ns);
+    CreateTag createRegularTag = new CreateTag();
+    createRegularTag.setName(ns.shortPrefix("regular_bulk_tag"));
+    createRegularTag.setClassification(regularClassification.getFullyQualifiedName());
+    createRegularTag.setDescription("Non-cert tag for bulk fetch test");
+    Tag regularTag = SdkClients.adminClient().tags().create(createRegularTag);
+
+    org.openmetadata.schema.entity.services.DatabaseService dbService =
+        createDatabaseService(ns, "cert_bulk_svc");
+    org.openmetadata.schema.entity.data.Database db =
+        createDatabase(ns, dbService.getFullyQualifiedName());
+
+    DatabaseSchema schemaWithCert = createDatabaseSchema(ns, db.getFullyQualifiedName());
+    DatabaseSchema schemaWithoutCert = createDatabaseSchema(ns, db.getFullyQualifiedName());
+    DatabaseSchema schemaWithRegularTag = createDatabaseSchema(ns, db.getFullyQualifiedName());
+
+    org.openmetadata.schema.type.TagLabel certTagLabel =
+        new org.openmetadata.schema.type.TagLabel()
+            .withTagFQN(certTag.getFullyQualifiedName())
+            .withSource(org.openmetadata.schema.type.TagLabel.TagSource.CLASSIFICATION)
+            .withLabelType(org.openmetadata.schema.type.TagLabel.LabelType.MANUAL);
+    schemaWithCert.setCertification(new AssetCertification().withTagLabel(certTagLabel));
+    client.databaseSchemas().update(schemaWithCert.getId().toString(), schemaWithCert);
+
+    org.openmetadata.schema.type.TagLabel regularTagLabel =
+        new org.openmetadata.schema.type.TagLabel()
+            .withTagFQN(regularTag.getFullyQualifiedName())
+            .withSource(org.openmetadata.schema.type.TagLabel.TagSource.CLASSIFICATION)
+            .withLabelType(org.openmetadata.schema.type.TagLabel.LabelType.MANUAL);
+    schemaWithRegularTag.setTags(List.of(regularTagLabel));
+    client
+        .databaseSchemas()
+        .update(schemaWithRegularTag.getId().toString(), schemaWithRegularTag);
+
+    org.openmetadata.sdk.models.ListParams listParams =
+        new org.openmetadata.sdk.models.ListParams()
+            .setDatabase(db.getFullyQualifiedName())
+            .setFields("certification");
+    org.openmetadata.sdk.models.ListResponse<DatabaseSchema> listed =
+        client.databaseSchemas().list(listParams);
+    assertNotNull(listed.getData(), "List response must contain data");
+
+    DatabaseSchema listedWithCert =
+        listed.getData().stream()
+            .filter(s -> s.getId().equals(schemaWithCert.getId()))
+            .findFirst()
+            .orElse(null);
+    DatabaseSchema listedWithoutCert =
+        listed.getData().stream()
+            .filter(s -> s.getId().equals(schemaWithoutCert.getId()))
+            .findFirst()
+            .orElse(null);
+    DatabaseSchema listedWithRegularTag =
+        listed.getData().stream()
+            .filter(s -> s.getId().equals(schemaWithRegularTag.getId()))
+            .findFirst()
+            .orElse(null);
+
+    assertNotNull(listedWithCert, "Cert-tagged schema must appear in list");
+    assertNotNull(listedWithoutCert, "Untagged schema must appear in list");
+    assertNotNull(listedWithRegularTag, "Regular-tagged schema must appear in list");
+
+    assertNotNull(
+        listedWithCert.getCertification(),
+        "Bulk fetch must populate certification for cert-tagged schema");
+    assertEquals(
+        certTag.getFullyQualifiedName(),
+        listedWithCert.getCertification().getTagLabel().getTagFQN(),
+        "Bulk fetch must return the exact cert tag applied");
+
+    assertNull(
+        listedWithoutCert.getCertification(),
+        "Schema without any cert tag must have null certification (no false positives)");
+
+    assertNull(
+        listedWithRegularTag.getCertification(),
+        "Schema with a non-cert tag from a different classification must have null certification "
+            + "(source filter + hash prefix must exclude tags outside the Certification classification)");
+  }
+
   @Test
   void test_certificationTagRenamePropagatesToEntityAndSearch(TestNamespace ns) throws Exception {
     OpenMetadataClient client = SdkClients.adminClient();

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

@@ -6476,13 +6476,15 @@ List<TagLabelWithFQNHash> getTagsInternalBatch(
     @SqlQuery(
         "SELECT targetFQNHash, source, tagFQN, labelType, state, reason, appliedAt, appliedBy, metadata "
             + "FROM tag_usage "
-            + "WHERE targetFQNHash IN (<targetFQNHashes>) "
-            + "AND tagFQN LIKE :tagFQNPrefix "
+            + "WHERE source = :source "
+            + "AND targetFQNHash IN (<targetFQNHashes>) "
+            + "AND tagFQNHash LIKE :tagFQNHashPrefix "
             + "ORDER BY targetFQNHash, tagFQN")
     @UseRowMapper(TagLabelWithFQNHashMapper.class)
     List<TagLabelWithFQNHash> getCertTagsInternalBatch(
+        @Bind("source") int source,
         @BindListFQN("targetFQNHashes") List<String> targetFQNHashes,
-        @Bind("tagFQNPrefix") String tagFQNPrefix);
+        @Bind("tagFQNHashPrefix") String tagFQNHashPrefix);
 
     /**
      * Batch fetch derived tags for multiple glossary term FQNs. Returns a map from glossary term

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

@@ -5266,7 +5266,9 @@ protected AssetCertification getCertification(T entity) {
         daoCollection
             .tagUsageDAO()
             .getCertTagsInternalBatch(
-                List.of(entity.getFullyQualifiedName()), certClassification + ".%");
+                TagLabel.TagSource.CLASSIFICATION.ordinal(),
+                List.of(entity.getFullyQualifiedName()),
+                FullyQualifiedName.buildHash(certClassification) + ".%");
     if (nullOrEmpty(certTags)) return null;
     return buildCertificationFromCertTag(certTags.get(0).toTagLabel());
   }
@@ -9944,7 +9946,12 @@ private Map<UUID, AssetCertification> batchFetchCertification(List<T> entities)
     List<CollectionDAO.TagUsageDAO.TagLabelWithFQNHash> certTags;
     try {
       certTags =
-          daoCollection.tagUsageDAO().getCertTagsInternalBatch(fqnList, certClassification + ".%");
+          daoCollection
+              .tagUsageDAO()
+              .getCertTagsInternalBatch(
+                  TagLabel.TagSource.CLASSIFICATION.ordinal(),
+                  fqnList,
+                  FullyQualifiedName.buildHash(certClassification) + ".%");
     } catch (Exception e) {
       LOG.warn(
           "batchFetchCertification: batch query failed, falling back to individual fetch: {}",