[iceberg] 03/09: Core: Fix snapshot log with intermediate transaction snapshots (#5568)

[bl...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

blue pushed a commit to branch 0.14.x
in repository https://gitbox.apache.org/repos/asf/iceberg.git

commit 7bb15a2d88654ed0346557db3f1d6935f83320f8
Author: Eduard Tudenhöfner <et...@gmail.com>
AuthorDate: Thu Aug 18 21:06:36 2022 +0200

    Core: Fix snapshot log with intermediate transaction snapshots (#5568)
---
 .../java/org/apache/iceberg/TableMetadata.java     |  8 ++--
 .../java/org/apache/iceberg/TestTransaction.java   | 50 ++++++++++++++--------
 2 files changed, 36 insertions(+), 22 deletions(-)

diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java b/core/src/main/java/org/apache/iceberg/TableMetadata.java
index 8c1d8f5dbf..f49b64dcb5 100644
--- a/core/src/main/java/org/apache/iceberg/TableMetadata.java
+++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java
@@ -1481,9 +1481,11 @@ public class TableMetadata implements Serializable {
       List<HistoryEntry> newSnapshotLog = Lists.newArrayList();
       for (HistoryEntry logEntry : snapshotLog) {
         long snapshotId = logEntry.snapshotId();
-        if (snapshotsById.containsKey(snapshotId) && !intermediateSnapshotIds.contains(snapshotId)) {
-          // copy the log entries that are still valid
-          newSnapshotLog.add(logEntry);
+        if (snapshotsById.containsKey(snapshotId)) {
+          if (!intermediateSnapshotIds.contains(snapshotId)) {
+            // copy the log entries that are still valid
+            newSnapshotLog.add(logEntry);
+          }
         } else {
           // any invalid entry causes the history before it to be removed. otherwise, there could be
           // history gaps that cause time-travel queries to produce incorrect results. for example,
diff --git a/core/src/test/java/org/apache/iceberg/TestTransaction.java b/core/src/test/java/org/apache/iceberg/TestTransaction.java
index 6c0fd69312..a11df6b628 100644
--- a/core/src/test/java/org/apache/iceberg/TestTransaction.java
+++ b/core/src/test/java/org/apache/iceberg/TestTransaction.java
@@ -92,22 +92,25 @@ public class TestTransaction extends TableTestBase {
   public void testMultipleOperationTransaction() {
     Assert.assertEquals("Table should be on version 0", 0, (int) version());
 
+    table.newAppend().appendFile(FILE_C).commit();
+    List<HistoryEntry> initialHistory = table.history();
+
     TableMetadata base = readMetadata();
 
     Transaction txn = table.newTransaction();
 
-    Assert.assertSame("Base metadata should not change when commit is created",
-        base, readMetadata());
-    Assert.assertEquals("Table should be on version 0 after txn create", 0, (int) version());
+    Assert.assertSame(
+        "Base metadata should not change when commit is created", base, readMetadata());
+    Assert.assertEquals("Table should be on version 1 after txn create", 1, (int) version());
 
     txn.newAppend()
         .appendFile(FILE_A)
         .appendFile(FILE_B)
         .commit();
 
-    Assert.assertSame("Base metadata should not change when commit is created",
-        base, readMetadata());
-    Assert.assertEquals("Table should be on version 0 after txn create", 0, (int) version());
+    Assert.assertSame(
+        "Base metadata should not change when commit is created", base, readMetadata());
+    Assert.assertEquals("Table should be on version 1 after txn create", 1, (int) version());
 
     Snapshot appendSnapshot = txn.table().currentSnapshot();
 
@@ -117,26 +120,35 @@ public class TestTransaction extends TableTestBase {
 
     Snapshot deleteSnapshot = txn.table().currentSnapshot();
 
-    Assert.assertSame("Base metadata should not change when an append is committed",
-        base, readMetadata());
-    Assert.assertEquals("Table should be on version 0 after append", 0, (int) version());
+    Assert.assertSame(
+        "Base metadata should not change when an append is committed", base, readMetadata());
+    Assert.assertEquals("Table should be on version 1 after append", 1, (int) version());
 
     txn.commitTransaction();
 
-    Assert.assertEquals("Table should be on version 1 after commit", 1, (int) version());
-    Assert.assertEquals("Table should have one manifest after commit",
-        1, readMetadata().currentSnapshot().allManifests(table.io()).size());
-    Assert.assertEquals("Table snapshot should be the delete snapshot",
-        deleteSnapshot, readMetadata().currentSnapshot());
-    validateManifestEntries(readMetadata().currentSnapshot().allManifests(table.io()).get(0),
+    Assert.assertEquals("Table should be on version 2 after commit", 2, (int) version());
+    Assert.assertEquals(
+        "Table should have two manifest after commit",
+        2,
+        readMetadata().currentSnapshot().allManifests(table.io()).size());
+    Assert.assertEquals(
+        "Table snapshot should be the delete snapshot",
+        deleteSnapshot,
+        readMetadata().currentSnapshot());
+    validateManifestEntries(
+        readMetadata().currentSnapshot().allManifests(table.io()).get(0),
         ids(deleteSnapshot.snapshotId(), appendSnapshot.snapshotId()),
         files(FILE_A, FILE_B), statuses(Status.DELETED, Status.EXISTING));
 
-    Assert.assertEquals("Table should have a snapshot for each operation",
-        2, readMetadata().snapshots().size());
-    validateManifestEntries(readMetadata().snapshots().get(0).allManifests(table.io()).get(0),
+    Assert.assertEquals(
+        "Table should have a snapshot for each operation", 3, readMetadata().snapshots().size());
+    validateManifestEntries(
+        readMetadata().snapshots().get(1).allManifests(table.io()).get(0),
         ids(appendSnapshot.snapshotId(), appendSnapshot.snapshotId()),
-        files(FILE_A, FILE_B), statuses(Status.ADDED, Status.ADDED));
+        files(FILE_A, FILE_B),
+        statuses(Status.ADDED, Status.ADDED));
+
+    org.assertj.core.api.Assertions.assertThat(table.history()).containsAll(initialHistory);
   }
 
   @Test