[WIP] Delta Core API

build.sbt

@@ -796,3 +796,20 @@ lazy val flink = (project in file("flink"))
     // Ensure unidoc is run with tests. Must be cleaned before test for unidoc to be generated.
     (Test / test) := ((Test / test) dependsOn (Compile / unidoc)).value
   )
+
+lazy val core = (project in file("core"))
+  .settings(
+    name := "delta-core",
+    commonSettings,
+    skipReleaseSettings,
+    libraryDependencies ++= Seq()
+  )
+
+lazy val defaultCore = (project in file("default-core"))
+  .dependsOn(core)
+  .settings(
+    name := "delta-core-default",
+    commonSettings,
+    skipReleaseSettings,
+    libraryDependencies ++= Seq()
+  )

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

@@ -0,0 +1,18 @@
+package io.delta.core;
+
+public enum ColumnMappingMode {
+    NONE("none"),
+    ID_MAPPING("id"),
+    NAME_MAPPING("name");
+
+    private final String name;
+
+    ColumnMappingMode(String name) {
+        this.name = name;
+    }
+
+    @Override
+    public String toString() {
+        return name;
+    }
+}

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

@@ -0,0 +1,8 @@
+package io.delta.core;
+
+import io.delta.core.utils.CloseableIterator;
+
+public interface Scan {
+
+    CloseableIterator<ScanTask> getTasks();
+}

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

@@ -0,0 +1,15 @@
+package io.delta.core;
+
+import io.delta.core.expressions.Expression;
+import io.delta.core.types.StructType;
+
+public interface ScanBuilder {
+
+    /** For column pruning. */
+    ScanBuilder withReadSchema(StructType schema);
+
+    /** For partition pruning and filter push down. */
+    ScanBuilder withFilter(Expression filter);
+
+    Scan build();
+}

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

@@ -0,0 +1,9 @@
+package io.delta.core;
+
+import io.delta.core.data.ColumnarBatch;
+import io.delta.core.utils.CloseableIterator;
+
+public interface ScanTask {
+
+    CloseableIterator<ColumnarBatch> getData();
+}

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

@@ -0,0 +1,10 @@
+package io.delta.core;
+
+import io.delta.core.types.StructType;
+
+public interface Snapshot {
+
+    StructType getSchema();
+
+    ScanBuilder getScanBuilder();
+}

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

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

core/src/main/java/io/delta/core/data/ColumnVector.java

@@ -0,0 +1,4 @@
+package io.delta.core.data;
+
+public interface ColumnVector extends AutoCloseable {
+}

core/src/main/java/io/delta/core/data/ColumnarBatch.java

@@ -0,0 +1,15 @@
+package io.delta.core.data;
+
+import io.delta.core.types.DataType;
+import io.delta.core.types.StructType;
+
+public interface ColumnarBatch {
+
+    StructType getSchema();
+
+    ColumnVector getColumnVector(String columnName);
+
+    void addConstantColumn(String topLevelColumnName, DataType dataType, Row value);
+
+    void rename(String currentColumnName, String newColumnName);
+}

core/src/main/java/io/delta/core/data/DeletionVector.java

@@ -0,0 +1,5 @@
+package io.delta.core.data;
+
+public interface DeletionVector extends ColumnVector {
+
+}

core/src/main/java/io/delta/core/data/Row.java

@@ -0,0 +1,42 @@
+package io.delta.core.data;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.List;
+import java.util.Map;
+
+public interface Row {
+
+    boolean isNullAt(int ordinal);
+
+    boolean getBoolean(int ordinal);
+
+    byte getByte(int ordinal);
+
+    short getShort(int ordinal);
+
+    int getInt(int ordinal);
+
+    long getLong(int ordinal);
+
+    float getFloat(int ordinal);
+
+    double getDouble(int ordinal);
+
+    BigDecimal getDecimal(int ordinal, int precision, int scale);
+
+    String getString(int ordinal);
+
+    byte[] getBinary(int ordinal);
+
+    Timestamp getTimestamp(int ordinal);
+
+    Date getDate(int ordinal);
+
+    Row getRecord(int ordinal);
+
+    <T> List<T> getList(int ordinal);
+
+    <K, V> Map<K, V> getMap(int ordinal);
+}

core/src/main/java/io/delta/core/expressions/Expression.java

@@ -0,0 +1,4 @@
+package io.delta.core.expressions;
+
+public interface Expression {
+}

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

@@ -0,0 +1,10 @@
+package io.delta.core.fs;
+
+public interface FileStatus {
+
+    String filePath();
+
+    long size();
+
+    long modificationTime();
+}

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

@@ -0,0 +1,26 @@
+package io.delta.core.helpers;
+
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.TimeZone;
+
+import io.delta.core.ColumnMappingMode;
+import io.delta.core.data.ColumnarBatch;
+import io.delta.core.expressions.Expression;
+import io.delta.core.fs.FileStatus;
+import io.delta.core.types.StructType;
+import io.delta.core.utils.CloseableIterator;
+
+public interface ScanHelper extends Serializable {
+
+    // Note: for production, will likely be a builder pattern to deal with all these optional params
+
+    CloseableIterator<ColumnarBatch> readParquetDataFile(
+        FileStatus fileStatus,
+        StructType physicalReadSchema,
+        TimeZone timeZone,
+        Optional<ColumnMappingMode> columnMappingMode,
+        Optional<Expression> skippingFilter,
+        Optional<FileStatus> deletionVectorFileStatus
+    );
+}

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

@@ -0,0 +1,9 @@
+package io.delta.core.helpers;
+
+import java.io.DataInputStream;
+
+/** Should only be extended by ScanHelper. */
+public interface SupportsDeletionVector {
+
+    DataInputStream readDeletionVector(String path);
+}

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

@@ -0,0 +1,29 @@
+package io.delta.core.helpers;
+
+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;
+
+public interface TableHelper {
+
+    CloseableIterator<FileStatus> listFiles(String path);
+
+    CloseableIterator<FileStatus> listFiles(String path, String prefixToListFrom);
+
+    CloseableIterator<Row> readJsonFile(String path, StructType readSchema);
+
+    /** Uses the readSchema for partition pruning. */
+    CloseableIterator<Row> readParquetFile(String path, StructType readSchema);
+
+    /** Uses the readSchema for partition pruning and the skippingFilter for data filtering. */
+    CloseableIterator<Row> readParquetFile(
+        String path,
+        StructType readSchema,
+        Expression skippingFilter);
+
+    Row parseStats(String statsJson);
+
+    ScanHelper getScanHelper();
+}

core/src/main/java/io/delta/core/types/DataType.java

@@ -0,0 +1,4 @@
+package io.delta.core.types;
+
+public abstract class DataType { }
+

core/src/main/java/io/delta/core/types/StructType.java

@@ -0,0 +1,3 @@
+package io.delta.core.types;
+
+public final class StructType extends DataType { }

core/src/main/java/io/delta/core/utils/CloseableIterator.java

@@ -0,0 +1,6 @@
+package io.delta.core.utils;
+
+import java.io.Closeable;
+import java.util.Iterator;
+
+public interface CloseableIterator<T> extends Iterator<T>, Closeable { }

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

@@ -0,0 +1,4 @@
+package io.delta.core.helpers;
+
+public class DefaultTableHelper implements TableHelper {
+}