Core: Ensure reactivated view version uses correct timestamp (#12821)

Author: lliangyu-lin

core/src/main/java/org/apache/iceberg/view/ViewMetadata.java

@@ -249,9 +249,17 @@ public Builder setCurrentVersionId(int newVersionId) {
         changes.add(new MetadataUpdate.SetCurrentViewVersion(newVersionId));
       }
 
+      // Use the timestamp from the view version if it was added in current set of changes.
+      // Otherwise, use the current system time. This handles cases where the view version
+      // was set as current in the past and is being re-activated.
+      boolean versionAddedInThisChange =
+          changes(MetadataUpdate.AddViewVersion.class)
+              .anyMatch(added -> added.viewVersion().versionId() == newVersionId);
+
       this.historyEntry =
           ImmutableViewHistoryEntry.builder()
-              .timestampMillis(version.timestampMillis())
+              .timestampMillis(
+                  versionAddedInThisChange ? version.timestampMillis() : System.currentTimeMillis())
               .versionId(version.versionId())
               .build();
 

core/src/test/java/org/apache/iceberg/view/TestViewMetadata.java

@@ -42,9 +42,13 @@ private ViewVersion newViewVersion(int id, String sql) {
   }
 
   private ViewVersion newViewVersion(int id, int schemaId, String sql) {
+    return newViewVersion(id, schemaId, System.currentTimeMillis(), sql);
+  }
+
+  private ViewVersion newViewVersion(int id, int schemaId, long timestampMillis, String sql) {
     return ImmutableViewVersion.builder()
         .versionId(id)
-        .timestampMillis(System.currentTimeMillis())
+        .timestampMillis(timestampMillis)
         .defaultCatalog("prod")
         .defaultNamespace(Namespace.of("default"))
         .putSummary("user", "some-user")
@@ -396,6 +400,70 @@ public void viewVersionHistoryIsCorrectlyRetained() {
         .hasMessage("Cannot set current version to unknown version: 1");
   }
 
+  @Test
+  public void versionHistoryEntryMaintainCorrectTimeline() {
+    ViewVersion viewVersionOne = newViewVersion(1, 0, 1000, "select * from ns.tbl");
+    ViewVersion viewVersionTwo = newViewVersion(2, 0, 2000, "select count(*) from ns.tbl");
+    ViewVersion viewVersionThree =
+        newViewVersion(3, 0, 3000, "select count(*) as count from ns.tbl");
+
+    ViewMetadata viewMetadata =
+        ViewMetadata.builder()
+            .setLocation("location")
+            .addSchema(new Schema(Types.NestedField.required(1, "x", Types.LongType.get())))
+            .addVersion(viewVersionOne)
+            .addVersion(viewVersionTwo)
+            .setCurrentVersionId(1)
+            .build();
+
+    // setting an existing view version as the new current should update the timestamp in the
+    // history
+    ViewMetadata updated = ViewMetadata.buildFrom(viewMetadata).setCurrentVersionId(2).build();
+
+    List<ViewHistoryEntry> history = updated.history();
+    assertThat(history)
+        .hasSize(2)
+        .element(0)
+        .isEqualTo(ImmutableViewHistoryEntry.builder().versionId(1).timestampMillis(1000).build());
+    assertThat(history)
+        .element(1)
+        .satisfies(
+            v -> {
+              assertThat(v.versionId()).isEqualTo(2);
+              assertThat(v.timestampMillis())
+                  .isGreaterThan(3000)
+                  .isLessThanOrEqualTo(System.currentTimeMillis());
+            });
+
+    // adding a new view version and setting it as current should use the view version's timestamp
+    // in the history (which has been set to a fixed value for testing)
+    updated =
+        ViewMetadata.buildFrom(updated).addVersion(viewVersionThree).setCurrentVersionId(3).build();
+    List<ViewHistoryEntry> historyTwo = updated.history();
+    assertThat(historyTwo)
+        .hasSize(3)
+        .containsAll(history)
+        .element(2)
+        .isEqualTo(ImmutableViewHistoryEntry.builder().versionId(3).timestampMillis(3000).build());
+
+    // setting an older view version as the new current (aka doing a rollback) should update the
+    // timestamp in the history
+    ViewMetadata reactiveOldViewVersion =
+        ViewMetadata.buildFrom(updated).setCurrentVersionId(1).build();
+    List<ViewHistoryEntry> historyThree = reactiveOldViewVersion.history();
+    assertThat(historyThree)
+        .hasSize(4)
+        .containsAll(historyTwo)
+        .element(3)
+        .satisfies(
+            v -> {
+              assertThat(v.versionId()).isEqualTo(1);
+              assertThat(v.timestampMillis())
+                  .isGreaterThan(3000)
+                  .isLessThanOrEqualTo(System.currentTimeMillis());
+            });
+  }
+
   @Test
   public void versionsAddedInCurrentBuildAreRetained() {
     ViewVersion v1 = newViewVersion(1, "select 1 as count");

core/src/test/java/org/apache/iceberg/view/TestViewMetadataParser.java

@@ -106,14 +106,14 @@ public void readAndWriteValidViewMetadata() throws Exception {
                     .assignUUID("fa6506c3-7681-40c8-86dc-e36561f83385")
                     .addSchema(TEST_SCHEMA)
                     .addVersion(version1)
-                    .addVersion(version2)
                     .setLocation("s3://bucket/test/location")
                     .setProperties(
                         ImmutableMap.of(
                             "some-key", "some-value", ViewProperties.COMMENT, "some-comment"))
                     .setCurrentVersionId(1)
                     .upgradeFormatVersion(1)
                     .build())
+            .addVersion(version2)
             .setCurrentVersionId(2)
             .build();
 
@@ -222,7 +222,6 @@ public void viewMetadataWithMetadataLocation() throws Exception {
                             .assignUUID("fa6506c3-7681-40c8-86dc-e36561f83385")
                             .addSchema(TEST_SCHEMA)
                             .addVersion(version1)
-                            .addVersion(version2)
                             .setLocation("s3://bucket/test/location")
                             .setProperties(
                                 ImmutableMap.of(
@@ -233,6 +232,7 @@ public void viewMetadataWithMetadataLocation() throws Exception {
                             .setCurrentVersionId(1)
                             .upgradeFormatVersion(1)
                             .build())
+                    .addVersion(version2)
                     .setCurrentVersionId(2)
                     .build())
             .setMetadataLocation(metadataLocation)