feature(new schema): add flag to write only latest version to metadat…

…a_aspect (#306)

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

@@ -103,11 +103,19 @@ public enum SchemaConfig {
   // true if metadata change will be persisted into the change log table (metadata_aspect)
   private boolean _changeLogEnabled = true;
 
+  // TODO: remove this logic once metadata_aspect has been completed removed from TMS
+  // regarding metadata_aspect table:
+  // false = read/bump 2nd latest version + insert latest version
+  // true = overwrite 2nd latest version with latest version (equivalent to keeping only version = 0 rows in metadata_aspect)
+  private boolean _overwriteLatestVersionEnabled = false;
+
   public void setChangeLogEnabled(boolean changeLogEnabled) {
     if (_schemaConfig == SchemaConfig.NEW_SCHEMA_ONLY) {
       _changeLogEnabled = changeLogEnabled;
     } else {
       // For non-new schema, _changeLog will be enforced to be true
+      log.warn("You can only enable or disable the change log in new schema mode."
+          + "In old and dual schema modes, this setting is always enabled.");
       _changeLogEnabled = true;
     }
   }
@@ -116,6 +124,22 @@ public boolean isChangeLogEnabled() {
     return _changeLogEnabled;
   }
 
+  public void setOverwriteLatestVersionEnabled(boolean overwriteLatestVersionEnabled) {
+    if (_schemaConfig == SchemaConfig.NEW_SCHEMA_ONLY) {
+      if (isChangeLogEnabled()) {
+        _overwriteLatestVersionEnabled = overwriteLatestVersionEnabled;
+      } else {
+        log.warn("You can only enable or disable overwriting the latest version when the change log is enabled as well.");
+        _overwriteLatestVersionEnabled = false;
+      }
+    } else {
+      // For non-new schema, _ovewriteLatestVersionEnabled will be enforced to be false
+      log.warn("You can only enable or disable overwriting the latest version in new schema mode."
+          + "In old and dual schema modes, this setting is always disabled.");
+      _overwriteLatestVersionEnabled = false;
+    }
+  }
+
   public enum FindMethodology {
     UNIQUE_ID,      // (legacy) https://javadoc.io/static/io.ebean/ebean/11.19.2/io/ebean/EbeanServer.html#find-java.lang.Class-java.lang.Object-
     DIRECT_SQL,     // https://javadoc.io/static/io.ebean/ebean/11.19.2/io/ebean/EbeanServer.html#findNative-java.lang.Class-java.lang.String-
@@ -551,8 +575,10 @@ protected <ASPECT extends RecordTemplate> long saveLatest(@Nonnull URN urn, @Non
           log.debug("Insert: {} => oldValue = {}, latest version = {}", urn, oldValue, largestVersion);
         }
       }
-      // Move latest version to historical version by insert a new record.
-      insert(urn, oldValue, aspectClass, oldAuditStamp, largestVersion, trackingContext);
+      // Move latest version to historical version by insert a new record only if we are not overwriting the latest version.
+      if (!_overwriteLatestVersionEnabled) {
+        insert(urn, oldValue, aspectClass, oldAuditStamp, largestVersion, trackingContext);
+      }
       // update latest version
       updateWithOptimisticLocking(urn, newValue, aspectClass, newAuditStamp, LATEST_VERSION,
           new Timestamp(oldAuditStamp.getTime()), trackingContext);

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

@@ -3332,6 +3332,42 @@ public void testDataNotWrittenIntoOldSchemaWhenChangeLogIsDisabled() {
     assertFalse(legacyDao.get(AspectFoo.class, fooUrn).isPresent());
   }
 
+  @Test
+  public void testOverwriteLatestVersion() {
+    if (!_enableChangeLog || _schemaConfig != SchemaConfig.NEW_SCHEMA_ONLY) {
+      // skip this test if the change log is not even enabled and/or if we are not operating in new schema mode.
+      return;
+    }
+
+    // new schema DAO, used for inserts and reads
+    EbeanLocalDAO<EntityAspectUnion, FooUrn> newSchemaDao = createDao(FooUrn.class);
+    newSchemaDao.setOverwriteLatestVersionEnabled(true);
+
+    // old schema DAO, used for reads only
+    EbeanLocalDAO<EntityAspectUnion, FooUrn> oldSchemaDao = createDao(FooUrn.class);
+    oldSchemaDao.setSchemaConfig(SchemaConfig.OLD_SCHEMA_ONLY);
+
+    // Given: first version of metadata is inserted
+    FooUrn fooUrn = makeFooUrn(1);
+    AspectFoo v1 = new AspectFoo().setValue("foo");
+    newSchemaDao.add(fooUrn, v1, _dummyAuditStamp);
+
+    // When: second version of metadata is inserted
+    AspectFoo v2 = new AspectFoo().setValue("bar");
+    newSchemaDao.add(fooUrn, v2, _dummyAuditStamp);
+
+    // Expect: second version of metadata inserted overwrote the first version in the metadata_aspect table
+    Optional<AspectFoo> newSchemaResult = newSchemaDao.get(AspectFoo.class, fooUrn);
+    Optional<AspectFoo> oldSchemaResultLatest = oldSchemaDao.get(AspectFoo.class, fooUrn, 0);
+    Optional<AspectFoo> oldSchemaResultNonLatest = oldSchemaDao.get(AspectFoo.class, fooUrn, 1);
+
+    assertTrue(newSchemaResult.isPresent());
+    assertEquals(newSchemaResult.get().getValue(), "bar");
+    assertTrue(oldSchemaResultLatest.isPresent());
+    assertEquals(oldSchemaResultLatest.get().getValue(), "bar");
+    assertFalse(oldSchemaResultNonLatest.isPresent());
+  }
+
   @Nonnull
   private EbeanMetadataAspect getMetadata(Urn urn, String aspectName, long version, @Nullable RecordTemplate metadata) {
     EbeanMetadataAspect aspect = new EbeanMetadataAspect();