fix reads

Author: richardc-db

kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/vector/DefaultVariantVector.java

@@ -49,7 +49,6 @@ public DefaultVariantVector(
         ColumnVector value,
         ColumnVector metadata) {
         super(size, type, nullability);
-        // checkArgument(offsets.length >= size + 1, "invalid offset array size");
         this.valueVector = requireNonNull(value, "value is null");
         this.metadataVector = requireNonNull(metadata, "metadata is null");
     }

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/internal/parquet/ParquetFileReaderSuite.scala

@@ -16,6 +16,9 @@
 package io.delta.kernel.defaults.internal.parquet
 
 import java.math.BigDecimal
+
+import org.apache.spark.sql.DataFrame
+
 import io.delta.golden.GoldenTableUtils.goldenTableFile
 import io.delta.kernel.defaults.utils.{ExpressionTestUtils, TestRow, VectorTestUtils}
 import io.delta.kernel.types._
@@ -139,4 +142,55 @@ class ParquetFileReaderSuite extends AnyFunSuite
 
     checkAnswer(actResult2, expResult2)
   }
+
+  private def testReadVariant(testName: String)(df: => DataFrame): Unit = {
+    test(testName) {
+      withTable("test_variant_table") {
+        df.write
+          .format("delta")
+          .mode("overwrite")
+          .saveAsTable("test_variant_table")
+        val path = spark.sql("describe table extended `test_variant_table`")
+          .where("col_name = 'Location'")
+          .collect()(0)
+          .getString(1)
+          .replace("file:", "")
+
+        val kernelSchema = tableSchema(path)
+        val actResult = readParquetFilesUsingKernel(path, kernelSchema)
+        val expResult = readParquetFilesUsingSpark(path, kernelSchema)
+        checkAnswer(actResult, expResult)
+      }
+    }
+  }
+
+  testReadVariant("basic read variant") {
+    spark.range(0, 10, 1, 1).selectExpr(
+      "parse_json(cast(id as string)) as basic_v",
+      "named_struct('v', parse_json(cast(id as string))) as struct_v",
+      """array(
+        parse_json(cast(id as string)),
+        parse_json(cast(id as string)),
+        parse_json(cast(id as string))
+      ) as array_v""",
+      "map('test', parse_json(cast(id as string))) as map_value_v",
+      "map(parse_json(cast(id as string)), parse_json(cast(id as string))) as map_key_v"
+    )
+  }
+
+  testReadVariant("basic null variant") {
+    spark.range(0, 10, 1, 1).selectExpr(
+      "cast(null as variant) basic_v",
+      "named_struct('v', cast(null as variant)) as struct_v",
+      """array(
+        parse_json(cast(id as string)),
+        parse_json(cast(id as string)),
+        null
+      ) as array_v""",
+      "map('test', cast(null as variant)) as map_value_v",
+      "map(cast(null as variant), parse_json(cast(id as string))) as map_key_v",
+    )
+  }
+
+  // TODO(richardc-db): Add nested variant tests once `parse_json` expression is implemented.
 }

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/utils/TestRow.scala

@@ -18,6 +18,7 @@ package io.delta.kernel.defaults.utils
 import scala.collection.JavaConverters._
 import org.apache.spark.sql.{types => sparktypes}
 import org.apache.spark.sql.{Row => SparkRow}
+import org.apache.spark.unsafe.types.VariantVal
 import io.delta.kernel.data.{ArrayValue, ColumnVector, MapValue, Row}
 import io.delta.kernel.types._
 
@@ -44,7 +45,7 @@ import java.time.{Instant, LocalDate, LocalDateTime, ZoneOffset}
  * - ArrayType --> Seq[Any]
  * - MapType --> Map[Any, Any]
  * - StructType --> TestRow
- *
+ * - VariantType --> VariantVal
  * For complex types array and map, the inner elements types should align with this mapping.
  */
 class TestRow(val values: Array[Any]) {
@@ -108,7 +109,9 @@ object TestRow {
         case _: ArrayType => arrayValueToScalaSeq(row.getArray(i))
         case _: MapType => mapValueToScalaMap(row.getMap(i))
         case _: StructType => TestRow(row.getStruct(i))
-        case _: VariantType => row.getVariant(i)
+        case _: VariantType =>
+          val kernelVariant = row.getVariant(i)
+          new VariantVal(kernelVariant.getValue(), kernelVariant.getMetadata())
         case _ => throw new UnsupportedOperationException("unrecognized data type")
       }
     }.toSeq)
@@ -141,6 +144,7 @@ object TestRow {
             decodeCellValue(mapType.keyType, k) -> decodeCellValue(mapType.valueType, v)
         }
         case _: sparktypes.StructType => TestRow(obj.asInstanceOf[SparkRow])
+        case _: sparktypes.VariantType => obj.asInstanceOf[VariantVal]
         case _ => throw new UnsupportedOperationException("unrecognized data type")
       }
     }
@@ -174,7 +178,7 @@ object TestRow {
             decodeCellValue(mapType.keyType, k) -> decodeCellValue(mapType.valueType, v)
           }
         case _: sparktypes.StructType => TestRow(row.getStruct(i))
-        case _: sparktypes.VariantType => row.getAs[Row](i)
+        case _: sparktypes.VariantType => row.getAs[VariantVal](i)
         case _ => throw new UnsupportedOperationException("unrecognized data type")
       }
     })
@@ -206,6 +210,9 @@ object TestRow {
         TestRow.fromSeq(Seq.range(0, dataType.length()).map { ordinal =>
           getAsTestObject(vector.getChild(ordinal), rowId)
         })
+      case _: VariantType =>
+        val kernelVariant = vector.getVariant(rowId)
+        new VariantVal(kernelVariant.getValue(), kernelVariant.getMetadata())
       case _ => throw new UnsupportedOperationException("unrecognized data type")
     }
   }

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/utils/TestUtils.scala

@@ -41,6 +41,7 @@ import org.apache.hadoop.shaded.org.apache.commons.io.FileUtils
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.{types => sparktypes}
 import org.apache.spark.sql.catalyst.plans.SQLHelper
+import org.apache.spark.unsafe.types.VariantVal
 import org.scalatest.Assertions
 
 trait TestUtils extends Assertions with SQLHelper {
@@ -117,6 +118,17 @@ trait TestUtils extends Assertions with SQLHelper {
     lazy val classLoader: ClassLoader = ResourceLoader.getClass.getClassLoader
   }
 
+  /**
+   * Drops table `tableName` after calling `f`.
+   */
+  def withTable(tableNames: String*)(f: => Unit): Unit = {
+    try f finally {
+      tableNames.foreach { name =>
+        spark.sql(s"DROP TABLE IF EXISTS $name")
+      }
+    }
+  }
+
   def withGoldenTable(tableName: String)(testFunc: String => Unit): Unit = {
     val tablePath = GoldenTableUtils.goldenTablePath(tableName)
     testFunc(tablePath)
@@ -404,6 +416,7 @@ trait TestUtils extends Assertions with SQLHelper {
         java.lang.Double.doubleToRawLongBits(a) == java.lang.Double.doubleToRawLongBits(b)
       case (a: Float, b: Float) =>
         java.lang.Float.floatToRawIntBits(a) == java.lang.Float.floatToRawIntBits(b)
+      case (a: VariantVal, b: VariantVal) => a.debugString() == b.debugString()
       case (a, b) =>
         if (!a.equals(b)) {
           val sds = 200;