Parallel calls

sumeet-db · Sep 11, 2024 · 3bd4a6e · 3bd4a6e
1 parent d7b94cd
commit 3bd4a6e
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 80 deletions.
diff --git a/spark/src/main/java/io/delta/dynamodbcommitcoordinator/CoordinatedCommitsUtils.java b/spark/src/main/java/io/delta/dynamodbcommitcoordinator/CoordinatedCommitsUtils.java
diff --git a/...orage/commit/CoordinatedCommitsUtils.java → ...orage/commit/CoordinatedCommitsUtils.java b/...orage/commit/CoordinatedCommitsUtils.java → ...orage/commit/CoordinatedCommitsUtils.java
@@ -33,12 +33,24 @@ public class CoordinatedCommitsUtils {
     private CoordinatedCommitsUtils() {}
 
     /** The subdirectory in which to store the unbackfilled commit files. */
-    final static String COMMIT_SUBDIR = "_commits";
+    final static public String COMMIT_SUBDIR = "_commits";
 
-    /** The configuration key for the coordinated commits owner. */
-    private static final String COORDINATED_COMMITS_COORDINATOR_CONF_KEY =
+    /** The configuration key for the coordinated commits owner name. */
+    private static final String COORDINATED_COMMITS_COORDINATOR_NAME_KEY =
             "delta.coordinatedCommits.commitCoordinator-preview";
 
+    /**
+     * Creates a new unbackfilled delta file path for the given commit version.
+     * The path is of the form `tablePath/_delta_log/_commits/00000000000000000001.uuid.json`.
+     */
+    public static Path generateUnbackfilledDeltaFilePath(
+            Path logPath,
+            long version) {
+        String uuid = UUID.randomUUID().toString();
+        Path basePath = new Path(logPath, COMMIT_SUBDIR);
+        return new Path(basePath, String.format("%020d.%s.json", version, uuid));
+    }
+
     /**
      * Returns the path to the backfilled delta file for the given commit version.
      * The path is of the form `tablePath/_delta_log/00000000000000000001.json`.
@@ -111,7 +123,7 @@ public static Path commitDirPath(Path logPath) {
     private static String getCoordinator(AbstractMetadata metadata) {
         String coordinator = metadata
                 .getConfiguration()
-                .get(COORDINATED_COMMITS_COORDINATOR_CONF_KEY);
+                .get(COORDINATED_COMMITS_COORDINATOR_NAME_KEY);
         return coordinator != null ? coordinator : "";
     }
 }
diff --git a/storage/src/test/scala/io/delta/storage/commit/InMemoryCommitCoordinator.scala b/storage/src/test/scala/io/delta/storage/commit/InMemoryCommitCoordinator.scala
@@ -22,6 +22,7 @@ import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.locks.ReentrantReadWriteLock
 
 import io.delta.storage.LogStore
+import io.delta.storage.commit.CoordinatedCommitsUtils
 import io.delta.storage.commit.actions.AbstractMetadata
 import io.delta.storage.commit.actions.AbstractProtocol
 import org.apache.hadoop.conf.Configuration