Fix nested char padding

Author: eshed-flarion

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala

@@ -303,12 +303,19 @@ object CharVarcharUtils extends Logging with SparkCharVarcharUtils {
           val fieldExpr = GetStructField(expr, i, Some(field.name))
           val padded = padCharToTargetLength(
             fieldExpr, field.dataType, targets(i).dataType, alwaysPad)
-          needPadding = padded.isDefined
+          needPadding |= padded.isDefined
           createStructExprs += Literal(field.name)
           createStructExprs += padded.getOrElse(fieldExpr)
           i += 1
         }
-        if (needPadding) Some(CreateNamedStruct(createStructExprs.toSeq)) else None
+        val struct = if (needPadding) Some(CreateNamedStruct(createStructExprs.toSeq)) else None
+        struct.map { padded =>
+          if (expr.nullable) {
+            If(IsNull(expr), Literal(null, padded.dataType), padded)
+          } else {
+            padded
+          }
+        }
 
       case (ArrayType(et, containsNull), ArrayType(target, _)) =>
         val param = NamedLambdaVariable("x", replaceCharVarcharWithString(et), containsNull)

sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala

@@ -541,6 +541,18 @@ trait CharVarcharTestSuite extends QueryTest with SQLTestUtils {
     }
   }
 
+  test("char type comparison: nested in struct with multiple fields") {
+    withTable("t") {
+      sql(s"CREATE TABLE t(c1 STRUCT<c: CHAR(2), d: CHAR(3)>, c2 STRUCT<c: CHAR(5), d: CHAR(3)>) " +
+        s"USING $format")
+      sql("INSERT INTO t VALUES (struct('a', 'b'), struct('a', 'b'))")
+      testConditions(spark.table("t"), Seq(
+        ("c1 = c2", true),
+        ("c1 < c2", false),
+        ("c1 IN (c2)", true)))
+    }
+  }
+
   test("char type comparison: nested in array") {
     withTable("t") {
       sql(s"CREATE TABLE t(c1 ARRAY<CHAR(2)>, c2 ARRAY<CHAR(5)>) USING $format")
@@ -576,6 +588,15 @@ trait CharVarcharTestSuite extends QueryTest with SQLTestUtils {
     }
   }
 
+  test("char type comparison: nested in array of struct with nulls") {
+    withTable("t") {
+      sql("CREATE TABLE t(c1 ARRAY<STRUCT<c: CHAR(2)>>, c2 ARRAY<STRUCT<c: CHAR(5)>>) " +
+          s"USING $format")
+      sql("INSERT INTO t VALUES (array(NULL), array(struct(NULL)))")
+      testConditions(spark.table("t"), Seq(("c1 = c2", false)))
+    }
+  }
+
   test("char type comparison: nested in array of array") {
     withTable("t") {
       sql("CREATE TABLE t(c1 ARRAY<ARRAY<CHAR(2)>>, c2 ARRAY<ARRAY<CHAR(5)>>) " +