Merge branch 'master' into SPARK-45959

Author: ashahid

core/src/test/scala/org/apache/spark/deploy/history/EventLogFileWritersSuite.scala

@@ -100,7 +100,7 @@ abstract class EventLogFileWritersSuite extends SparkFunSuite with LocalSparkCon
     }
   }
 
-  test("Use the defalut value of spark.eventLog.compression.codec") {
+  test("Use the default value of spark.eventLog.compression.codec") {
     val conf = new SparkConf
     conf.set(EVENT_LOG_COMPRESS, true)
     val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)

sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala

@@ -279,7 +279,7 @@ private object RowToColumnConverter {
 
     if (nullable) {
       dataType match {
-        case CalendarIntervalType => new StructNullableTypeConverter(core)
+        case CalendarIntervalType | VariantType => new StructNullableTypeConverter(core)
         case st: StructType => new StructNullableTypeConverter(core)
         case _ => new BasicNullableTypeConverter(core)
       }

sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchSuite.scala

@@ -45,7 +45,7 @@ import org.apache.spark.sql.util.ArrowUtils
 import org.apache.spark.sql.vectorized.{ArrowColumnVector, ColumnarBatch, ColumnarBatchRow, ColumnVector}
 import org.apache.spark.tags.ExtendedSQLTest
 import org.apache.spark.unsafe.Platform
-import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String}
+import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String, VariantVal}
 import org.apache.spark.util.ArrayImplicits._
 
 @ExtendedSQLTest
@@ -1650,6 +1650,7 @@ class ColumnarBatchSuite extends SparkFunSuite {
         StructField("int_to_int", MapType(IntegerType, IntegerType)) ::
         StructField("binary", BinaryType) ::
         StructField("ts_ntz", TimestampNTZType) ::
+        StructField("variant", VariantType) ::
         Nil)
     var mapBuilder = new ArrayBasedMapBuilder(IntegerType, IntegerType)
     mapBuilder.put(1, 10)
@@ -1664,6 +1665,9 @@ class ColumnarBatchSuite extends SparkFunSuite {
     val tsNTZ2 =
       DateTimeUtils.localDateTimeToMicros(LocalDateTime.parse(tsString2.replace(" ", "T")))
 
+    val variantVal1 = new VariantVal(Array[Byte](1, 2, 3), Array[Byte](4, 5))
+    val variantVal2 = new VariantVal(Array[Byte](6), Array[Byte](7, 8))
+
     val row1 = new GenericInternalRow(Array[Any](
       UTF8String.fromString("a string"),
       true,
@@ -1681,7 +1685,8 @@ class ColumnarBatchSuite extends SparkFunSuite {
       new GenericInternalRow(Array[Any](5.asInstanceOf[Any], 10)),
       mapBuilder.build(),
       "Spark SQL".getBytes(),
-      tsNTZ1
+      tsNTZ1,
+      variantVal1
     ))
 
     mapBuilder = new ArrayBasedMapBuilder(IntegerType, IntegerType)
@@ -1704,7 +1709,8 @@ class ColumnarBatchSuite extends SparkFunSuite {
       new GenericInternalRow(Array[Any](20.asInstanceOf[Any], null)),
       mapBuilder.build(),
       "Parquet".getBytes(),
-      tsNTZ2
+      tsNTZ2,
+      variantVal2
     ))
 
     val row3 = new GenericInternalRow(Array[Any](
@@ -1724,6 +1730,7 @@ class ColumnarBatchSuite extends SparkFunSuite {
       null,
       null,
       null,
+      null,
       null
     ))
 
@@ -1852,6 +1859,13 @@ class ColumnarBatchSuite extends SparkFunSuite {
       assert(columns(16).getLong(0) == tsNTZ1)
       assert(columns(16).getLong(1) == tsNTZ2)
       assert(columns(16).isNullAt(2))
+
+      assert(columns(17).dataType() == VariantType)
+      assert(columns(17).getVariant(0).debugString() == variantVal1.debugString())
+      assert(columns(17).getVariant(1).debugString() == variantVal2.debugString())
+      assert(columns(17).isNullAt(2))
+      assert(columns(17).getChild(0).isNullAt(2))
+      assert(columns(17).getChild(1).isNullAt(2))
     } finally {
       batch.close()
     }