Update table metadata from changes in UDB

This allows the sync command to update the metadata of existing
partitions, not only new partitions. This means, if an existing
partition is updated in hive, it will be detected and updated during a
sync command.

For this functionality, table version numbers and partition version
numbers have been added. The concurrency model was also updated for
these changes. It is an optimistic concurrency control model, where
multiple updates can be concurrent, but only 1 will succeed in updating
the metadata to the next version.

pr-link: #10836
change-id: cid-6c29985d05a17081014183ca87dd83516c36656c

core/common/src/test/java/alluxio/AlluxioTestDirectory.java

@@ -44,8 +44,19 @@ public final class AlluxioTestDirectory {
    * @return the created directory
    */
   public static File createTemporaryDirectory(String prefix) {
-    final File file = new File(ALLUXIO_TEST_DIRECTORY, prefix + "-" + UUID.randomUUID());
-    if (!file.mkdir()) {
+    return createTemporaryDirectory(ALLUXIO_TEST_DIRECTORY, prefix);
+  }
+
+  /**
+   * Creates a directory with the given prefix inside the Alluxio temporary directory.
+   *
+   * @param parent a parent directory
+   * @param prefix a prefix to use in naming the temporary directory
+   * @return the created directory
+   */
+  public static File createTemporaryDirectory(File parent, String prefix) {
+    final File file = new File(parent, prefix + "-" + UUID.randomUUID());
+    if (!file.mkdirs()) {
       throw new RuntimeException("Failed to create directory " + file.getAbsolutePath());
     }
 

core/server/common/src/main/java/alluxio/master/journal/JournalEntryAssociation.java

@@ -59,6 +59,7 @@ public static String getMasterForEntry(JournalEntry entry) {
     }
     if (entry.hasAttachDb()
         || entry.hasAddTable()
+        || entry.hasRemoveTable()
         || entry.hasDetachDb()
         || entry.hasUpdateDatabaseInfo()
         || entry.hasAddTransformJobInfo()

core/server/common/src/test/java/alluxio/master/journal/JournalEntryAssociationTest.java

@@ -86,6 +86,7 @@ public class JournalEntryAssociationTest {
       JournalEntry.newBuilder().setPersistDirectory(PersistDirectoryEntry.getDefaultInstance()).build(),
       JournalEntry.newBuilder().setRemovePathProperties(RemovePathPropertiesEntry.getDefaultInstance()).build(),
       JournalEntry.newBuilder().setRemoveSyncPoint(RemoveSyncPointEntry.getDefaultInstance()).build(),
+      JournalEntry.newBuilder().setRemoveTable(Table.RemoveTableEntry.getDefaultInstance()).build(),
       JournalEntry.newBuilder().setRename(RenameEntry.getDefaultInstance()).build(),
       JournalEntry.newBuilder().setSetAcl(SetAclEntry.getDefaultInstance()).build(),
       JournalEntry.newBuilder().setSetAttribute(SetAttributeEntry.getDefaultInstance()).build(),

core/transport/src/grpc/table/table_master.proto

@@ -35,6 +35,7 @@ message Database {
     optional string comment = 7;
 }
 
+// next available id: 12
 message TableInfo {
     optional string db_name = 1;
     optional string table_name = 2;
@@ -50,6 +51,10 @@ message TableInfo {
 
     // partitioning scheme
     repeated FieldSchema partition_cols = 8;
+
+    optional int64 previous_version = 9;
+    optional int64 version = 10;
+    optional int64 version_creation_time = 11;
 }
 
 // TODO(gpang): update
@@ -74,13 +79,16 @@ message Transformation {
     optional string definition = 2;
 }
 
+// next available id: 6
 message Partition {
     optional PartitionSpec partition_spec = 1;
     optional Layout base_layout = 2;
     /**
      * The latest transformation is in the back of the list.
      */
     repeated Transformation transformations = 3;
+    optional int64 version = 4;
+    optional int64 version_creation_time = 5;
 }
 
 message ColumnStatisticsInfo {