Address comments.

Author: viirya

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveUnion.scala

@@ -44,18 +44,18 @@ object ResolveUnion extends Rule[LogicalPlan] {
     val resolver = SQLConf.get.resolver
     val missingFields =
       StructType.findMissingFields(col.dataType.asInstanceOf[StructType], target, resolver)
-    if (missingFields.length == 0) {
+    if (missingFields.isEmpty) {
       None
     } else {
-      Some(addFieldsInto(col, "", missingFields.fields))
+      missingFields.map(s => addFieldsInto(col, "", s.fields))
     }
   }
 
   /**
-   *  Adds missing fields recursively into given `col` expression. The missing fields are given
-   *  in `fields`. For example, given `col` as "a struct<a:int, b:int>, b int", and `fields` is
-   *  "a struct<c:long>, c string". This method will add a nested `a.c` field and a top-level
-   *  `c` field to `col` and fill null values for them.
+   * Adds missing fields recursively into given `col` expression. The missing fields are given
+   * in `fields`. For example, given `col` as "a struct<a:int, b:int>, b int", and `fields` is
+   * "a struct<c:long>, c string". This method will add a nested `a.c` field and a top-level
+   * `c` field to `col` and fill null values for them.
    */
   private def addFieldsInto(col: Expression, base: String, fields: Seq[StructField]): Expression = {
     fields.foldLeft(col) { case (currCol, field) =>
@@ -134,6 +134,7 @@ object ResolveUnion extends Rule[LogicalPlan] {
     // Builds a project list for `right` based on `left` output names
     val (rightProjectList, aliased) = compareAndAddFields(left, right, allowMissingCol)
 
+
     // Delegates failure checks to `CheckAnalysis`
     val notFoundAttrs = rightOutputAttrs.diff(rightProjectList ++ aliased)
     val rightChild = Project(rightProjectList ++ notFoundAttrs, right)

sql/catalyst/src/main/scala/org/apache/spark/sql/types/StructType.scala

@@ -646,7 +646,10 @@ object StructType extends AbstractDataType {
    * Returns a `StructType` that contains missing fields recursively from `source` to `target`.
    * Note that this doesn't support looking into array type and map type recursively.
    */
-  def findMissingFields(source: StructType, target: StructType, resolver: Resolver): StructType = {
+  def findMissingFields(
+      source: StructType,
+      target: StructType,
+      resolver: Resolver): Option[StructType] = {
     def bothStructType(dt1: DataType, dt2: DataType): Boolean =
       dt1.isInstanceOf[StructType] && dt2.isInstanceOf[StructType]
 
@@ -660,11 +663,17 @@ object StructType extends AbstractDataType {
       } else if (bothStructType(found.get.dataType, field.dataType) &&
           !found.get.dataType.sameType(field.dataType)) {
         // Found a field with same name, but different data type.
-        newFields += found.get.copy(dataType =
-          findMissingFields(found.get.dataType.asInstanceOf[StructType],
-            field.dataType.asInstanceOf[StructType], resolver))
+        findMissingFields(found.get.dataType.asInstanceOf[StructType],
+          field.dataType.asInstanceOf[StructType], resolver).map { missingType =>
+          newFields += found.get.copy(dataType = missingType)
+        }
       }
     }
-    StructType(newFields)
+
+    if (newFields.isEmpty) {
+      None
+    } else {
+      Some(StructType(newFields))
+    }
   }
 }

sql/catalyst/src/test/scala/org/apache/spark/sql/types/StructTypeSuite.scala

@@ -113,61 +113,71 @@ class StructTypeSuite extends SparkFunSuite {
     val source1 = StructType.fromDDL("c1 INT")
     val missing1 = StructType.fromDDL(
       "c2 STRUCT<c3: INT, c4: STRUCT<c5: INT, c6: INT>>")
-    assert(StructType.findMissingFields(source1, schema, resolver).sameType(missing1))
+    assert(StructType.findMissingFields(source1, schema, resolver)
+      .map(_.sameType(missing1)).getOrElse(false))
 
     val source2 = StructType.fromDDL("c1 INT, c3 STRING")
     val missing2 = StructType.fromDDL(
       "c2 STRUCT<c3: INT, c4: STRUCT<c5: INT, c6: INT>>")
-    assert(StructType.findMissingFields(source2, schema, resolver).sameType(missing2))
+    assert(StructType.findMissingFields(source2, schema, resolver)
+      .map(_.sameType(missing2)).getOrElse(false))
 
     val source3 = StructType.fromDDL("c1 INT, c2 STRUCT<c3: INT>")
     val missing3 = StructType.fromDDL(
       "c2 STRUCT<c4: STRUCT<c5: INT, c6: INT>>")
-    assert(StructType.findMissingFields(source3, schema, resolver).sameType(missing3))
+    assert(StructType.findMissingFields(source3, schema, resolver)
+      .map(_.sameType(missing3)).getOrElse(false))
 
     val source4 = StructType.fromDDL("c1 INT, c2 STRUCT<c3: INT, c4: STRUCT<c6: INT>>")
     val missing4 = StructType.fromDDL(
       "c2 STRUCT<c4: STRUCT<c5: INT>>")
-    assert(StructType.findMissingFields(source4, schema, resolver).sameType(missing4))
+    assert(StructType.findMissingFields(source4, schema, resolver)
+      .map(_.sameType(missing4)).getOrElse(false))
 
     val schemaWithArray = StructType.fromDDL(
       "c1 INT, c2 ARRAY<STRUCT<c3: INT, c4: LONG>>")
     val source5 = StructType.fromDDL(
       "c1 INT")
     val missing5 = StructType.fromDDL(
       "c2 ARRAY<STRUCT<c3: INT, c4: LONG>>")
-    assert(StructType.findMissingFields(source5, schemaWithArray, resolver).sameType(missing5))
+    assert(
+      StructType.findMissingFields(source5, schemaWithArray, resolver)
+        .map(_.sameType(missing5)).getOrElse(false))
 
     val schemaWithMap1 = StructType.fromDDL(
       "c1 INT, c2 MAP<STRUCT<c3: INT, c4: LONG>, STRING>, c3 LONG")
     val source6 = StructType.fromDDL(
       "c1 INT, c3 LONG")
     val missing6 = StructType.fromDDL(
       "c2 MAP<STRUCT<c3: INT, c4: LONG>, STRING>")
-    assert(StructType.findMissingFields(source6, schemaWithMap1, resolver).sameType(missing6))
+    assert(
+      StructType.findMissingFields(source6, schemaWithMap1, resolver)
+        .map(_.sameType(missing6)).getOrElse(false))
 
     val schemaWithMap2 = StructType.fromDDL(
       "c1 INT, c2 MAP<STRING, STRUCT<c3: INT, c4: LONG>>, c3 STRING")
     val source7 = StructType.fromDDL(
       "c1 INT, c3 STRING")
     val missing7 = StructType.fromDDL(
       "c2 MAP<STRING, STRUCT<c3: INT, c4: LONG>>")
-    assert(StructType.findMissingFields(source7, schemaWithMap2, resolver).sameType(missing7))
+    assert(
+      StructType.findMissingFields(source7, schemaWithMap2, resolver)
+        .map(_.sameType(missing7)).getOrElse(false))
 
     // Unsupported: nested struct in array, map
     val source8 = StructType.fromDDL(
       "c1 INT, c2 ARRAY<STRUCT<c3: INT>>")
     // `findMissingFields` doesn't support looking into nested struct in array type.
-    assert(StructType.findMissingFields(source8, schemaWithArray, resolver).length == 0)
+    assert(StructType.findMissingFields(source8, schemaWithArray, resolver).isEmpty)
 
     val source9 = StructType.fromDDL(
       "c1 INT, c2 MAP<STRUCT<c3: INT>, STRING>, c3 LONG")
     // `findMissingFields` doesn't support looking into nested struct in map type.
-    assert(StructType.findMissingFields(source9, schemaWithMap1, resolver).length == 0)
+    assert(StructType.findMissingFields(source9, schemaWithMap1, resolver).isEmpty)
 
     val source10 = StructType.fromDDL(
       "c1 INT, c2 MAP<STRING, STRUCT<c3: INT>>, c3 STRING")
     // `findMissingFields` doesn't support looking into nested struct in map type.
-    assert(StructType.findMissingFields(source10, schemaWithMap2, resolver).length == 0)
+    assert(StructType.findMissingFields(source10, schemaWithMap2, resolver).isEmpty)
   }
 }