Fixes 32792: let CreateTableRequest carry certification directly

ingestion/src/metadata/utils/source_hash.py

@@ -135,6 +135,11 @@
     4. Sorts owners by FQN/name/id
     5. Removes volatile EntityReference fields (href, deleted, inherited)
     6. Normalizes schemaDefinition whitespace
+    7. Drops certification.appliedDate/expiryDate, which the backend always
+       recomputes server-side from AssetCertificationSettings and never takes
+       from the request, so they carry no change-detection signal and would
+       otherwise destabilize the hash if a connector ever populates them with
+       a run-time-relative value
     """
     result = _remove_volatile_fields(data)
 
@@ -153,6 +158,11 @@
     if "schemaDefinition" in result and result["schemaDefinition"]:  # noqa: RUF019
         result["schemaDefinition"] = _normalize_whitespace(result["schemaDefinition"])
 
+    if "certification" in result and isinstance(result["certification"], dict):
+        result["certification"] = {
+            k: v for k, v in result["certification"].items() if k not in ("appliedDate", "expiryDate")
+        }
+
     return result
 
 

ingestion/tests/unit/utils/test_source_hash.py

@@ -21,6 +21,7 @@
     DataType,
     TableConstraint,
 )
+from metadata.generated.schema.type.assetCertification import AssetCertification
 from metadata.generated.schema.type.entityReference import EntityReference
 from metadata.generated.schema.type.tagLabel import (
     LabelType,
@@ -41,6 +42,19 @@
 )
 
 
+def _certification(tag_fqn: str = "Certification.Bronze") -> AssetCertification:
+    return AssetCertification(
+        tagLabel=TagLabel(
+            tagFQN=tag_fqn,
+            source=TagSource.Classification,
+            labelType=LabelType.Automated,
+            state=State.Confirmed,
+        ),
+        appliedDate=1700000000000,
+        expiryDate=1731536000000,
+    )
+
+
 class TestNormalizeWhitespace:
     def test_normalize_whitespace_none(self):
         assert _normalize_whitespace(None) is None
@@ -470,6 +484,90 @@ def test_hash_excludes_source_hash_field(self):
         )
         assert generate_source_hash(request1) == generate_source_hash(request2)
 
+    def test_hash_changes_with_certification_added(self):
+        request1 = CreateTableRequest(
+            name="test_table",
+            databaseSchema="service.db.schema",
+            columns=[Column(name="id", dataType=DataType.INT)],
+        )
+        request2 = CreateTableRequest(
+            name="test_table",
+            databaseSchema="service.db.schema",
+            columns=[Column(name="id", dataType=DataType.INT)],
+            certification=_certification("Certification.Bronze"),
+        )
+        assert generate_source_hash(request1) != generate_source_hash(request2)
+
+    def test_hash_changes_with_certification_value_change(self):
+        request1 = CreateTableRequest(
+            name="test_table",
+            databaseSchema="service.db.schema",
+            columns=[Column(name="id", dataType=DataType.INT)],
+            certification=_certification("Certification.Bronze"),
+        )
+        request2 = CreateTableRequest(
+            name="test_table",
+            databaseSchema="service.db.schema",
+            columns=[Column(name="id", dataType=DataType.INT)],
+            certification=_certification("Certification.Gold"),
+        )
+        assert generate_source_hash(request1) != generate_source_hash(request2)
+
+    def test_hash_stable_when_certification_omitted(self):
+        request1 = CreateTableRequest(
+            name="test_table",
+            databaseSchema="service.db.schema",
+            columns=[Column(name="id", dataType=DataType.INT)],
+        )
+        request2 = CreateTableRequest(
+            name="test_table",
+            databaseSchema="service.db.schema",
+            columns=[Column(name="id", dataType=DataType.INT)],
+            certification=None,
+        )
+        assert generate_source_hash(request1) == generate_source_hash(request2)
+
+    def test_hash_stable_with_equivalent_certification_payload(self):
+        request1 = CreateTableRequest(
+            name="test_table",
+            databaseSchema="service.db.schema",
+            columns=[Column(name="id", dataType=DataType.INT)],
+            certification=_certification("Certification.Silver"),
+        )
+        request2 = CreateTableRequest(
+            name="test_table",
+            databaseSchema="service.db.schema",
+            columns=[Column(name="id", dataType=DataType.INT)],
+            certification=_certification("Certification.Silver"),
+        )
+        assert generate_source_hash(request1) == generate_source_hash(request2)
+
+    def test_hash_stable_across_certification_applied_and_expiry_dates(self):
+        """appliedDate/expiryDate are always recomputed server-side from
+        AssetCertificationSettings when a certification is applied, so a
+        connector populating them with a run-time-relative value (e.g. now())
+        must not defeat the bulk fast-path when the certification itself is
+        unchanged. Only tagLabel is a real change-detection signal."""
+        tag_label = TagLabel(
+            tagFQN="Certification.Gold",
+            source=TagSource.Classification,
+            labelType=LabelType.Automated,
+            state=State.Confirmed,
+        )
+        request1 = CreateTableRequest(
+            name="test_table",
+            databaseSchema="service.db.schema",
+            columns=[Column(name="id", dataType=DataType.INT)],
+            certification=AssetCertification(tagLabel=tag_label, appliedDate=1700000000000, expiryDate=1731536000000),
+        )
+        request2 = CreateTableRequest(
+            name="test_table",
+            databaseSchema="service.db.schema",
+            columns=[Column(name="id", dataType=DataType.INT)],
+            certification=AssetCertification(tagLabel=tag_label, appliedDate=1800000000000, expiryDate=1999999999999),
+        )
+        assert generate_source_hash(request1) == generate_source_hash(request2)
+
     def test_hash_with_custom_exclude_fields(self):
         request1 = CreateTableRequest(
             name="test_table",

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

@@ -535,7 +535,7 @@
   @Getter protected final Set<String> allowedFields;
   public final boolean supportsSoftDelete;
   @Getter protected final boolean supportsTags;
   @Getter protected final boolean supportsOwners;
   @Getter protected final boolean supportsStyle;
   @Getter protected final boolean supportsLifeCycle;
   @Getter protected final boolean supportsCertification;
@@ -10029,10 +10029,16 @@
           origCertification,
           updatedCertification);
 
-      if (operation.isPut() && !nullOrEmpty(original.getCertification()) && updatedByBot()) {
-        // Revert change to non-empty certification if it is being updated by a bot
-        // This is to prevent bots from overwriting the certification. Certification need to be
-        // updated with a PATCH request
+      if (operation.isPut()
+          && !nullOrEmpty(original.getCertification())
+          && updatedByBot()
+          && !overrideMetadata
+          && updatedCertification == null) {
+        // A bot's PUT/create request that omits certification (most connectors never populate
+        // it) must not blank out a certification set through the UI or a prior explicit request.
+        // Certification can still be updated with a PATCH request, an explicit certification
+        // value in the request (e.g. CreateTableRequest.certification), or via the bulk path with
+        // overrideMetadata=true.
         updated.setCertification(original.getCertification());
         return;
       }
@@ -10044,8 +10050,25 @@
         return;
       }
 
-      if (Objects.equals(origCertification, updatedCertification)) {
+      // Compare by tagLabel.tagFQN only, not full-object equality: appliedDate/expiryDate are
+      // always recomputed server-side below and stored back, so a request that legitimately
+      // doesn't know the server's current dates (e.g. an ingestion connector re-sending the same
+      // certification every run) would otherwise never compare equal, causing a spurious
+      // version bump and re-apply on every non-bulk PUT. Other TagLabel fields (labelType,
+      // state, etc.) are ignored - only the certification tag's identity matters here.
+      boolean certificationTagUnchanged =
+          origCertification != null
+              && origCertification.getTagLabel() != null
+              && updatedCertification.getTagLabel() != null
+              && Objects.equals(
+                  origCertification.getTagLabel().getTagFQN(),
+                  updatedCertification.getTagLabel().getTagFQN());
+      if (certificationTagUnchanged) {
         LOG.debug("Certification unchanged");
+        // Restore the stored (server-authoritative) certification, including its real
+        // appliedDate/expiryDate, so the request's arbitrary date fields are never persisted -
+        // this method only skips the re-apply/recordChange, not the eventual entity write.
+        updated.setCertification(origCertification);
         return;
       }
 

openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/TableMapper.java

@@ -28,7 +28,8 @@ public Table createToEntity(CreateTable create, String user) {
                     getEntityReference(Entity.DATABASE_SCHEMA, create.getDatabaseSchema())))
         .withDatabaseSchema(getEntityReference(Entity.DATABASE_SCHEMA, create.getDatabaseSchema()))
         .withRetentionPeriod(create.getRetentionPeriod())
-        .withSourceHash(create.getSourceHash());
+        .withSourceHash(create.getSourceHash())
+        .withCertification(create.getCertification());
   }
 
   public CustomMetric createCustomMetricToEntity(CreateCustomMetric create, String user) {

openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/EntityRepositoryCertificationTest.java

@@ -16,6 +16,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import java.lang.reflect.Method;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.List;
@@ -27,7 +28,9 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
+import org.openmetadata.schema.configuration.AssetCertificationSettings;
 import org.openmetadata.schema.entity.data.Pipeline;
+import org.openmetadata.schema.entity.teams.User;
 import org.openmetadata.schema.type.AssetCertification;
 import org.openmetadata.schema.type.TagLabel;
 import org.openmetadata.schema.type.TagLabelMetadata;
@@ -96,6 +99,177 @@ void tearDown() {
     Entity.setJobDAO(null);
     Entity.setSearchRepository(null);
     Entity.setEntityRelationshipRepository(null);
+    Entity.setSystemRepository(null);
+    Entity.cleanup();
+  }
+
+  private static EntityRepository<Pipeline>.EntityUpdater newUpdater(
+      TestPipelineRepo repo, Pipeline original, Pipeline updated, EntityRepository.Operation op) {
+    return repo.new EntityUpdater(original, updated, op);
+  }
+
+  private static void invokeUpdateCertification(EntityRepository<Pipeline>.EntityUpdater updater)
+      throws Exception {
+    Method method = EntityRepository.EntityUpdater.class.getDeclaredMethod("updateCertification");
+    method.setAccessible(true);
+    method.invoke(updater);
+  }
+
+  @SuppressWarnings("unchecked")
+  private static void registerBotUser(String botName) {
+    EntityRepository<User> mockUserRepo = mock(EntityRepository.class);
+    User bot = new User().withName(botName).withIsBot(true);
+    when(mockUserRepo.findByNameOrNull(anyString(), any()))
+        .thenAnswer(inv -> botName.equals(inv.getArgument(0)) ? bot : null);
+    Entity.registerEntity(User.class, Entity.USER, mockUserRepo);
+  }
+
+  @SuppressWarnings("unchecked")
+  private static void registerNoUsersFound() {
+    EntityRepository<User> mockUserRepo = mock(EntityRepository.class);
+    when(mockUserRepo.findByNameOrNull(anyString(), any())).thenReturn(null);
+    Entity.registerEntity(User.class, Entity.USER, mockUserRepo);
+  }
+
+  private static void registerSystemRepository() {
+    SystemRepository systemRepository = mock(SystemRepository.class);
+    when(systemRepository.getAssetCertificationSettingOrDefault())
+        .thenReturn(
+            new AssetCertificationSettings()
+                .withAllowedClassification("Certification")
+                .withValidityPeriod("P30D"));
+    Entity.setSystemRepository(systemRepository);
+  }
+
+  private static Pipeline pipelineWithCertification(String botName, AssetCertification cert) {
+    return new Pipeline()
+        .withId(UUID.randomUUID())
+        .withName("my-pipeline")
+        .withFullyQualifiedName("service.my-pipeline")
+        .withUpdatedBy(botName)
+        .withCertification(cert);
+  }
+
+  @Test
+  void updateCertificationBotPutOmittingCertificationPreservesExisting() throws Exception {
+    registerBotUser("ingestion-bot");
+    TagLabel origLabel = new TagLabel().withTagFQN("Certification.Gold");
+    AssetCertification origCert = new AssetCertification().withTagLabel(origLabel);
+
+    Pipeline original = pipelineWithCertification("ingestion-bot", origCert);
+    Pipeline updated = pipelineWithCertification("ingestion-bot", null);
+
+    EntityRepository<Pipeline>.EntityUpdater updater =
+        newUpdater(repo, original, updated, EntityRepository.Operation.PUT);
+
+    invokeUpdateCertification(updater);
+
+    assertNotNull(updated.getCertification());
+    assertEquals("Certification.Gold", updated.getCertification().getTagLabel().getTagFQN());
+  }
+
+  @Test
+  void updateCertificationBotPutWithExplicitDifferentCertificationIsApplied() throws Exception {
+    registerBotUser("ingestion-bot");
+    registerSystemRepository();
+
+    TagLabel origLabel = new TagLabel().withTagFQN("Certification.Bronze");
+    AssetCertification origCert = new AssetCertification().withTagLabel(origLabel);
+    TagLabel newLabel = new TagLabel().withTagFQN("Certification.Gold");
+    AssetCertification newCert = new AssetCertification().withTagLabel(newLabel);
+
+    Pipeline original = pipelineWithCertification("ingestion-bot", origCert);
+    Pipeline updated = pipelineWithCertification("ingestion-bot", newCert);
+
+    when(tagUsageDAO.getCertTagsInternalBatch(anyInt(), anyList(), anyString()))
+        .thenReturn(List.of());
+
+    EntityRepository<Pipeline>.EntityUpdater updater =
+        newUpdater(repo, original, updated, EntityRepository.Operation.PUT);
+
+    invokeUpdateCertification(updater);
+
+    assertNotNull(updated.getCertification());
+    assertEquals("Certification.Gold", updated.getCertification().getTagLabel().getTagFQN());
+  }
+
+  @Test
+  void updateCertificationBotPutOmittingCertificationWithOverrideMetadataClearsIt()
+      throws Exception {
+    registerBotUser("ingestion-bot");
+    TagLabel origLabel = new TagLabel().withTagFQN("Certification.Gold");
+    AssetCertification origCert = new AssetCertification().withTagLabel(origLabel);
+
+    Pipeline original = pipelineWithCertification("ingestion-bot", origCert);
+    Pipeline updated = pipelineWithCertification("ingestion-bot", null);
+
+    EntityRepository<Pipeline>.EntityUpdater updater =
+        newUpdater(repo, original, updated, EntityRepository.Operation.PUT);
+    updater.setOverrideMetadata(true);
+
+    invokeUpdateCertification(updater);
+
+    assertNull(updated.getCertification());
+  }
+
+  @Test
+  void updateCertificationHumanPutOmittingCertificationClearsIt() throws Exception {
+    registerNoUsersFound();
+    TagLabel origLabel = new TagLabel().withTagFQN("Certification.Gold");
+    AssetCertification origCert = new AssetCertification().withTagLabel(origLabel);
+
+    Pipeline original = pipelineWithCertification("a-human", origCert);
+    Pipeline updated = pipelineWithCertification("a-human", null);
+
+    EntityRepository<Pipeline>.EntityUpdater updater =
+        newUpdater(repo, original, updated, EntityRepository.Operation.PUT);
+
+    invokeUpdateCertification(updater);
+
+    assertNull(updated.getCertification());
+  }
+
+  @Test
+  void updateCertificationSameTagLabelWithDifferentDatesIsNotReapplied() throws Exception {
+    // The server always recomputes appliedDate/expiryDate when a certification is applied, so a
+    // request that legitimately can't know the server's current dates (e.g. an ingestion
+    // connector re-sending the same certification every run) must not be treated as a change.
+    // Comparing the full AssetCertification object (including dates) would otherwise never
+    // compare equal, causing a spurious re-apply and version bump on every non-bulk PUT.
+    registerBotUser("ingestion-bot");
+    TagLabel tagLabel = new TagLabel().withTagFQN("Certification.Gold");
+    AssetCertification origCert =
+        new AssetCertification()
+            .withTagLabel(tagLabel)
+            .withAppliedDate(1700000000000L)
+            .withExpiryDate(1731536000000L);
+    AssetCertification updatedCert =
+        new AssetCertification()
+            .withTagLabel(tagLabel)
+            .withAppliedDate(4000000000000L)
+            .withExpiryDate(4100000000000L);
+
+    Pipeline original = pipelineWithCertification("ingestion-bot", origCert);
+    Pipeline updated = pipelineWithCertification("ingestion-bot", updatedCert);
+
+    EntityRepository<Pipeline>.EntityUpdater updater =
+        newUpdater(repo, original, updated, EntityRepository.Operation.PUT);
+
+    invokeUpdateCertification(updater);
+
+    assertEquals(1700000000000L, updated.getCertification().getAppliedDate());
+    assertEquals(1731536000000L, updated.getCertification().getExpiryDate());
+    verify(tagUsageDAO, never())
+        .applyTag(
+            anyInt(),
+            anyString(),
+            anyString(),
+            anyString(),
+            anyInt(),
+            anyInt(),
+            nullable(String.class),
+            nullable(String.class),
+            nullable(TagLabelMetadata.class));
   }
 
   @Test