Add more unit tests

Author: chenjunjiedada

core/src/main/java/org/apache/iceberg/BaseSnapshot.java

@@ -84,24 +84,24 @@ class BaseSnapshot implements Snapshot {
                Map<String, String> summary,
                InputFile manifestList,
                Long sequenceNumber) {
-    this(ops, snapshotId, parentId, timestampMillis, operation, summary, manifestList);
+    this(io, snapshotId, parentId, timestampMillis, operation, summary, manifestList);
     this.sequenceNumber = sequenceNumber;
   }
 
-  BaseSnapshot(TableOperations ops,
+  BaseSnapshot(FileIO io,
                long snapshotId,
                Long parentId,
                long timestampMillis,
                String operation,
                Map<String, String> summary,
                List<ManifestFile> manifests,
                Long sequenceNumber) {
-    this(ops, snapshotId, parentId, timestampMillis, operation, summary, (InputFile) null);
+    this(io, snapshotId, parentId, timestampMillis, operation, summary, (InputFile) null);
     this.manifests = manifests;
     this.sequenceNumber = sequenceNumber;
   }
 
-  BaseSnapshot(TableOperations ops,
+  BaseSnapshot(FileIO io,
                long snapshotId,
                Long parentId,
                long timestampMillis,

core/src/main/java/org/apache/iceberg/util/JsonUtil.java

@@ -76,7 +76,7 @@ public static Long getLongOrNull(String property, JsonNode node) {
     }
     JsonNode pNode = node.get(property);
     Preconditions.checkArgument(pNode != null && !pNode.isNull() && pNode.isNumber() && pNode.canConvertToLong(),
-        "Cannot parse %s from non-string value: %s", property, pNode);
+        "Cannot parse %s from non-numeric value: %s", property, pNode);
     return pNode.asLong();
   }
 

core/src/test/java/org/apache/iceberg/TestSequenceNumber.java

@@ -19,42 +19,149 @@
 
 package org.apache.iceberg;
 
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import org.junit.Assert;
 import org.junit.Test;
 
-
 public class TestSequenceNumber extends TableTestBase {
 
   @Test
-  public void testWriteSequenceNumber() {
+  public void testReadWriteSequenceNumber() {
     table.newFastAppend().appendFile(FILE_A).commit();
+    Assert.assertEquals(1, table.currentSnapshot().sequenceNumber().longValue());
+    table.newFastAppend().appendFile(FILE_B).commit();
+    Assert.assertEquals(2, table.currentSnapshot().sequenceNumber().longValue());
+  }
 
-    Assert.assertEquals("sequence number should be 1", 1,
-        table.currentSnapshot().sequenceNumber().longValue());
+  @Test
+  public void testCommitConflict() {
+    Transaction txn = table.newTransaction();
 
+    txn.newFastAppend().appendFile(FILE_A).apply();
     table.newFastAppend().appendFile(FILE_B).commit();
 
-    Assert.assertEquals("sequence number should be 2", 2,
-        table.currentSnapshot().sequenceNumber().longValue());
+    AssertHelpers.assertThrows("Should failed due to conflict",
+        IllegalStateException.class, "last operation has not committed", txn::commitTransaction);
+
+    Assert.assertEquals(1, TestTables.load(tableDir, "test").currentSnapshot().sequenceNumber()
+        .longValue());
+  }
+
+  @Test
+  public void testConcurrentCommit() throws InterruptedException {
+    ExecutorService threadPool = Executors.newFixedThreadPool(4);
+    List<Callable<Void>> tasks = new ArrayList<>();
+
+    Callable<Void> write1 = () -> {
+      Transaction txn = table.newTransaction();
+      txn.newFastAppend().appendFile(FILE_A).commit();
+      txn.commitTransaction();
+      return null;
+    };
+
+    Callable<Void> write2 = () -> {
+      Transaction txn = table.newTransaction();
+      txn.newAppend().appendFile(FILE_B).commit();
+      txn.commitTransaction();
+      return null;
+    };
+
+    Callable<Void> write3 = () -> {
+      Transaction txn = table.newTransaction();
+      txn.newDelete().deleteFile(FILE_A).commit();
+      txn.commitTransaction();
+      return null;
+    };
+
+    Callable<Void> write4 = () -> {
+      Transaction txn = table.newTransaction();
+      txn.newOverwrite().addFile(FILE_D).commit();
+      txn.commitTransaction();
+      return null;
+    };
+
+    tasks.add(write1);
+    tasks.add(write2);
+    tasks.add(write3);
+    tasks.add(write4);
+    threadPool.invokeAll(tasks);
+    threadPool.shutdown();
 
+    Assert.assertEquals(4, TestTables.load(tableDir, "test").currentSnapshot().sequenceNumber()
+        .longValue());
   }
 
   @Test
-  public void testReadSequenceNumber() {
-    long curSeqNum;
-
-    if (table.currentSnapshot() == null) {
-      curSeqNum = 0;
-    } else {
-      curSeqNum = table.currentSnapshot().sequenceNumber();
-    }
+  public void testRollBack() {
     table.newFastAppend().appendFile(FILE_A).commit();
+    long snapshotId = table.currentSnapshot().snapshotId();
     table.newFastAppend().appendFile(FILE_B).commit();
 
-    Assert.assertEquals(curSeqNum + 2,
-        TestTables.load(tableDir, "test")
-            .currentSnapshot()
-            .sequenceNumber()
-            .longValue());
+    Assert.assertEquals(2, TestTables.load(tableDir, "test").currentSnapshot().sequenceNumber()
+        .longValue());
+
+    table.rollback().toSnapshotId(snapshotId).commit();
+
+    Assert.assertEquals(1, TestTables.load(tableDir, "test").currentSnapshot().sequenceNumber()
+        .longValue());
+  }
+
+  @Test
+  public void testMultipleTxnOperations() {
+    Snapshot snapshot;
+    Transaction txn = table.newTransaction();
+    txn.newOverwrite().addFile(FILE_A).commit();
+    txn.commitTransaction();
+    Assert.assertEquals(1, TestTables.load(tableDir, "test").currentSnapshot().sequenceNumber()
+        .longValue());
+
+    txn = table.newTransaction();
+    Set<DataFile> toAddFiles = new HashSet<>();
+    Set<DataFile> toDeleteFiles = new HashSet<>();
+    toAddFiles.add(FILE_B);
+    toDeleteFiles.add(FILE_A);
+    txn.newRewrite().rewriteFiles(toDeleteFiles, toAddFiles).commit();
+    txn.commitTransaction();
+    Assert.assertEquals(2, TestTables.load(tableDir, "test").currentSnapshot().sequenceNumber()
+        .longValue());
+
+    txn = table.newTransaction();
+    txn.newReplacePartitions().addFile(FILE_C).commit();
+    txn.commitTransaction();
+    Assert.assertEquals(3, TestTables.load(tableDir, "test").currentSnapshot().sequenceNumber()
+        .longValue());
+
+    txn = table.newTransaction();
+    txn.newDelete().deleteFile(FILE_C).commit();
+    txn.commitTransaction();
+
+    Assert.assertEquals(4, TestTables.load(tableDir, "test").currentSnapshot().sequenceNumber()
+        .longValue());
+
+    txn = table.newTransaction();
+    txn.newAppend().appendFile(FILE_C).commit();
+    txn.commitTransaction();
+    Assert.assertEquals(5, TestTables.load(tableDir, "test").currentSnapshot().sequenceNumber()
+        .longValue());
+
+    snapshot = table.currentSnapshot();
+
+    txn = table.newTransaction();
+    txn.newOverwrite().addFile(FILE_D).deleteFile(FILE_C).commit();
+    txn.commitTransaction();
+    Assert.assertEquals(6, TestTables.load(tableDir, "test").currentSnapshot().sequenceNumber()
+        .longValue());
+
+    txn = table.newTransaction();
+    txn.expireSnapshots().expireOlderThan(snapshot.timestampMillis()).commit();
+    txn.commitTransaction();
+    Assert.assertEquals(6, TestTables.load(tableDir, "test").currentSnapshot().sequenceNumber()
+        .longValue());
   }
 }