Delta core 1

build.sbt

@@ -84,10 +84,10 @@ lazy val commonSettings = Seq(
   // Can be run explicitly via: build/sbt $module/checkstyle
   // Will automatically be run during compilation (e.g. build/sbt compile)
   // and during tests (e.g. build/sbt test)
-  checkstyleConfigLocation := CheckstyleConfigLocation.File("dev/checkstyle.xml"),
-  checkstyleSeverityLevel := Some(CheckstyleSeverityLevel.Error),
-  (Compile / checkstyle) := (Compile / checkstyle).triggeredBy(Compile / compile).value,
-  (Test / checkstyle) := (Test / checkstyle).triggeredBy(Test / compile).value
+//  checkstyleConfigLocation := CheckstyleConfigLocation.File("dev/checkstyle.xml"),
+//  checkstyleSeverityLevel := Some(CheckstyleSeverityLevel.Error),
+//  (Compile / checkstyle) := (Compile / checkstyle).triggeredBy(Compile / compile).value,
+//  (Test / checkstyle) := (Test / checkstyle).triggeredBy(Test / compile).value
 )
 
 lazy val releaseSettings = Seq(
@@ -797,19 +797,32 @@ lazy val flink = (project in file("flink"))
     (Test / test) := ((Test / test) dependsOn (Compile / unidoc)).value
   )
 
-lazy val core = (project in file("core"))
+lazy val kernel = (project in file("kernel"))
   .settings(
-    name := "delta-core",
+    name := "delta-kernel",
     commonSettings,
     skipReleaseSettings,
-    libraryDependencies ++= Seq()
+    libraryDependencies ++= Seq(
+
+    )
   )
 
-lazy val defaultCore = (project in file("default-core"))
-  .dependsOn(core)
+lazy val kernelDefault = (project in file("kernel-default"))
+  .dependsOn(kernel)
   .settings(
-    name := "delta-core-default",
+    name := "delta-kernel-default",
     commonSettings,
     skipReleaseSettings,
-    libraryDependencies ++= Seq()
+    libraryDependencies ++= Seq(
+      "org.apache.hadoop" % "hadoop-client-api" % "3.3.1", // Configuration, Path
+      "io.delta" % "delta-storage" % "2.2.0", // LogStore
+      "com.fasterxml.jackson.core" % "jackson-databind" % "2.13.5", // ObjectMapper
+
+      "org.scalatest" %% "scalatest" % "3.2.15" % "test",
+      "io.delta" %% "delta-core" % "2.2.0" % "test",
+      "org.apache.spark" %% "spark-sql" % "3.3.2" % "test", // SparkSession
+      "org.apache.spark" %% "spark-sql" % "3.3.2" % "test" classifier "tests",
+      "org.apache.spark" %% "spark-core" % "3.3.2" % "test" classifier "tests",
+      "org.apache.spark" %% "spark-catalyst" % "3.3.2" % "test" classifier "tests",
+    )
   )

build/sbt-config/repositories

@@ -7,5 +7,5 @@
   typesafe-ivy-releases: https://repo.typesafe.com/typesafe/ivy-releases/, [organization]/[module]/[revision]/[type]s/[artifact](-[classifier]).[ext], bootOnly
   sbt-ivy-snapshots: https://repo.scala-sbt.org/scalasbt/ivy-snapshots/, [organization]/[module]/[revision]/[type]s/[artifact](-[classifier]).[ext], bootOnly
   sbt-plugin-releases: https://repo.scala-sbt.org/scalasbt/sbt-plugin-releases/, [organization]/[module]/(scala_[scalaVersion]/)(sbt_[sbtVersion]/)[revision]/[type]s/[artifact](-[classifier]).[ext]
-  bintray-spark-packages: https://dl.bintray.com/spark-packages/maven/
+  repos-spark-packages: https://repos.spark-packages.org
   typesafe-releases: http://repo.typesafe.com/typesafe/releases/

kernel-default/src/main/java/io/delta/core/data/JsonRow.java

@@ -0,0 +1,43 @@
+package io.delta.core.data;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.delta.core.types.DataType;
+import io.delta.core.types.LongType;
+import io.delta.core.types.StructField;
+import io.delta.core.types.StructType;
+
+public class JsonRow implements Row {
+
+    // TODO: we can do this cleaner / smarter / better
+
+    private static Object parse(ObjectNode rootNode, String fieldName, DataType dataType) {
+        if (dataType instanceof LongType) return rootNode.get(fieldName).longValue();
+
+        throw new UnsupportedOperationException(
+            String.format("Unsupported DataType %s", dataType.typeName())
+        );
+    }
+
+    private final Map<Integer, Object> ordinalToValueMap;
+    private final StructType readSchema;
+
+    public JsonRow(ObjectNode rootNode, StructType readSchema) {
+        this.readSchema = readSchema;
+        this.ordinalToValueMap = new HashMap<>();
+
+        for (int i = 0; i < readSchema.length(); i++) {
+            final StructField field = readSchema.at(i);
+            Object val = parse(rootNode, field.name, field.dataType);
+            ordinalToValueMap.put(i, val);
+        }
+    }
+
+    @Override
+    public long getLong(int ordinal) {
+        assert (readSchema.at(ordinal).dataType instanceof LongType);
+        return (long) ordinalToValueMap.get(ordinal);
+    }
+}

kernel-default/src/main/java/io/delta/core/helpers/DefaultTableHelper.java

@@ -0,0 +1,139 @@
+package io.delta.core.helpers;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Iterator;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.delta.core.data.JsonRow;
+import org.apache.hadoop.conf.Configuration;
+
+import io.delta.core.data.Row;
+import io.delta.core.expressions.Expression;
+import io.delta.core.fs.FileStatus;
+import io.delta.core.types.StructType;
+import io.delta.core.utils.CloseableIterator;
+import io.delta.storage.LocalLogStore;
+import io.delta.storage.LogStore;
+import org.apache.hadoop.fs.Path;
+
+public class DefaultTableHelper implements TableHelper {
+
+    private final Configuration hadoopConf;
+    private final LogStore logStore;
+    private final ObjectMapper objectMapper;
+
+    public DefaultTableHelper() {
+        this.hadoopConf = new Configuration();
+        this.logStore = new LocalLogStore(hadoopConf);
+        this.objectMapper = new ObjectMapper();
+    }
+
+    @Override
+    public CloseableIterator<FileStatus> listFiles(String path) {
+        return new CloseableIterator<FileStatus>() {
+            private final Iterator<org.apache.hadoop.fs.FileStatus> iter;
+
+            {
+                try {
+                    iter = logStore.listFrom(new Path(path), hadoopConf);
+                } catch (IOException ex) {
+                    throw new RuntimeException("Could not resolve the FileSystem", ex);
+                }
+            }
+
+            @Override
+            public boolean hasNext() {
+                return iter.hasNext();
+            }
+
+            @Override
+            public FileStatus next() {
+                return new FileStatus() {
+                    final org.apache.hadoop.fs.FileStatus impl = iter.next();
+
+                    @Override
+                    public String pathStr() {
+                        return impl.getPath().toString();
+                    }
+
+                    @Override
+                    public long length() {
+                        return impl.getLen();
+                    }
+
+                    @Override
+                    public long modificationTime() {
+                        return impl.getModificationTime();
+                    }
+                };
+            }
+
+            @Override
+            public void close() throws IOException { }
+        };
+    }
+
+    @Override
+    public CloseableIterator<Row> readJsonFile(String path, StructType readSchema) throws FileNotFoundException {
+        return new CloseableIterator<Row>() {
+            private final io.delta.storage.CloseableIterator<String> iter;
+
+            {
+                try {
+                    iter = logStore.read(new Path(path), hadoopConf);
+                } catch (IOException ex) {
+                    if (ex instanceof FileNotFoundException) {
+                        throw (FileNotFoundException) ex;
+                    }
+
+                    throw new RuntimeException("Could not resolve the FileSystem", ex);
+                }
+            }
+
+            @Override
+            public void close() throws IOException {
+                iter.close();
+            }
+
+            @Override
+            public boolean hasNext() {
+                return iter.hasNext();
+            }
+
+            @Override
+            public Row next() {
+                final String json = iter.next();
+                try {
+                    final JsonNode jsonNode = objectMapper.readTree(json);
+                    return new JsonRow((ObjectNode) jsonNode, readSchema);
+                } catch (JsonProcessingException ex) {
+                    throw new RuntimeException(String.format("Could not parse JSON: %s", json), ex);
+                }
+            }
+        };
+    }
+
+    @Override
+    public CloseableIterator<Row> readParquetFile(String path, StructType readSchema) {
+        return null;
+    }
+
+    @Override
+    public CloseableIterator<Row> readParquetFile(String path, StructType readSchema, Expression skippingFilter) {
+        return null;
+    }
+
+    @Override
+    public Row parseStats(String statsJson) {
+        return null;
+    }
+
+    @Override
+    public ScanHelper getScanHelper() {
+        return null;
+    }
+}

kernel-default/src/test/resources/basic-no-checkpoint/_delta_log/00000000000000000000.json

@@ -0,0 +1,5 @@
+{"commitInfo":{"timestamp":1679943456996,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"2","numOutputRows":"10","numOutputBytes":"1003"},"engineInfo":"Apache-Spark/3.3.2 Delta-Lake/2.2.0","txnId":"ff312e01-4714-45f0-868a-3fb36e51ca7d"}}
+{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}
+{"metaData":{"id":"testId","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1679943453303}}
+{"add":{"path":"part-00000-b490279f-72b0-4690-aa25-6a82eb94a985-c000.snappy.parquet","partitionValues":{},"size":500,"modificationTime":1679943456000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":0},\"maxValues\":{\"id\":4},\"nullCount\":{\"id\":0}}"}}
+{"add":{"path":"part-00001-da6606e8-9cda-457f-a155-46fa5da29690-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943456000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":5},\"maxValues\":{\"id\":9},\"nullCount\":{\"id\":0}}"}}

kernel-default/src/test/resources/basic-no-checkpoint/_delta_log/00000000000000000001.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1679943462467,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"readVersion":0,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"2","numOutputRows":"10","numOutputBytes":"1006"},"engineInfo":"Apache-Spark/3.3.2 Delta-Lake/2.2.0","txnId":"0f9a79b7-4882-477d-914c-bce60f36c12d"}}
+{"add":{"path":"part-00000-fd1e7a9b-aa91-4982-b013-e2ad17e4cc32-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943462000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":10},\"maxValues\":{\"id\":14},\"nullCount\":{\"id\":0}}"}}
+{"add":{"path":"part-00001-70abc612-9190-4a6e-9a55-1583bc1ca391-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943462000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":15},\"maxValues\":{\"id\":19},\"nullCount\":{\"id\":0}}"}}

kernel-default/src/test/resources/basic-no-checkpoint/_delta_log/00000000000000000002.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1679943463952,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"readVersion":1,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"2","numOutputRows":"10","numOutputBytes":"1006"},"engineInfo":"Apache-Spark/3.3.2 Delta-Lake/2.2.0","txnId":"bce9b0ef-dba2-48e7-a379-a76f37d67c9c"}}
+{"add":{"path":"part-00000-6570604d-1129-4e82-92a0-d9593195b971-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943463000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":20},\"maxValues\":{\"id\":24},\"nullCount\":{\"id\":0}}"}}
+{"add":{"path":"part-00001-91825ecf-7ff9-410d-89d4-aeaba097c11b-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943463000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":25},\"maxValues\":{\"id\":29},\"nullCount\":{\"id\":0}}"}}

kernel-default/src/test/resources/basic-no-checkpoint/_delta_log/00000000000000000003.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1679943465380,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"readVersion":2,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"2","numOutputRows":"10","numOutputBytes":"1006"},"engineInfo":"Apache-Spark/3.3.2 Delta-Lake/2.2.0","txnId":"8286031e-4dc6-45e4-b3da-72417fe33bb0"}}
+{"add":{"path":"part-00000-20c6bed6-b3e2-4a4e-8239-bc4c989e71f2-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943465000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":30},\"maxValues\":{\"id\":34},\"nullCount\":{\"id\":0}}"}}
+{"add":{"path":"part-00001-52ce751a-5e35-46c2-8bfe-41b88781814e-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943465000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":35},\"maxValues\":{\"id\":39},\"nullCount\":{\"id\":0}}"}}

kernel-default/src/test/resources/basic-no-checkpoint/_delta_log/00000000000000000004.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1679943466719,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"readVersion":3,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"2","numOutputRows":"10","numOutputBytes":"1006"},"engineInfo":"Apache-Spark/3.3.2 Delta-Lake/2.2.0","txnId":"d49e242a-b0e4-4a2d-aebb-44482dfed808"}}
+{"add":{"path":"part-00000-3883de18-7996-4e08-809e-6dbe967d580f-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943466000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":40},\"maxValues\":{\"id\":44},\"nullCount\":{\"id\":0}}"}}
+{"add":{"path":"part-00001-6efe57e5-3a3f-4430-a0f3-ce3d61130b9a-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943466000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":45},\"maxValues\":{\"id\":49},\"nullCount\":{\"id\":0}}"}}

kernel-default/src/test/resources/basic-no-checkpoint/_delta_log/00000000000000000005.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1679943467753,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"readVersion":4,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"2","numOutputRows":"10","numOutputBytes":"1006"},"engineInfo":"Apache-Spark/3.3.2 Delta-Lake/2.2.0","txnId":"8e241f39-f3f5-4bbf-81dd-5cc68ee1fa2c"}}
+{"add":{"path":"part-00000-7fbd89a3-73a8-4553-af2d-482507955b73-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943467000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":50},\"maxValues\":{\"id\":54},\"nullCount\":{\"id\":0}}"}}
+{"add":{"path":"part-00001-4f0d5e2d-a6ae-442f-9277-b8b670447b37-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943467000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":55},\"maxValues\":{\"id\":59},\"nullCount\":{\"id\":0}}"}}

kernel-default/src/test/resources/basic-no-checkpoint/_delta_log/00000000000000000006.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1679943468830,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"readVersion":5,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"2","numOutputRows":"10","numOutputBytes":"1006"},"engineInfo":"Apache-Spark/3.3.2 Delta-Lake/2.2.0","txnId":"84e116f1-8e66-42e1-848c-6367c666a2e3"}}
+{"add":{"path":"part-00000-8b1211c5-9d57-4fca-b723-a926205f38d2-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943468000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":60},\"maxValues\":{\"id\":64},\"nullCount\":{\"id\":0}}"}}
+{"add":{"path":"part-00001-e8342fee-6cef-452b-aaf7-2ab71f835e9d-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943468000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":65},\"maxValues\":{\"id\":69},\"nullCount\":{\"id\":0}}"}}

kernel-default/src/test/resources/basic-no-checkpoint/_delta_log/00000000000000000007.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1679943469859,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"readVersion":6,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"2","numOutputRows":"10","numOutputBytes":"1006"},"engineInfo":"Apache-Spark/3.3.2 Delta-Lake/2.2.0","txnId":"13a62249-9801-486e-8723-d86add00c182"}}
+{"add":{"path":"part-00000-ee495f65-0a7b-4cd9-9072-e07a2f0206a1-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943469000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":70},\"maxValues\":{\"id\":74},\"nullCount\":{\"id\":0}}"}}
+{"add":{"path":"part-00001-7da76162-c51d-41d8-822f-1ecdbbd5bf52-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943469000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":75},\"maxValues\":{\"id\":79},\"nullCount\":{\"id\":0}}"}}

kernel-default/src/test/resources/basic-no-checkpoint/_delta_log/00000000000000000008.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1679943470857,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"readVersion":7,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"2","numOutputRows":"10","numOutputBytes":"1006"},"engineInfo":"Apache-Spark/3.3.2 Delta-Lake/2.2.0","txnId":"b201d339-b75a-43ec-8748-9575608cab8a"}}
+{"add":{"path":"part-00000-e6d1d2e6-7511-48a7-9a8b-b22c1ecfef32-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943470000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":80},\"maxValues\":{\"id\":84},\"nullCount\":{\"id\":0}}"}}
+{"add":{"path":"part-00001-1876aac0-c5fe-450c-b843-e44277d5f1a3-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943470000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":85},\"maxValues\":{\"id\":89},\"nullCount\":{\"id\":0}}"}}

kernel-default/src/test/resources/basic-with-checkpoint/_delta_log/00000000000000000000.json

@@ -0,0 +1,5 @@
+{"commitInfo":{"timestamp":1679943471996,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"2","numOutputRows":"10","numOutputBytes":"1003"},"engineInfo":"Apache-Spark/3.3.2 Delta-Lake/2.2.0","txnId":"fa272a31-18c1-4c57-ae5c-6b52fbe83e92"}}
+{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}
+{"metaData":{"id":"testId","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1679943471575}}
+{"add":{"path":"part-00000-a65ab59f-72fd-44c9-a73e-e2d09459f836-c000.snappy.parquet","partitionValues":{},"size":500,"modificationTime":1679943471000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":0},\"maxValues\":{\"id\":4},\"nullCount\":{\"id\":0}}"}}
+{"add":{"path":"part-00001-39e6196f-2259-4ba4-b1d6-005712cd7784-c000.snappy.parquet","partitionValues":{},"size":503,"modificationTime":1679943471000,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":5},\"maxValues\":{\"id\":9},\"nullCount\":{\"id\":0}}"}}