(feat) Add write support for Soft Delete Aspect

dao-impl/ebean-dao/src/main/java/com/linkedin/metadata/dao/EbeanLocalAccess.java

@@ -103,6 +103,19 @@ public <ASPECT extends RecordTemplate> int add(@Nonnull URN urn, @Nullable ASPEC
     return addWithOptimisticLocking(urn, newValue, aspectClass, auditStamp, null, ingestionTrackingContext, isTestMode);
   }
 
+  @Override
+  @Transactional
+  public <ASPECT extends RecordTemplate> int addWithOldValue(
+      @Nonnull URN urn,
+      @Nullable ASPECT oldValue,
+      @Nullable ASPECT newValue,
+      @Nonnull Class<ASPECT> aspectClass,
+      @Nonnull AuditStamp auditStamp,
+      @Nullable IngestionTrackingContext ingestionTrackingContext,
+      boolean isTestMode) {
+    return addWithOptimisticLocking(urn, oldValue, newValue, aspectClass, auditStamp, null, ingestionTrackingContext, isTestMode);
+  }
+
   @Override
   public <ASPECT extends RecordTemplate> int addWithOptimisticLocking(
       @Nonnull URN urn,
@@ -112,6 +125,18 @@ public <ASPECT extends RecordTemplate> int addWithOptimisticLocking(
       @Nullable Timestamp oldTimestamp,
       @Nullable IngestionTrackingContext ingestionTrackingContext,
       boolean isTestMode) {
+    return addWithOptimisticLocking(urn, null, newValue, aspectClass, auditStamp, oldTimestamp, ingestionTrackingContext, isTestMode);
+  }
+
+  private <ASPECT extends RecordTemplate> int addWithOptimisticLocking(
+      @Nonnull URN urn,
+      @Nullable ASPECT oldValue,
+      @Nullable ASPECT newValue,
+      @Nonnull Class<ASPECT> aspectClass,
+      @Nonnull AuditStamp auditStamp,
+      @Nullable Timestamp oldTimestamp,
+      @Nullable IngestionTrackingContext ingestionTrackingContext,
+      boolean isTestMode) {
 
     final long timestamp = auditStamp.hasTime() ? auditStamp.getTime() : System.currentTimeMillis();
     final String actor = auditStamp.hasActor() ? auditStamp.getActor().toString() : DEFAULT_ACTOR;
@@ -136,9 +161,18 @@ public <ASPECT extends RecordTemplate> int addWithOptimisticLocking(
       sqlUpdate.setParameter("a_urn", toJsonString(urn));
     }
 
-    // newValue is null if aspect is to be soft-deleted.
+    // newValue is null if aspect is to be (soft-)deleted.
     if (newValue == null) {
-      return sqlUpdate.setParameter("metadata", DELETED_VALUE).execute();
+      if (oldValue == null) {
+        // Shouldn't happen: shouldn't run delete() on an already-null/deleted aspect, or if calling to soft-delete, should
+        // pass in the old value.
+        // Since Old Schema logic that calls this method does NOT have access to the old value (without significant refactoring)
+        // we make the decision to log a warning and proceed with the soft-deletion with an empty "deletedContent" value.
+        log.warn(String.format(
+            "OldValue should not be null when NewValue is null (soft-deletion). Urn: <%s> and aspect: <%s>", urn, aspectClass));
+      }
+      return sqlUpdate.setParameter("metadata", RecordUtils.toJsonString(
+          createSoftDeletedAspect(aspectClass, oldValue, new Timestamp(timestamp), actor))).execute();
     }
 
     AuditedAspect auditedAspect = new AuditedAspect()
@@ -152,8 +186,8 @@ public <ASPECT extends RecordTemplate> int addWithOptimisticLocking(
       auditedAspect.setEmitter(ingestionTrackingContext.getEmitter(), SetMode.IGNORE_NULL);
     }
 
-      final String metadata = toJsonString(auditedAspect);
-      return sqlUpdate.setParameter("metadata", metadata).execute();
+    final String metadata = toJsonString(auditedAspect);
+    return sqlUpdate.setParameter("metadata", metadata).execute();
   }
 
   /**

dao-impl/ebean-dao/src/main/java/com/linkedin/metadata/dao/EbeanLocalDAO.java

@@ -634,7 +634,7 @@ protected <ASPECT extends RecordTemplate> long saveLatest(@Nonnull URN urn, @Non
         }
       }
 
-      insert(urn, newValue, aspectClass, newAuditStamp, LATEST_VERSION, trackingContext, isTestMode);
+      insertWithOldValue(urn, oldValue, newValue, aspectClass, newAuditStamp, LATEST_VERSION, trackingContext, isTestMode);
     }
 
     // This method will handle relationship ingestions and soft-deletions
@@ -784,7 +784,7 @@ protected <ASPECT extends RecordTemplate> AspectEntry<ASPECT> getLatest(@Nonnull
     }
     final ExtraInfo extraInfo = toExtraInfo(latest);
 
-    if (isSoftDeletedAspect(latest, aspectClass)) {
+    if (isSoftDeletedAspect(latest)) {
       return new AspectEntry<>(null, extraInfo, true);
     }
 
@@ -905,11 +905,34 @@ protected <ASPECT extends RecordTemplate> void updateWithOptimisticLocking(@Nonn
   protected <ASPECT extends RecordTemplate> void insert(@Nonnull URN urn, @Nullable RecordTemplate value,
       @Nonnull Class<ASPECT> aspectClass, @Nonnull AuditStamp auditStamp, long version,
       @Nullable IngestionTrackingContext trackingContext, boolean isTestMode) {
-    final EbeanMetadataAspect aspect = buildMetadataAspectBean(urn, value, aspectClass, auditStamp, version);
+    insertWithOldValue(urn, null, value, aspectClass, auditStamp, version, trackingContext, isTestMode);
+  }
+
+  /**
+   * Insert that also takes the old value of the aspect into account. This argument is used in the case
+   * Aspect soft-deletion is supposed to occur (ie. adding null).
+   *
+   * <p>TODO: Figure out if this should be surfaced to {@link BaseLocalDAO}'s {@link BaseLocalDAO#insert} signature.
+   * Right now, since it only has 1 use case -- soft-deletion and doesn't intuitively fit into the idea of an "insert",
+   * we are keeping it here only.
+   *
+   * @param urn                 entity urn
+   * @param oldValue            old value of the aspect
+   * @param newValue            new value of the aspect
+   * @param aspectClass         aspect class
+   * @param auditStamp          audit stamp
+   * @param version             version of the record
+   * @param trackingContext     tracking context
+   * @param isTestMode          test mode
+   */
+  protected <ASPECT extends RecordTemplate> void insertWithOldValue(@Nonnull URN urn, @Nullable RecordTemplate oldValue,
+      @Nullable RecordTemplate newValue, @Nonnull Class<ASPECT> aspectClass, @Nonnull AuditStamp auditStamp, long version,
+      @Nullable IngestionTrackingContext trackingContext, boolean isTestMode) {
+    final EbeanMetadataAspect aspect = buildMetadataAspectBean(urn, newValue, aspectClass, auditStamp, version);
     if (_schemaConfig != SchemaConfig.OLD_SCHEMA_ONLY && version == LATEST_VERSION) {
       // insert() could be called when updating log table (moving current versions into new history version)
       // the metadata entity tables shouldn't been updated.
-      _localAccess.add(urn, (ASPECT) value, aspectClass, auditStamp, trackingContext, isTestMode);
+      _localAccess.addWithOldValue(urn, (ASPECT) oldValue, (ASPECT) newValue, aspectClass, auditStamp, trackingContext, isTestMode);
     }
 
     // DO append change log table (metadata_aspect) if:
@@ -1459,7 +1482,7 @@ URN getUrn(@Nonnull String urn) {
   @Nonnull
   static <ASPECT extends RecordTemplate> Optional<ASPECT> toRecordTemplate(@Nonnull Class<ASPECT> aspectClass,
       @Nonnull EbeanMetadataAspect aspect) {
-    if (isSoftDeletedAspect(aspect, aspectClass)) {
+    if (isSoftDeletedAspect(aspect)) {
       return Optional.empty();
     }
     return Optional.of(RecordUtils.toRecordTemplate(aspectClass, aspect.getMetadata()));
@@ -1468,7 +1491,7 @@ static <ASPECT extends RecordTemplate> Optional<ASPECT> toRecordTemplate(@Nonnul
   @Nonnull
   static <ASPECT extends RecordTemplate> Optional<AspectWithExtraInfo<ASPECT>> toRecordTemplateWithExtraInfo(
       @Nonnull Class<ASPECT> aspectClass, @Nonnull EbeanMetadataAspect aspect) {
-    if (aspect.getMetadata() == null || isSoftDeletedAspect(aspect, aspectClass)) {
+    if (aspect.getMetadata() == null || isSoftDeletedAspect(aspect)) {
       return Optional.empty();
     }
     final ExtraInfo extraInfo = toExtraInfo(aspect);

dao-impl/ebean-dao/src/main/java/com/linkedin/metadata/dao/IEbeanLocalAccess.java

@@ -22,7 +22,7 @@ public interface IEbeanLocalAccess<URN extends Urn> {
   void setUrnPathExtractor(@Nonnull UrnPathExtractor<URN> urnPathExtractor);
 
   /**
-   * Upsert aspect into entity table.
+   * Upsert aspect into entity table (without Optimistic Locking).
    *
    * @param <ASPECT>                 metadata aspect value
    * @param urn                      entity urn
@@ -36,6 +36,27 @@ public interface IEbeanLocalAccess<URN extends Urn> {
   <ASPECT extends RecordTemplate> int add(@Nonnull URN urn, @Nullable ASPECT newValue, @Nonnull Class<ASPECT> aspectClass,
       @Nonnull AuditStamp auditStamp, @Nullable IngestionTrackingContext ingestionTrackingContext, boolean isTestMode);
 
+  /**
+   * Same as {@link #add(Urn, RecordTemplate, Class, AuditStamp, IngestionTrackingContext, boolean)} but takes into
+   * account the old value of the aspect.
+   *
+   * <p>TODO: Consider whether to deprecate the no-OldValue version of this method to avoid confusion.
+   * The only use case of the OldValue at the moment is to support soft-deletion (on the new schema).
+   *
+   * @param <ASPECT>                 metadata aspect value
+   * @param urn                      entity urn
+   * @param oldValue                 old aspect value in {@link RecordTemplate}
+   * @param newValue                 aspect value in {@link RecordTemplate}
+   * @param aspectClass              class of the aspect
+   * @param auditStamp               audit timestamp
+   * @param ingestionTrackingContext the ingestionTrackingContext of the MCE responsible for this update
+   * @param isTestMode               whether the test mode is enabled or not
+   * @return number of rows inserted or updated
+   */
+  <ASPECT extends RecordTemplate> int addWithOldValue(@Nonnull URN urn, @Nullable ASPECT oldValue, @Nullable ASPECT newValue,
+      @Nonnull Class<ASPECT> aspectClass, @Nonnull AuditStamp auditStamp, @Nullable IngestionTrackingContext ingestionTrackingContext,
+      boolean isTestMode);
+
   /**
    * Update aspect on entity table with optimistic locking. (compare-and-update on oldTimestamp).
    *

dao-impl/ebean-dao/src/main/java/com/linkedin/metadata/dao/utils/EBeanDAOUtils.java

@@ -19,6 +19,7 @@
 import com.linkedin.metadata.query.LocalRelationshipValue;
 import com.linkedin.metadata.query.RelationshipField;
 import com.linkedin.metadata.query.UrnField;
+import com.linkedin.metadata.validator.ValidationUtils;
 import io.ebean.EbeanServer;
 import io.ebean.SqlRow;
 import java.lang.reflect.InvocationTargetException;
@@ -175,21 +176,6 @@ public static <T> boolean compareResults(@Nullable ListResult<T> resultOld, @Nul
     return false;
   }
 
-  /**
-   * Checks whether the aspect record has been soft deleted.
-   *
-   * @param aspect aspect value
-   * @param aspectClass the type of the aspect
-   * @return boolean representing whether the aspect record has been soft deleted
-   */
-  public static <ASPECT extends RecordTemplate> boolean isSoftDeletedAspect(@Nonnull EbeanMetadataAspect aspect,
-      @Nonnull Class<ASPECT> aspectClass) {
-    // Convert metadata string to record template object
-    final RecordTemplate metadataRecord = RecordUtils.toRecordTemplate(aspectClass, aspect.getMetadata());
-    return metadataRecord.equals(DELETED_METADATA);
-  }
-
-
   /**
    * Read {@link SqlRow} list into a {@link EbeanMetadataAspect} list.
    * @param sqlRows list of {@link SqlRow}
@@ -277,15 +263,37 @@ private static <ASPECT extends RecordTemplate> EbeanMetadataAspect readSqlRow(Sq
   }
 
   /**
-   * Checks whether the entity table record has been soft deleted.
+   * Checks whether a record is Soft Deleted.
+   *
+   * <p>NOTE: Since soft deleted aspects are modeled as a specific aspect type -- SoftDeletedAspect -- the ability
+   * to cast the aspect to a {@link SoftDeletedAspect} is sufficient to determine whether the aspect *is* Soft Deleted.
+   * In other words, there are no current use cases where we store a SoftDeletedAspect with the `gma_deleted` flag set to
+   * anything other than "true".
+   *
+   * <p>This validation approach is necessary because many usages of checking soft-deletion are followed by
+   * a deserialization call to {@link RecordUtils#toRecordTemplate(Class, String)}, which will fail if
+   * we try to deserialize a SoftDeletedAspect -- to another Aspect Type -- with the flag set to "false".
+   *
    * @param sqlRow {@link SqlRow} result from MySQL server
    * @param columnName column name of entity table
    * @return boolean representing whether the aspect record has been soft deleted
    */
   public static boolean isSoftDeletedAspect(@Nonnull SqlRow sqlRow, @Nonnull String columnName) {
+    return isSoftDeletedAspect(sqlRow.getString(columnName));
+  }
+
+  /**
+   * Same as {@link #isSoftDeletedAspect(SqlRow, String)}, but for {@link EbeanMetadataAspect}.
+   */
+  public static boolean isSoftDeletedAspect(@Nonnull EbeanMetadataAspect aspect) {
+    return isSoftDeletedAspect(aspect.getMetadata());
+  }
+
+  private static boolean isSoftDeletedAspect(@Nonnull String metadata) {
     try {
-      SoftDeletedAspect aspect = RecordUtils.toRecordTemplate(SoftDeletedAspect.class, sqlRow.getString(columnName));
-      return aspect.hasGma_deleted() && aspect.isGma_deleted();
+      SoftDeletedAspect softDeletedAspect = RecordUtils.toRecordTemplate(SoftDeletedAspect.class, metadata);
+      ValidationUtils.validateAgainstSchema(softDeletedAspect);
+      return true;
     } catch (Exception e) {
       return false;
     }
@@ -423,6 +431,31 @@ public static LocalRelationshipCriterion buildRelationshipFieldCriterion(LocalRe
     return relationshipMap;
   }
 
+  /**
+   * Create a soft deleted aspect with the given old value and timestamp.
+   *
+   * <p>TODO: Make oldValue @Nonnull once the Old Schema is completed deprecated OR if old schema write pathways are
+   * refactored to be able to process (retrieve and then write) the old value.
+   *
+   * @param aspectClass the class of the aspect
+   * @param oldValue the old value of the aspect
+   * @param deletedTimestamp the timestamp of the old value
+   * @param deletedBy the user who deleted the aspect
+   * @return a SoftDeletedAspect with the given old value and timestamp
+   */
+  @Nonnull
+  public static <ASPECT extends RecordTemplate> SoftDeletedAspect createSoftDeletedAspect(
+      @Nonnull Class<ASPECT> aspectClass, @Nullable ASPECT oldValue, @Nonnull Timestamp deletedTimestamp, @Nonnull String deletedBy) {
+    SoftDeletedAspect sda = new SoftDeletedAspect().setGma_deleted(true)
+        .setCanonicalName(aspectClass.getCanonicalName())
+        .setDeletedTimestamp(deletedTimestamp.toString())
+        .setDeletedBy(deletedBy);
+    if (oldValue != null) {
+      sda.setDeletedContent(RecordUtils.toJsonString(oldValue));
+    }
+    return sda;
+  }
+
   // Using the GmaAnnotationParser, extract the model type from the @gma.model annotation on any models.
   private static ModelType parseModelTypeFromGmaAnnotation(RecordTemplate model) {
     try {

dao-impl/ebean-dao/src/test/java/com/linkedin/metadata/dao/EbeanLocalDAOTest.java

@@ -2432,7 +2432,7 @@ public void testGetSoftDeletedAspect() {
 
     // latest version of metadata should be null
     EbeanMetadataAspect aspect = getMetadata(urn, aspectName, 0);
-    assertTrue(isSoftDeletedAspect(aspect, AspectFoo.class));
+    assertTrue(isSoftDeletedAspect(aspect));
     Optional<AspectFoo> fooOptional = dao.get(AspectFoo.class, urn);
     assertFalse(fooOptional.isPresent());
 
@@ -2618,7 +2618,7 @@ public void testUndeleteSoftDeletedAspect() {
     if (dao.isChangeLogEnabled()) {
       // version=3 should correspond to soft deleted metadata
       EbeanMetadataAspect aspect = getMetadata(urn, aspectName, 3);
-      assertTrue(isSoftDeletedAspect(aspect, AspectFoo.class));
+      assertTrue(isSoftDeletedAspect(aspect));
       fooOptional = dao.get(AspectFoo.class, urn, 3);
       assertFalse(fooOptional.isPresent());
 

dao-impl/ebean-dao/src/test/java/com/linkedin/metadata/dao/utils/EBeanDAOUtilsTest.java

@@ -6,6 +6,7 @@
 import com.linkedin.data.template.RecordTemplate;
 import com.linkedin.data.template.StringArray;
 import com.linkedin.metadata.aspect.AuditedAspect;
+import com.linkedin.metadata.aspect.SoftDeletedAspect;
 import com.linkedin.metadata.dao.BaseLocalDAO;
 import com.linkedin.metadata.dao.EbeanLocalAccess;
 import com.linkedin.metadata.dao.EbeanMetadataAspect;
@@ -52,12 +53,14 @@
 import static org.mockito.Mockito.*;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
-import static org.testng.AssertJUnit.assertEquals;
-import static org.testng.AssertJUnit.assertNotNull;
+import static org.testng.AssertJUnit.*;
 
 
 public class EBeanDAOUtilsTest {
 
+  final static String SOFT_DELETED_ASPECT_WITH_DELETED_CONTENT = "{\"gma_deleted\": true,"
+      + "\"deletedContent\": \"{removed: false}\", \"canonicalName\": \"com.linkedin.common.Status\","
+      + "\"deletedTimestamp\": \"0L\", \"deletedBy\": \"urn:li:tester\"}";
 
   @Nonnull
   private String readSQLfromFile(@Nonnull String resourcePath) {
@@ -573,6 +576,44 @@ public void testIsSoftDeletedAspect() {
 
     when(sqlRow.getString("a_aspectbaz")).thenReturn("{\"random_value\": \"baz\"}");
     assertFalse(EBeanDAOUtils.isSoftDeletedAspect(sqlRow, "a_aspectbaz"));
+
+    when(sqlRow.getString("a_aspectbaz")).thenReturn("{\"random_value\": \"baz\"}");
+    assertFalse(EBeanDAOUtils.isSoftDeletedAspect(sqlRow, "a_aspectbaz"));
+
+    // NOTE how this is should return "true"; see the explanation in the method javadoc
+    when(sqlRow.getString("a_aspectqux")).thenReturn("{\"gma_deleted\": false}");
+    assertTrue(EBeanDAOUtils.isSoftDeletedAspect(sqlRow, "a_aspectqux"));
+
+    when(sqlRow.getString("a_aspectbax")).thenReturn(SOFT_DELETED_ASPECT_WITH_DELETED_CONTENT);
+    assertTrue(EBeanDAOUtils.isSoftDeletedAspect(sqlRow, "a_aspectbax"));
+
+    when(sqlRow.getString("a_aspectbazz"))
+        .thenReturn("input that should fail being able to serialize as a RecordTemplate entirely");
+    assertFalse(EBeanDAOUtils.isSoftDeletedAspect(sqlRow, "a_aspectbazz"));
+  }
+
+  @Test
+  public void testIsSoftDeletedAspectEbeanMetadataAspect() {
+    EbeanMetadataAspect ebeanMetadataAspect = new EbeanMetadataAspect();
+
+    ebeanMetadataAspect.setMetadata("{\"gma_deleted\": true}");
+    assertTrue(EBeanDAOUtils.isSoftDeletedAspect(ebeanMetadataAspect));
+
+    ebeanMetadataAspect.setMetadata(SOFT_DELETED_ASPECT_WITH_DELETED_CONTENT);
+    assertTrue(EBeanDAOUtils.isSoftDeletedAspect(ebeanMetadataAspect));
+
+    // NOTE how this is should return "true"; see the explanation in the method javadoc
+    ebeanMetadataAspect.setMetadata("{\"gma_deleted\": false}");
+    assertTrue(EBeanDAOUtils.isSoftDeletedAspect(ebeanMetadataAspect));
+
+    ebeanMetadataAspect.setMetadata("{\"aspect\": {\"value\": \"bar\"}, \"lastmodifiedby\": \"urn:li:tester\"}");
+    assertFalse(EBeanDAOUtils.isSoftDeletedAspect(ebeanMetadataAspect));
+
+    ebeanMetadataAspect.setMetadata("{\"random_value\": \"baz\"}");
+    assertFalse(EBeanDAOUtils.isSoftDeletedAspect(ebeanMetadataAspect));
+
+    ebeanMetadataAspect.setMetadata("input that should fail being able to serialize as a RecordTemplate entirely");
+    assertFalse(EBeanDAOUtils.isSoftDeletedAspect(ebeanMetadataAspect));
   }
 
   @Test
@@ -668,4 +709,20 @@ public void testExtractRelationshipsFromAspect() throws URISyntaxException {
     assertTrue(results.get(AnnotatedRelationshipFoo.class).contains(test3));
     assertTrue(results.get(AnnotatedRelationshipBar.class).contains(new AnnotatedRelationshipBar()));
   }
+
+  @Test
+  public void testCreateSoftDeletedAspect() {
+    // ideal case: nonnull oldvalue and oldtimestamp
+    AspectFoo fooAspect = new AspectFoo().setValue("foo");
+    Timestamp timestamp = new Timestamp(0L);
+    String actor = "actor";
+
+    SoftDeletedAspect softDeletedAspect =
+        EBeanDAOUtils.createSoftDeletedAspect(AspectFoo.class, fooAspect, timestamp, actor);
+    assertTrue(softDeletedAspect.isGma_deleted());
+    assertEquals(softDeletedAspect.getCanonicalName(), "com.linkedin.testing.AspectFoo");
+    assertEquals(softDeletedAspect.getDeletedContent(), RecordUtils.toJsonString(fooAspect));
+    assertEquals(softDeletedAspect.getDeletedTimestamp(), timestamp.toString());
+    assertEquals(softDeletedAspect.getDeletedBy(), actor);
+  }
 }