WIP

Author: scottsand-db

build.sbt

@@ -802,7 +802,9 @@ lazy val core = (project in file("core"))
     name := "delta-core",
     commonSettings,
     skipReleaseSettings,
-    libraryDependencies ++= Seq()
+    libraryDependencies ++= Seq(
+//      "com.google.guava" % "guava" % "31.1-jre" // Streams.stream
+    )
   )
 
 lazy val defaultCore = (project in file("default-core"))
@@ -811,5 +813,8 @@ lazy val defaultCore = (project in file("default-core"))
     name := "delta-core-default",
     commonSettings,
     skipReleaseSettings,
-    libraryDependencies ++= Seq()
+    libraryDependencies ++= Seq(
+
+      "com.fasterxml.jackson.core" % "jackson-databind" % "2.14.2" // ObjectMapper
+    )
   )

core/src/main/java/io/delta/core/fs/FileStatus.java

@@ -2,9 +2,9 @@
 
 public interface FileStatus {
 
-    String filePath();
+    String path();
 
-    long size();
+    long length();
 
     long modificationTime();
 }

core/src/main/java/io/delta/core/helpers/TableHelper.java

@@ -1,5 +1,7 @@
 package io.delta.core.helpers;
 
+import java.io.FileNotFoundException;
+
 import io.delta.core.data.Row;
 import io.delta.core.expressions.Expression;
 import io.delta.core.fs.FileStatus;
@@ -8,11 +10,17 @@
 
 public interface TableHelper {
 
-    CloseableIterator<FileStatus> listFiles(String path);
+    CloseableIterator<FileStatus> listFiles(String path) throws FileNotFoundException;
 
-    CloseableIterator<FileStatus> listFiles(String path, String prefixToListFrom);
+    CloseableIterator<FileStatus> listFiles(String path, String prefixToListFrom) throws FileNotFoundException;
 
+    // DECISION 1 OPTION 1
     CloseableIterator<Row> readJsonFile(String path, StructType readSchema);
+    // each class needs a fromRow()
+
+    // DECISION 1 OPTION 2
+    CloseableIterator<String> readJsonFile(String path);
+    <T> T fromJson(String json, TypeReference<T> typeReference);
 
     /** Uses the readSchema for partition pruning. */
     CloseableIterator<Row> readParquetFile(String path, StructType readSchema);

core/src/main/java/io/delta/core/helpers/TypeReference.java

@@ -0,0 +1,18 @@
+package io.delta.core.helpers;
+
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+
+public abstract class TypeReference<T> {
+
+    private final Type type;
+
+    public TypeReference() {
+        Type superclass = getClass().getGenericSuperclass();
+        type = ((ParameterizedType) superclass).getActualTypeArguments()[0];
+    }
+
+    public Type getType() {
+        return type;
+    }
+}

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

@@ -0,0 +1,44 @@
+package io.delta.core.internal;
+
+import java.util.List;
+import java.util.Optional;
+
+import io.delta.core.fs.FileStatus;
+
+public class LogSegment {
+    public final String logPath;
+    public final long version;
+    public final List<FileStatus> deltas;
+    public final List<FileStatus> checkpoints;
+    public final Optional<Long> checkpointVersionOpt;
+    public final long lastCommitTimestamp;
+
+    /**
+     * Provides information around which files in the transaction log need to be read to create
+     * the given version of the log.
+     *
+     * @param logPath The path to the _delta_log directory
+     * @param version The Snapshot version to generate
+     * @param deltas The delta commit files (.json) to read
+     * @param checkpoints The checkpoint file(s) to read
+     * @param checkpointVersionOpt The checkpoint version used to start replay
+     * @param lastCommitTimestamp The "unadjusted" timestamp of the last commit within this segment.
+     *                            By unadjusted, we mean that the commit timestamps may not
+     *                            necessarily be monotonically increasing for the commits within
+     *                            this segment.
+     */
+    public LogSegment(
+            String logPath,
+            long version,
+            List<FileStatus> deltas,
+            List<FileStatus> checkpoints,
+            Optional<Long> checkpointVersionOpt,
+            long lastCommitTimestamp) {
+        this.logPath = logPath;
+        this.version = version;
+        this.deltas = deltas;
+        this.checkpoints = checkpoints;
+        this.checkpointVersionOpt = checkpointVersionOpt;
+        this.lastCommitTimestamp = lastCommitTimestamp;
+    }
+}