[Kernel] Add missing data types and support for data type (de)seriali…

…zation

This PR is part of #1783. It adds additional data types supported by Delta Lake protocol that were missing from the interfaces PR #1808.

It also adds serialization and deserailization of table schema represented as `StructType`.

UTs

Closes #1842

kernel/build.sbt

@@ -46,7 +46,13 @@ lazy val kernelApi = (project in file("kernel-api"))
     commonSettings,
     scalaStyleSettings,
     releaseSettings,
-    libraryDependencies ++= Seq(),
+    libraryDependencies ++= Seq(
+
+      "com.fasterxml.jackson.core" % "jackson-databind" % "2.13.5" % "test",
+      "org.scalatest" %% "scalatest" % scalaTestVersion % "test",
+      "junit" % "junit" % "4.11" % "test",
+      "com.novocode" % "junit-interface" % "0.11" % "test"
+    ),
     Compile / doc / javacOptions := Seq(
       "-public",
       "-windowtitle", "Delta Kernel API " + version.value.replaceAll("-SNAPSHOT", "") + " JavaDoc",

kernel/kernel-api/src/main/java/io/delta/kernel/data/Row.java

@@ -65,13 +65,13 @@ public interface Row {
      * Return struct value of the column located at the given ordinal.
      * Throws error if the column at given ordinal is not of struct type,
      */
-    Row getRecord(int ordinal);
+    Row getStruct(int ordinal);
 
     /**
      * Return array value of the column located at the given ordinal.
      * Throws error if the column at given ordinal is not of array type,
      */
-    <T> List<T> getList(int ordinal);
+    <T> List<T> getArray(int ordinal);
 
     /**
      * Return map value of the column located at the given ordinal.

kernel/kernel-api/src/main/java/io/delta/kernel/expressions/And.java

@@ -48,8 +48,8 @@ public And(Expression left, Expression right) {
             throw new IllegalArgumentException(
                 String.format(
                     "'And' requires expressions of type boolean. Got %s and %s.",
-                    left.dataType().typeName(),
-                    right.dataType().typeName()
+                    left.dataType(),
+                    right.dataType()
                 )
             );
         }

kernel/kernel-api/src/main/java/io/delta/kernel/expressions/BinaryComparison.java

@@ -18,6 +18,8 @@
 
 import java.util.Comparator;
 
+import io.delta.kernel.internal.expressions.CastingComparator;
+
 /**
  * A {@link BinaryOperator} that compares the left and right {@link Expression}s and evaluates to a
  * boolean value.

kernel/kernel-api/src/main/java/io/delta/kernel/expressions/BinaryOperator.java

@@ -29,13 +29,13 @@ protected BinaryOperator(Expression left, Expression right, String symbol) {
         super(left, right);
         this.symbol = symbol;
 
-        if (!left.dataType().equivalent(right.dataType())) {
+        if (!left.dataType().equals(right.dataType())) {
             throw new IllegalArgumentException(
                 String.format(
                     "BinaryOperator left and right DataTypes must be the same. Found %s and %s.",
-                    left.dataType().typeName(),
-                    right.dataType().typeName())
-            );
+                    left.dataType(),
+                    right.dataType()
+            ));
         }
     }
 

kernel/kernel-api/src/main/java/io/delta/kernel/expressions/Column.java

@@ -21,7 +21,6 @@
 import java.util.Set;
 
 import io.delta.kernel.data.Row;
-import io.delta.kernel.types.*;
 import io.delta.kernel.types.BooleanType;
 import io.delta.kernel.types.DataType;
 import io.delta.kernel.types.IntegerType;
@@ -61,7 +60,7 @@ public Column(int ordinal, String name, DataType dataType) {
             throw new UnsupportedOperationException(
                 String.format(
                     "The data type %s of column %s at ordinal %s is not supported",
-                    dataType.typeName(),
+                    dataType,
                     name,
                     ordinal)
             );

kernel/kernel-api/src/main/java/io/delta/kernel/expressions/Literal.java

@@ -19,7 +19,6 @@
 import java.util.Objects;
 
 import io.delta.kernel.data.Row;
-import io.delta.kernel.types.*;
 import io.delta.kernel.types.BooleanType;
 import io.delta.kernel.types.DataType;
 import io.delta.kernel.types.IntegerType;

kernel/kernel-api/src/main/java/io/delta/kernel/expressions/CastingComparator.java → kernel/kernel-api/src/main/java/io/delta/kernel/internal/expressions/CastingComparator.java

@@ -14,40 +14,53 @@
  * limitations under the License.
  */
 
-package io.delta.kernel.expressions;
+package io.delta.kernel.internal.expressions;
 
 import java.util.Comparator;
 
-import io.delta.kernel.types.*;
+import io.delta.kernel.types.BinaryType;
 import io.delta.kernel.types.BooleanType;
+import io.delta.kernel.types.ByteType;
 import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.DateType;
+import io.delta.kernel.types.DoubleType;
 import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.FloatType;
 import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.ShortType;
 import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.TimestampType;
 
-// TODO: exclude from public interfaces (move to "internal" somewhere?)
 public class CastingComparator<T extends Comparable<T>> implements Comparator<Object> {
 
     public static Comparator<Object> forDataType(DataType dataType) {
-        if (dataType instanceof IntegerType) {
-            return new CastingComparator<Integer>();
-        }
-
         if (dataType instanceof BooleanType) {
             return new CastingComparator<Boolean>();
-        }
-
-        if (dataType instanceof LongType) {
+        } else if (dataType instanceof ByteType) {
+            return new CastingComparator<Byte>();
+        } else if (dataType instanceof ShortType) {
+            return new CastingComparator<Short>();
+        } else if (dataType instanceof IntegerType) {
+            return new CastingComparator<Integer>();
+        } else if (dataType instanceof LongType) {
             return new CastingComparator<Long>();
-        }
-
-        if (dataType instanceof StringType) {
+        } else if (dataType instanceof FloatType) {
+            return new CastingComparator<Float>();
+        } else if (dataType instanceof DoubleType) {
+            return new CastingComparator<Double>();
+        } else if (dataType instanceof StringType) {
             return new CastingComparator<String>();
+        } else if (dataType instanceof DateType) {
+            // Date value is accessed as integer (number of days since epoch).
+            // This may change in the future.
+            return new CastingComparator<Integer>();
+        } else if (dataType instanceof TimestampType) {
+            // Timestamp value is accessed as long (epoch seconds). This may change in the future.
+            return new CastingComparator<Long>();
         }
 
         throw new IllegalArgumentException(
-            String.format("Unsupported DataType: %s", dataType.typeName())
-        );
+            String.format("Unsupported DataType: %s", dataType));
     }
 
     private final Comparator<T> comparator;