Rename to DecodeNestedZ85EncodedVariantSuite

Author: harshmotw-db

…ns/ReplaceVariantZ85WithVariantVal.scala …DecodeNestedZ85EncodedVariantSuite.scalaspark/src/main/scala/org/apache/spark/sql/delta/expressions/ReplaceVariantZ85WithVariantVal.scala renamed to spark/src/main/scala/org/apache/spark/sql/delta/expressions/DecodeNestedZ85EncodedVariantSuite.scala spark/src/main/scala/org/apache/spark/sql/delta/expressions/ReplaceVariantZ85WithVariantVal.scala renamed to spark/src/main/scala/org/apache/spark/sql/delta/expressions/DecodeNestedZ85EncodedVariantSuite.scala

@@ -35,17 +35,21 @@ import org.apache.spark.unsafe.types.VariantVal
  *
  * @param child The expression producing the row with Z85-encoded variants
  */
-case class ReplaceVariantZ85WithVariantVal(child: Expression)
+case class DecodeNestedZ85EncodedVariant(child: Expression)
   extends UnaryExpression with CodegenFallback {
 
   override def dataType: DataType = child.dataType
 
   override def nullable: Boolean = child.nullable
 
   override def checkInputDataTypes(): TypeCheckResult = {
-    if (!isValidType(child.dataType)) {
+    if (!child.dataType.isInstanceOf[StructType]) {
+      TypeCheckResult.TypeCheckFailure(s"The top-level data type for the input to " +
+        s"DecodeNestedZ85EncodedVariant must be StructType but this is not true " +
+        s"in: ${child.dataType}.")
+    } else if (!isValidType(child.dataType)) {
       TypeCheckResult.TypeCheckFailure(
-        s"ReplaceVariantZ85WithVariantVal does not support arrays or maps in schema. " +
+        s"DecodeNestedZ85EncodedVariant does not support arrays or maps in schema. " +
         s"Found: ${child.dataType}")
     } else {
       TypeCheckResult.TypeCheckSuccess
@@ -100,5 +104,5 @@ case class ReplaceVariantZ85WithVariantVal(child: Expression)
   override def prettyName: String = "replace_variant_z85_with_variant_val"
 
   override protected def withNewChildInternal(newChild: Expression)
-    : ReplaceVariantZ85WithVariantVal = copy(child = newChild)
+      : DecodeNestedZ85EncodedVariant = copy(child = newChild)
 }

spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala

@@ -25,7 +25,7 @@ import org.apache.spark.sql.delta.skipping.clustering.{ClusteredTableUtils, Clus
 import org.apache.spark.sql.delta.{DeltaColumnMapping, DeltaLog, DeltaTableUtils}
 import org.apache.spark.sql.delta.ClassicColumnConversions._
 import org.apache.spark.sql.delta.actions.{AddFile, Metadata}
-import org.apache.spark.sql.delta.expressions.ReplaceVariantZ85WithVariantVal
+import org.apache.spark.sql.delta.expressions.DecodeNestedZ85EncodedVariant
 import org.apache.spark.sql.delta.implicits._
 import org.apache.spark.sql.delta.metering.DeltaLogging
 import org.apache.spark.sql.delta.schema.SchemaUtils
@@ -268,7 +268,7 @@ trait DataSkippingReaderBase
   /** Returns a DataFrame expression to obtain a list of files with parsed statistics. */
   private def withStatsInternal0: DataFrame = {
     val parsedStats = from_json(col("stats"), statsSchema)
-    val decodedStats = Column(ReplaceVariantZ85WithVariantVal(parsedStats.expr))
+    val decodedStats = Column(DecodeNestedZ85EncodedVariant(parsedStats.expr))
     allFiles.withColumn("stats", decodedStats)
   }
 

…placeVariantZ85WithVariantValSuite.scala …DecodeNestedZ85EncodedVariantSuite.scalaspark/src/test/scala/org/apache/spark/sql/delta/expressions/ReplaceVariantZ85WithVariantValSuite.scala renamed to spark/src/test/scala/org/apache/spark/sql/delta/expressions/DecodeNestedZ85EncodedVariantSuite.scala spark/src/test/scala/org/apache/spark/sql/delta/expressions/ReplaceVariantZ85WithVariantValSuite.scala renamed to spark/src/test/scala/org/apache/spark/sql/delta/expressions/DecodeNestedZ85EncodedVariantSuite.scala

@@ -28,7 +28,7 @@ import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructFiel
 import org.apache.spark.types.variant.Variant
 import org.apache.spark.unsafe.types.{UTF8String, VariantVal}
 
-class ReplaceVariantZ85WithVariantValSuite extends QueryTest with DeltaSQLCommandTest {
+class DecodeNestedZ85EncodedVariantSuite extends QueryTest with DeltaSQLCommandTest {
 
   test("RoundTrip alternateVariantEncoding Z85") {
     val jsonValues = Seq(
@@ -53,10 +53,10 @@ class ReplaceVariantZ85WithVariantValSuite extends QueryTest with DeltaSQLComman
       val statsSchema = StructType(Seq(StructField("v", VariantType)))
       val parsedDf = df.withColumn("parsed", from_json(col("z85_string"), statsSchema))
 
-      // Apply ReplaceVariantZ85WithVariantVal
+      // Apply DecodeNestedZ85EncodedVariant
       val decodedDf = parsedDf.withColumn(
         "decoded",
-        Column(ReplaceVariantZ85WithVariantVal(col("parsed").expr))
+        Column(DecodeNestedZ85EncodedVariant(col("parsed").expr))
       )
 
       // Extract the decoded variant and verify
@@ -109,7 +109,7 @@ class ReplaceVariantZ85WithVariantValSuite extends QueryTest with DeltaSQLComman
     val parsedDf = df.withColumn("parsed", from_json(col("stats"), statsSchema))
     val decodedDf = parsedDf.withColumn(
       "decoded",
-      Column(ReplaceVariantZ85WithVariantVal(col("parsed").expr))
+      Column(DecodeNestedZ85EncodedVariant(col("parsed").expr))
     )
 
     val result = decodedDf.select(