refactor: remove isLatest field from ApplicationMeta

cdap-app-fabric/src/main/java/io/cdap/cdap/app/store/Store.java

@@ -368,6 +368,19 @@ List<ProgramHistory> getRuns(Collection<ProgramReference> programs, ProgramRunSt
    */
   int addApplication(ApplicationId id, ApplicationMeta meta) throws ConflictException;
 
+  /**
+   * Creates new application if it doesn't exist. Updates existing one otherwise.
+   * Marks the application as latest based on the isLatest param.
+   *
+   * @param id            application id
+   * @param meta          application metadata to store
+   * @param isLatest      boolean, indicating if the application should be marked latest
+   * @return              the number of edits to the application. A new application will return 0.
+   * @throws ConflictException if the app cannot be deployed when the user provided parent-version doesn't match the
+   *     current latest version
+   */
+  int addApplication(ApplicationId id, ApplicationMeta meta, boolean isLatest) throws ConflictException;
+
   /**
    * Marks existing applications as latest.
    *

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/deploy/pipeline/ApplicationRegistrationStage.java

@@ -67,9 +67,9 @@ public void process(ApplicationWithPrograms input) throws Exception {
     boolean ownerAdded = addOwnerIfRequired(input, allAppVersionsAppIds);
     ApplicationMeta appMeta = new ApplicationMeta(applicationSpecification.getName(),
         input.getSpecification(),
-        input.getChangeDetail(), input.getSourceControlMeta(), !input.isSkipMarkingLatest());
+        input.getChangeDetail(), input.getSourceControlMeta());
     try {
-      int editCount = store.addApplication(input.getApplicationId(), appMeta);
+      int editCount = store.addApplication(input.getApplicationId(), appMeta, !input.isSkipMarkingLatest());
 
       if (input.isSkipMarkingLatest()) {
         // TODO [CDAP-20848]

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/store/AppMetadataStore.java

@@ -413,8 +413,7 @@ public ApplicationMeta getLatest(ApplicationReference appReference) throws IOExc
     // the -SNAPSHOT version as the latest.
     List<Field<?>> fields = getApplicationPrimaryKeys(
         appReference.app(ApplicationId.DEFAULT_VERSION));
-    ApplicationMeta appMeta = appSpecTable.read(fields).map(row -> decodeRow(row, true))
-        .orElse(null);
+    ApplicationMeta appMeta = appSpecTable.read(fields).map(this::decodeRow).orElse(null);
     if (appMeta != null) {
       return appMeta;
     }
@@ -425,7 +424,7 @@ public ApplicationMeta getLatest(ApplicationReference appReference) throws IOExc
         StoreDefinition.AppMetadataStore.VERSION_FIELD,
         SortOrder.DESC)) {
       if (iterator.hasNext()) {
-        return decodeRow(iterator.next(), true);
+        return decodeRow(iterator.next());
       }
     }
     // This is the case when the app currently doesn't exist
@@ -640,10 +639,25 @@ public void markAsLatest(ApplicationId id)
    */
   public int createApplicationVersion(ApplicationId id, ApplicationMeta appMeta)
       throws IOException, ConflictException {
+    return createApplicationVersion(id, appMeta, true);
+  }
+
+  /**
+   * Persisting a new application version in the table.
+   *
+   * @param id the application id
+   * @param appMeta the application metadata to be written
+   * @param markAsLatest boolean, indicating if the application should be marked as latest
+   * @return the number of edits to the application. A new application will return 0.
+   * @throws IOException if failed to write app
+   * @throws ConflictException if parent-version provided in the request doesn't match the
+   *     latest version, do not allow app to be created
+   */
+  public int createApplicationVersion(ApplicationId id, ApplicationMeta appMeta, boolean markAsLatest)
+      throws IOException, ConflictException {
     String parentVersion = Optional.ofNullable(appMeta.getChange())
         .map(ChangeDetail::getParentVersion).orElse(null);
 
-    boolean markAsLatest = appMeta.getIsLatest();
     // Fetch the latest version
     ApplicationMeta latest = getLatest(id.getAppReference());
     String latestVersion = latest == null ? null : latest.getSpec().getAppVersion();
@@ -695,7 +709,7 @@ void writeApplication(String namespaceId, String appId, String versionId,
       @Nullable SourceControlMeta sourceControlMeta, boolean markAsLatest) throws IOException {
     writeApplicationSerialized(namespaceId, appId, versionId,
         GSON.toJson(
-            new ApplicationMeta(appId, spec, null, null, markAsLatest)),
+            new ApplicationMeta(appId, spec, null, null)),
         change, sourceControlMeta, markAsLatest);
     updateApplicationEdit(namespaceId, appId);
   }
@@ -2625,7 +2639,7 @@ private <T> List<T> scanWithRange(Range range, Type typeofT, StructuredTable tab
     return result;
   }
 
-  private ApplicationMeta decodeRow(StructuredRow row, boolean isLatest) {
+  private ApplicationMeta decodeRow(StructuredRow row) {
     String author = row.getString(StoreDefinition.AppMetadataStore.AUTHOR_FIELD);
     String changeSummary = row.getString(StoreDefinition.AppMetadataStore.CHANGE_SUMMARY_FIELD);
     Long creationTimeMillis = row.getLong(StoreDefinition.AppMetadataStore.CREATION_TIME_FIELD);
@@ -2646,12 +2660,7 @@ private ApplicationMeta decodeRow(StructuredRow row, boolean isLatest) {
       changeDetail = new ChangeDetail(changeSummary, null, author, creationTimeMillis, latest);
     }
 
-    return new ApplicationMeta(id, spec, changeDetail, sourceControl,
-        isLatest || Boolean.TRUE.equals(latest));
-  }
-
-  private ApplicationMeta decodeRow(StructuredRow row) {
-    return decodeRow(row, false);
+    return new ApplicationMeta(id, spec, changeDetail, sourceControl);
   }
 
   private void writeToStructuredTableWithPrimaryKeys(

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/store/ApplicationMeta.java

@@ -36,23 +36,13 @@ public class ApplicationMeta {
   private final ChangeDetail change;
   @Nullable
   private final SourceControlMeta sourceControlMeta;
-  // the isLatest field does not need to be serialized in the ApplicationMetadata object
-  // as it's stored as a separate column in the app spec table.
-  private final transient boolean isLatest;
 
   public ApplicationMeta(String id, ApplicationSpecification spec,
-      @Nullable ChangeDetail change, @Nullable SourceControlMeta sourceControlMeta,
-      boolean isLatest) {
+      @Nullable ChangeDetail change, @Nullable SourceControlMeta sourceControlMeta) {
     this.id = id;
     this.spec = spec;
     this.change = change;
     this.sourceControlMeta = sourceControlMeta;
-    this.isLatest = isLatest;
-  }
-
-  public ApplicationMeta(String id, ApplicationSpecification spec,
-      @Nullable ChangeDetail change, @Nullable SourceControlMeta sourceControlMeta) {
-    this(id, spec, change, sourceControlMeta, true);
   }
 
   public ApplicationMeta(String id, ApplicationSpecification spec, @Nullable ChangeDetail change) {
@@ -77,18 +67,13 @@ public SourceControlMeta getSourceControlMeta() {
     return sourceControlMeta;
   }
 
-  public boolean getIsLatest() {
-    return isLatest;
-  }
-
   @Override
   public String toString() {
     return Objects.toStringHelper(this)
         .add("id", id)
         .add("spec", ADAPTER.toJson(spec))
         .add("change", change)
         .add("sourceControlMeta", sourceControlMeta)
-        .add("isLatest", isLatest)
         .toString();
   }
 }

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/store/DefaultStore.java

@@ -600,6 +600,13 @@ public int addApplication(ApplicationId id, ApplicationMeta meta) throws Conflic
     }, ConflictException.class);
   }
 
+  @Override
+  public int addApplication(ApplicationId id, ApplicationMeta meta, boolean isLatest) throws ConflictException {
+    return TransactionRunners.run(transactionRunner, context -> {
+      return getAppMetadataStore(context).createApplicationVersion(id, meta, isLatest);
+    }, ConflictException.class);
+  }
+
   @Override
   public void updateApplicationSourceControlMeta(Map<ApplicationId, SourceControlMeta> updateRequests)
       throws IOException {

cdap-app-fabric/src/test/java/io/cdap/cdap/internal/app/store/AppMetadataStoreTest.java

@@ -1552,7 +1552,7 @@ public void testGetLatestOnLegacyRows() throws Exception {
     ArtifactId artifactId = NamespaceId.DEFAULT.artifact("testArtifact", "1.0").toApiArtifactId();
     ApplicationId appId = appRef.app(appVersion);
     ApplicationSpecification spec = createDummyAppSpec(appId.getApplication(), appId.getVersion(), artifactId);
-    ApplicationMeta appMeta = new ApplicationMeta(appName, spec, null, null, false);
+    ApplicationMeta appMeta = new ApplicationMeta(appName, spec, null, null);
 
     TransactionRunners.run(transactionRunner, context -> {
       AppMetadataStore metaStore = AppMetadataStore.create(context);
@@ -1573,7 +1573,6 @@ public void testGetLatestOnLegacyRows() throws Exception {
 
     Assert.assertEquals(appName, latestAppMeta.getId());
     Assert.assertEquals(appVersion, latestAppMeta.getSpec().getAppVersion());
-    Assert.assertTrue(latestAppMeta.getIsLatest());
   }
 
   /**

cdap-app-fabric/src/test/java/io/cdap/cdap/internal/app/store/DefaultStoreTest.java

@@ -523,15 +523,18 @@ public void testUpdateChangedApplication() throws ConflictException {
   public void testAddApplicationWithoutMarkingLatest()
       throws ConflictException {
     long creationTime = System.currentTimeMillis();
-    ApplicationId appId = new ApplicationId("account1", "app1");
-    ApplicationMeta appMeta = new ApplicationMeta("app1", Specifications.from(new FooApp()),
-        new ChangeDetail(null, null, null, creationTime), null, false);
-    store.addApplication(appId, appMeta);
+    String appName = "notLatestApp1";
+    ApplicationId appId = new ApplicationId("account1", appName, "v1");
+    ApplicationMeta appMeta = new ApplicationMeta(appName, Specifications.from(new FooApp()),
+        new ChangeDetail(null, null, null, creationTime), null);
+    store.addApplication(appId, appMeta, false);
 
     ApplicationMeta storedMeta = store.getApplicationMetadata(appId);
-    Assert.assertEquals("app1", storedMeta.getId());
+    Assert.assertEquals(appName, storedMeta.getId());
     Assert.assertEquals(creationTime, storedMeta.getChange().getCreationTimeMillis());
-    Assert.assertFalse(storedMeta.getIsLatest());
+
+    ApplicationMeta latestMeta = store.getLatest(appId.getAppReference());
+    Assert.assertNull(latestMeta);
   }
 
   @Test
@@ -541,26 +544,24 @@ public void testMarkApplicationsLatestWithNewApps()
     // Add 2 new applications without marking them latest
     ApplicationId appId1 = new ApplicationId("account1", "newApp1");
     ApplicationMeta appMeta1 = new ApplicationMeta("newApp1", Specifications.from(new FooApp()),
-        new ChangeDetail(null, null, null, creationTime), null, false);
-    store.addApplication(appId1, appMeta1);
+        new ChangeDetail(null, null, null, creationTime), null);
+    store.addApplication(appId1, appMeta1, false);
 
     ApplicationId appId2 = new ApplicationId("account1", "newApp2");
     ApplicationMeta appMeta2 = new ApplicationMeta("newApp2", Specifications.from(new FooApp()),
-        new ChangeDetail(null, null, null, creationTime), null, false);
-    store.addApplication(appId2, appMeta2);
+        new ChangeDetail(null, null, null, creationTime), null);
+    store.addApplication(appId2, appMeta2, false);
 
     // Now mark them as latest in bulk
     store.markApplicationsLatest(Arrays.asList(appId1, appId2));
 
-    ApplicationMeta storedMeta1 = store.getApplicationMetadata(appId1);
+    ApplicationMeta storedMeta1 = store.getLatest(appId1.getAppReference());
     Assert.assertEquals("newApp1", storedMeta1.getId());
     Assert.assertEquals(creationTime, storedMeta1.getChange().getCreationTimeMillis());
-    Assert.assertTrue(storedMeta1.getIsLatest());
 
-    ApplicationMeta storedMeta2 = store.getApplicationMetadata(appId2);
+    ApplicationMeta storedMeta2 = store.getLatest(appId2.getAppReference());
     Assert.assertEquals("newApp2", storedMeta2.getId());
     Assert.assertEquals(creationTime, storedMeta2.getChange().getCreationTimeMillis());
-    Assert.assertTrue(storedMeta2.getIsLatest());
   }
 
   @Test
@@ -581,23 +582,24 @@ public void testMarkApplicationsLatestWithExistingLatest()
     // Add a new version of the application without marking latest
     ApplicationId appIdV2 = new ApplicationId("account1", appName, newVersion);
     ApplicationMeta appMetaV2 = new ApplicationMeta(appName, Specifications.from(new FooApp(), appName, newVersion),
-        new ChangeDetail(null, null, null, v2CreationTime), null, false);
-    store.addApplication(appIdV2, appMetaV2);
+        new ChangeDetail(null, null, null, v2CreationTime), null);
+    store.addApplication(appIdV2, appMetaV2, false);
 
     // Now mark the new version as latest
     store.markApplicationsLatest(Collections.singletonList(appIdV2));
+    ApplicationMeta latestMeta = store.getLatest(appIdV1.getAppReference());
 
     ApplicationMeta storedMetaV1 = store.getApplicationMetadata(appIdV1);
     Assert.assertEquals(appName, storedMetaV1.getId());
     Assert.assertEquals(oldVersion, storedMetaV1.getSpec().getAppVersion());
     Assert.assertEquals(creationTime, storedMetaV1.getChange().getCreationTimeMillis());
-    Assert.assertFalse(storedMetaV1.getIsLatest());
+    Assert.assertNotEquals(latestMeta.getSpec().getAppVersion(), storedMetaV1.getSpec().getAppVersion());
 
     ApplicationMeta storedMetaV2 = store.getApplicationMetadata(appIdV2);
     Assert.assertEquals(appName, storedMetaV2.getId());
     Assert.assertEquals(newVersion, storedMetaV2.getSpec().getAppVersion());
     Assert.assertEquals(v2CreationTime, storedMetaV2.getChange().getCreationTimeMillis());
-    Assert.assertTrue(storedMetaV2.getIsLatest());
+    Assert.assertEquals(latestMeta.getSpec().getAppVersion(), storedMetaV2.getSpec().getAppVersion());
   }
 
   @Test(expected = ApplicationNotFoundException.class)