Add some TableImpl/SnapshotManager/Checkpointer skeletons

Author: scottsand-db

core/src/main/java/io/delta/core/Table.java

@@ -1,11 +1,12 @@
 package io.delta.core;
 
 import io.delta.core.helpers.TableHelper;
+import io.delta.core.internal.TableImpl;
 
 public interface Table {
 
     static Table forPath(String path, TableHelper helper) {
-        return null;
+        return TableImpl.forPath(path, helper);
     }
 
     Snapshot getLatestSnapshot();

core/src/main/java/io/delta/core/internal/SnapshotManager.java

@@ -0,0 +1,35 @@
+package io.delta.core.internal;
+
+import java.util.Optional;
+
+import io.delta.core.Snapshot;
+import io.delta.core.internal.checkpoint.CheckpointMetaData;
+
+public class SnapshotManager {
+
+    private final TableImpl tableImpl;
+    volatile private Snapshot currentSnapshot;
+
+    public SnapshotManager(TableImpl tableImpl) {
+        this.tableImpl = tableImpl;
+        this.currentSnapshot = getSnapshotAtInit();
+    }
+
+    /**
+     * Update current snapshot by applying the new delta files if any.
+     */
+    public Snapshot update() {
+
+    }
+
+    /**
+     * Load the Snapshot for this Delta table at initialization. This method uses the
+     * `lastCheckpoint` file as a hint on where to start listing the transaction log directory. If
+     * the _delta_log directory doesn't exist, this method will return an `InitialSnapshot`.
+     */
+    private Snapshot getSnapshotAtInit() {
+        final long currentTimestamp = System.currentTimeMillis();
+        final Optional<CheckpointMetaData> lastCheckpointOpt =
+            tableImpl.checkpointer.readLastCheckpointFile();
+    }
+}

core/src/main/java/io/delta/core/internal/TableImpl.java

@@ -0,0 +1,62 @@
+package io.delta.core.internal;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.locks.ReentrantLock;
+
+import io.delta.core.Snapshot;
+import io.delta.core.Table;
+import io.delta.core.helpers.TableHelper;
+import io.delta.core.internal.checkpoint.Checkpointer;
+
+public class TableImpl implements Table {
+
+    public static Table forPath(String path, TableHelper helper) {
+        final String logPath = path + "/_delta_log"; // TODO: FileUtils, makeQualified, etc.
+
+        return new TableImpl(logPath, path, helper);
+    }
+
+    public final String logPath;
+    public final String dataPath;
+    public final TableHelper tableHelper;
+    public final Checkpointer checkpointer;
+    public final SnapshotManager snapshotManager;
+
+    private final ReentrantLock lock;
+
+    public TableImpl(
+            String logPath,
+            String dataPath,
+            TableHelper tableHelper) {
+        this.logPath = logPath;
+        this.dataPath = dataPath;
+        this.tableHelper = tableHelper;
+
+        this.lock = new ReentrantLock();
+        this.checkpointer = new Checkpointer(this);
+        this.snapshotManager = new SnapshotManager(this);
+    }
+
+    @Override
+    public Snapshot getLatestSnapshot() {
+        return snapshotManager.update();
+    }
+
+    public <T> T lockInterruptibly(Callable<T> body) {
+        try {
+            lock.lockInterruptibly();
+
+            try {
+                return body.call();
+            } catch (Exception e) {
+                // failed body.call()
+                throw new RuntimeException(e);
+            } finally {
+                lock.unlock();
+            }
+        } catch (InterruptedException e) {
+            // failed lock.lockInterruptibly();
+            throw new RuntimeException(e);
+        }
+    }
+}

core/src/main/java/io/delta/core/internal/checkpoint/CheckpointMetaData.java

@@ -0,0 +1,11 @@
+package io.delta.core.internal.checkpoint;
+
+public class CheckpointMetaData {
+    public final long version;
+    public final long size;
+
+    public CheckpointMetaData(long version, long size) {
+        this.version = version;
+        this.size = size;
+    }
+}

core/src/main/java/io/delta/core/internal/checkpoint/Checkpointer.java

@@ -0,0 +1,23 @@
+package io.delta.core.internal.checkpoint;
+
+import java.util.Optional;
+
+import io.delta.core.internal.TableImpl;
+
+public class Checkpointer {
+    private final TableImpl tableImpl;
+
+    public Checkpointer(TableImpl tableImpl) {
+        this.tableImpl = tableImpl;
+    }
+
+    /** Returns information about the most recent checkpoint. */
+    public Optional<CheckpointMetaData> readLastCheckpointFile() {
+        return loadMetadataFromFile(0);
+    }
+
+    /** Loads the checkpoint metadata from the _last_checkpoint file. */
+    private Optional<CheckpointMetaData> loadMetadataFromFile(int tries) {
+
+    }
+}