[SPARK-23772][SQL] Provide an option to ignore column of all null values or empty array during JSON schema inference

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala

@@ -73,6 +73,9 @@ private[sql] class JSONOptions(
   val columnNameOfCorruptRecord =
     parameters.getOrElse("columnNameOfCorruptRecord", defaultColumnNameOfCorruptRecord)
 
+  // Whether to ignore column of all null values or empty array/struct during schema inference
+  val dropFieldIfAllNull = parameters.get("dropFieldIfAllNull").map(_.toBoolean).getOrElse(false)
+
   val timeZone: TimeZone = DateTimeUtils.getTimeZone(
     parameters.getOrElse(DateTimeUtils.TIMEZONE_OPTION, defaultTimeZoneId))
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala

@@ -79,7 +79,7 @@ class JacksonParser(
         val array = convertArray(parser, elementConverter)
         // Here, as we support reading top level JSON arrays and take every element
         // in such an array as a row, this case is possible.
-        if (array.numElements() == 0) {
+        if (array == null || array.numElements() == 0) {
           Nil
         } else {
           array.toArray[InternalRow](schema).toSeq
@@ -329,8 +329,12 @@ class JacksonParser(
     while (nextUntil(parser, JsonToken.END_ARRAY)) {
       values += fieldConverter.apply(parser)
     }
-
-    new GenericArrayData(values.toArray)
+    // Canonicalize arrays; an array is null if all its elements are null
+    if (options.dropFieldIfAllNull && values.forall(_ == null)) {
+      null
+    } else {
+      new GenericArrayData(values.toArray)
+    }
   }
 
   /**

sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala

@@ -379,6 +379,8 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
    * that should be used for parsing.</li>
    * <li>`samplingRatio` (default is 1.0): defines fraction of input JSON objects used
    * for schema inferring.</li>
+   * <li>`dropFieldIfAllNull` (default `false`): whether to ignore column of all null values or
+   * empty array/struct during schema inference.</li>
    * </ul>
    *
    * @since 2.0.0

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonInferSchema.scala

@@ -70,7 +70,7 @@ private[sql] object JsonInferSchema {
     }.fold(StructType(Nil))(
       compatibleRootType(columnNameOfCorruptRecord, parseMode))
 
-    canonicalizeType(rootType) match {
+    canonicalizeType(rootType, configOptions) match {
       case Some(st: StructType) => st
       case _ =>
         // canonicalizeType erases all empty structs, including the only one we want to keep
@@ -126,9 +126,13 @@ private[sql] object JsonInferSchema {
             nullable = true)
         }
         val fields: Array[StructField] = builder.result()
-        // Note: other code relies on this sorting for correctness, so don't remove it!
-        java.util.Arrays.sort(fields, structFieldComparator)
-        StructType(fields)
+        if (configOptions.dropFieldIfAllNull && fields.isEmpty) {
+          NullType
+        } else {
+          // Note: other code relies on this sorting for correctness, so don't remove it!
+          java.util.Arrays.sort(fields, structFieldComparator)
+          StructType(fields)
+        }
 
       case START_ARRAY =>
         // If this JSON array is empty, we use NullType as a placeholder.
@@ -140,7 +144,11 @@ private[sql] object JsonInferSchema {
             elementType, inferField(parser, configOptions))
         }
 
-        ArrayType(elementType)
+        if (configOptions.dropFieldIfAllNull && elementType == NullType) {
+          NullType
+        } else {
+          ArrayType(elementType)
+        }
 
       case (VALUE_NUMBER_INT | VALUE_NUMBER_FLOAT) if configOptions.primitivesAsString => StringType
 
@@ -178,10 +186,10 @@ private[sql] object JsonInferSchema {
   /**
    * Convert NullType to StringType and remove StructTypes with no fields
    */
-  private def canonicalizeType(tpe: DataType): Option[DataType] = tpe match {
+  private def canonicalizeType(tpe: DataType, options: JSONOptions): Option[DataType] = tpe match {
     case at @ ArrayType(elementType, _) =>
       for {
-        canonicalType <- canonicalizeType(elementType)
+        canonicalType <- canonicalizeType(elementType, options)
       } yield {
         at.copy(canonicalType)
       }
@@ -190,7 +198,7 @@ private[sql] object JsonInferSchema {
       val canonicalFields: Array[StructField] = for {
         field <- fields
         if field.name.length > 0
-        canonicalType <- canonicalizeType(field.dataType)
+        canonicalType <- canonicalizeType(field.dataType, options)
       } yield {
         field.copy(dataType = canonicalType)
       }
@@ -202,7 +210,7 @@ private[sql] object JsonInferSchema {
         None
       }
 
-    case NullType => Some(StringType)
+    case NullType if !options.dropFieldIfAllNull => Some(StringType)
     case other => Some(other)
   }
 

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala

@@ -2408,4 +2408,24 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
       spark.read.option("mode", "PERMISSIVE").option("encoding", "UTF-8").json(Seq(badJson).toDS()),
       Row(badJson))
   }
+
+  test("SPARK-23772 ignore column of all null values or empty array during schema inference") {
+     withTempPath { tempDir =>
+      val path = tempDir.getAbsolutePath
+      Seq(
+        """{"a":null, "b":[null, null], "c":null, "d":[[], [null]], "e":{}}""",
+        """{"a":null, "b":[null], "c":[], "d": [null, []], "e":{}}""",
+        """{"a":null, "b":[], "c":[], "d": null, "e":null}""")
+        .toDS().write.mode("overwrite").text(path)
+      val df = spark.read.format("json")
+        .option("dropFieldIfAllNull", true)
+        .load(path)
+      val expectedSchema = new StructType()
+        .add("a", NullType).add("b", NullType).add("c", NullType).add("d", NullType)
+        .add("e", NullType)
+      assert(df.schema === expectedSchema)
+      val nullRow = Row(null, null, null, null, null)
+      checkAnswer(df, nullRow :: nullRow :: nullRow :: Nil)
+    }
+  }
 }