From 3a7559b8809757ba32491a9c882ec40c8986c3b0 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Sat, 26 May 2018 18:46:08 +0200
Subject: [PATCH 01/20] Support arrays by from_json

---
 .../expressions/jsonExpressions.scala         |  8 +-
 .../sql/catalyst/json/JacksonParser.scala     |  8 ++
 .../apache/spark/sql/JsonFunctionsSuite.scala | 77 +++++++++++++++++--
 3 files changed, 84 insertions(+), 9 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index 04a4eb0ffc032..ffbc0fecb1d8c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -548,7 +548,7 @@ case class JsonToStructs(
       forceNullableSchema = SQLConf.get.getConf(SQLConf.FROM_JSON_FORCE_NULLABLE_SCHEMA))
 
   override def checkInputDataTypes(): TypeCheckResult = nullableSchema match {
-    case _: StructType | ArrayType(_: StructType, _) | _: MapType =>
+    case _: StructType | _: ArrayType | _: MapType =>
       super.checkInputDataTypes()
     case _ => TypeCheckResult.TypeCheckFailure(
       s"Input schema ${nullableSchema.simpleString} must be a struct or an array of structs.")
@@ -557,7 +557,7 @@ case class JsonToStructs(
   @transient
   lazy val rowSchema = nullableSchema match {
     case st: StructType => st
-    case ArrayType(st: StructType, _) => st
+    case at: ArrayType => at
     case mt: MapType => mt
   }
 
@@ -566,8 +566,8 @@ case class JsonToStructs(
   lazy val converter = nullableSchema match {
     case _: StructType =>
       (rows: Seq[InternalRow]) => if (rows.length == 1) rows.head else null
-    case ArrayType(_: StructType, _) =>
-      (rows: Seq[InternalRow]) => new GenericArrayData(rows)
+    case _: ArrayType =>
+      (rows: Seq[InternalRow]) => rows.head.getArray(0)
     case _: MapType =>
       (rows: Seq[InternalRow]) => rows.head.getMap(0)
   }
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
index c3a4ca8f64bf6..e23e0af9e9a7b 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
@@ -61,6 +61,7 @@ class JacksonParser(
     dt match {
       case st: StructType => makeStructRootConverter(st)
       case mt: MapType => makeMapRootConverter(mt)
+      case at: ArrayType => makeArrayRootConverter(at)
     }
   }
 
@@ -101,6 +102,13 @@ class JacksonParser(
     }
   }
 
+  private def makeArrayRootConverter(at: ArrayType): JsonParser => Seq[InternalRow] = {
+    val elemConverter = makeConverter(at.elementType)
+    (parser: JsonParser) => parseJsonToken[Seq[InternalRow]](parser, at) {
+      case START_ARRAY => Seq(InternalRow(convertArray(parser, elemConverter)))
+    }
+  }
+
   /**
    * Create a converter which converts the JSON documents held by the `JsonParser`
    * to a value according to a desired schema.
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
index 055e1fc5640f3..f4f739517c258 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
@@ -136,12 +136,11 @@ class JsonFunctionsSuite extends QueryTest with SharedSQLContext {
   test("from_json invalid schema") {
     val df = Seq("""{"a" 1}""").toDS()
     val schema = ArrayType(StringType)
-    val message = intercept[AnalysisException] {
-      df.select(from_json($"value", schema))
-    }.getMessage
 
-    assert(message.contains(
-      "Input schema array<string> must be a struct or an array of structs."))
+    checkAnswer(
+      df.select(from_json($"value", schema)),
+      Seq(Row(null))
+    )
   }
 
   test("from_json array support") {
@@ -392,4 +391,72 @@ class JsonFunctionsSuite extends QueryTest with SharedSQLContext {
     checkAnswer(Seq("""{"{"f": 1}": "a"}""").toDS().select(from_json($"value", schema)),
       Row(null))
   }
+
+  test("from_json - array of primitive types") {
+    val df = Seq("[1, 2, 3]").toDF("a")
+    val schema = new ArrayType(IntegerType, false)
+
+    checkAnswer(df.select(from_json($"a", schema)), Seq(Row(Array(1, 2, 3))))
+  }
+
+  test("from_json - array of primitive types - malformed row") {
+    val df = Seq("[1, 2 3]").toDF("a")
+    val schema = new ArrayType(IntegerType, false)
+
+    checkAnswer(df.select(from_json($"a", schema)), Seq(Row(null)))
+  }
+
+  test("from_json - array of arrays") {
+    val jsonDF = Seq("[[1], [2, 3], [4, 5, 6]]").toDF("a")
+    val schema = new ArrayType(ArrayType(IntegerType, false), false)
+    jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
+
+    checkAnswer(
+      sql("select json[0][0], json[1][1], json[2][2] from jsonTable"),
+      Seq(Row(1, 3, 6)))
+  }
+
+  test("from_json - array of arrays - malformed row") {
+    val jsonDF = Seq("[[1], [2, 3], 4, 5, 6]]").toDF("a")
+    val schema = new ArrayType(ArrayType(IntegerType, false), false)
+    jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
+
+    checkAnswer(sql("select json[0] from jsonTable"), Seq(Row(null)))
+  }
+
+  test("from_json - array of structs") {
+    val jsonDF = Seq("""[{"a":1}, {"a":2}, {"a":3}]""").toDF("a")
+    val schema = new ArrayType(new StructType().add("a", IntegerType), false)
+    jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
+
+    checkAnswer(
+      sql("select json[0], json[1], json[2] from jsonTable"),
+      Seq(Row(Row(1), Row(2), Row(3))))
+  }
+
+  test("from_json - array of structs - malformed row") {
+    val jsonDF = Seq("""[{"a":1}, {"a:2}, {"a":3}]""").toDF("a")
+    val schema = new ArrayType(new StructType().add("a", IntegerType), false)
+    jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
+
+    checkAnswer(sql("select json[0], json[1]from jsonTable"), Seq(Row(null, null)))
+  }
+
+  test("from_json - array of maps") {
+    val jsonDF = Seq("""[{"a":1}, {"b":2}]""").toDF("a")
+    val schema = new ArrayType(MapType(StringType, IntegerType, false), false)
+    jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
+
+    checkAnswer(
+      sql("""select json[0], json[1] from jsonTable"""),
+      Seq(Row(Map("a" -> 1), Map("b" -> 2))))
+  }
+
+  test("from_json - array of maps - malformed row") {
+    val jsonDF = Seq("""[{"a":1} "b":2}]""").toDF("a")
+    val schema = new ArrayType(MapType(StringType, IntegerType, false), false)
+    jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
+
+    checkAnswer(sql("""select json[0] from jsonTable"""), Seq(Row(null)))
+  }
 }

From b601a9365e12744c408a8c198a94a5c6a9e4607e Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Sat, 26 May 2018 19:08:49 +0200
Subject: [PATCH 02/20] Fix comments

---
 python/pyspark/sql/functions.py                        |  7 ++++++-
 .../main/scala/org/apache/spark/sql/functions.scala    | 10 +++++-----
 2 files changed, 11 insertions(+), 6 deletions(-)

diff --git a/python/pyspark/sql/functions.py b/python/pyspark/sql/functions.py
index fbc8a2d038f8f..7100c666f4451 100644
--- a/python/pyspark/sql/functions.py
+++ b/python/pyspark/sql/functions.py
@@ -2115,7 +2115,7 @@ def json_tuple(col, *fields):
 def from_json(col, schema, options={}):
     """
     Parses a column containing a JSON string into a :class:`MapType` with :class:`StringType`
-    as keys type, :class:`StructType` or :class:`ArrayType` of :class:`StructType`\\s with
+    as keys type, :class:`StructType` or :class:`ArrayType` with
     the specified schema. Returns `null`, in the case of an unparseable string.
 
     :param col: string column in json format
@@ -2141,6 +2141,11 @@ def from_json(col, schema, options={}):
     >>> df = spark.createDataFrame(data, ("key", "value"))
     >>> df.select(from_json(df.value, schema).alias("json")).collect()
     [Row(json=[Row(a=1)])]
+    >>> data = [(1, '''[1, 2, 3]''')]
+    >>> schema = ArrayType(IntegerType())
+    >>> df = spark.createDataFrame(data, ("key", "value"))
+    >>> df.select(from_json(df.value, schema).alias("json")).collect()
+    [Row(json=[1, 2, 3])]
     """
 
     sc = SparkContext._active_spark_context
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/functions.scala b/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
index 5ab9cb3fb86a5..50d0e80aa7e18 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
@@ -3243,7 +3243,7 @@ object functions {
 
   /**
    * (Scala-specific) Parses a column containing a JSON string into a `MapType` with `StringType`
-   * as keys type, `StructType` or `ArrayType` of `StructType`s with the specified schema.
+   * as keys type, `StructType` or `ArrayType` with the specified schema.
    * Returns `null`, in the case of an unparseable string.
    *
    * @param e a string column containing JSON data.
@@ -3275,7 +3275,7 @@ object functions {
 
   /**
    * (Java-specific) Parses a column containing a JSON string into a `MapType` with `StringType`
-   * as keys type, `StructType` or `ArrayType` of `StructType`s with the specified schema.
+   * as keys type, `StructType` or `ArrayType` with the specified schema.
    * Returns `null`, in the case of an unparseable string.
    *
    * @param e a string column containing JSON data.
@@ -3304,7 +3304,7 @@ object functions {
 
   /**
    * Parses a column containing a JSON string into a `MapType` with `StringType` as keys type,
-   * `StructType` or `ArrayType` of `StructType`s with the specified schema.
+   * `StructType` or `ArrayType` with the specified schema.
    * Returns `null`, in the case of an unparseable string.
    *
    * @param e a string column containing JSON data.
@@ -3318,7 +3318,7 @@ object functions {
 
   /**
    * (Java-specific) Parses a column containing a JSON string into a `MapType` with `StringType`
-   * as keys type, `StructType` or `ArrayType` of `StructType`s with the specified schema.
+   * as keys type, `StructType` or `ArrayType` with the specified schema.
    * Returns `null`, in the case of an unparseable string.
    *
    * @param e a string column containing JSON data.
@@ -3335,7 +3335,7 @@ object functions {
 
   /**
    * (Scala-specific) Parses a column containing a JSON string into a `MapType` with `StringType`
-   * as keys type, `StructType` or `ArrayType` of `StructType`s with the specified schema.
+   * as keys type, `StructType` or `ArrayType` with the specified schema.
    * Returns `null`, in the case of an unparseable string.
    *
    * @param e a string column containing JSON data.

From 02a97acdde938af1479c7bd93236c353fefc35d7 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Sat, 26 May 2018 23:19:12 +0200
Subject: [PATCH 03/20] Support array of struct unpacking for backward
 compatibility

---
 .../spark/sql/catalyst/expressions/jsonExpressions.scala  | 7 +++++++
 .../sql/catalyst/expressions/JsonExpressionsSuite.scala   | 8 ++++++--
 2 files changed, 13 insertions(+), 2 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index ffbc0fecb1d8c..a8c23e87a6d84 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -523,6 +523,8 @@ case class JsonToStructs(
   // can generate incorrect files if values are missing in columns declared as non-nullable.
   val nullableSchema = if (forceNullableSchema) schema.asNullable else schema
 
+  val unpackArray: Boolean = options.get("unpackArray").map(_.toBoolean).getOrElse(false)
+
   override def nullable: Boolean = true
 
   // Used in `FunctionRegistry`
@@ -548,6 +550,8 @@ case class JsonToStructs(
       forceNullableSchema = SQLConf.get.getConf(SQLConf.FROM_JSON_FORCE_NULLABLE_SCHEMA))
 
   override def checkInputDataTypes(): TypeCheckResult = nullableSchema match {
+    case ArrayType(_: StructType, _) if unpackArray =>
+      super.checkInputDataTypes()
     case _: StructType | _: ArrayType | _: MapType =>
       super.checkInputDataTypes()
     case _ => TypeCheckResult.TypeCheckFailure(
@@ -557,6 +561,7 @@ case class JsonToStructs(
   @transient
   lazy val rowSchema = nullableSchema match {
     case st: StructType => st
+    case ArrayType(st: StructType, _) if unpackArray => st
     case at: ArrayType => at
     case mt: MapType => mt
   }
@@ -566,6 +571,8 @@ case class JsonToStructs(
   lazy val converter = nullableSchema match {
     case _: StructType =>
       (rows: Seq[InternalRow]) => if (rows.length == 1) rows.head else null
+    case ArrayType(_: StructType, _) if unpackArray =>
+      (rows: Seq[InternalRow]) => new GenericArrayData(rows)
     case _: ArrayType =>
       (rows: Seq[InternalRow]) => rows.head.getArray(0)
     case _: MapType =>
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
index 00e97637eee7e..3cad76eefcfa3 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
@@ -423,7 +423,9 @@ class JsonExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper with
     val input = """{"a": 1}"""
     val schema = ArrayType(StructType(StructField("a", IntegerType) :: Nil))
     val output = InternalRow(1) :: Nil
-    checkEvaluation(JsonToStructs(schema, Map.empty, Literal(input), gmtId, true), output)
+    checkEvaluation(
+      JsonToStructs(schema, Map("unpackArray" -> "true"), Literal(input), gmtId, true),
+      output)
   }
 
   test("from_json - input=empty array, schema=array, output=empty array") {
@@ -437,7 +439,9 @@ class JsonExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper with
     val input = "{ }"
     val schema = ArrayType(StructType(StructField("a", IntegerType) :: Nil))
     val output = InternalRow(null) :: Nil
-    checkEvaluation(JsonToStructs(schema, Map.empty, Literal(input), gmtId, true), output)
+    checkEvaluation(
+      JsonToStructs(schema, Map("unpackArray" -> "true"), Literal(input), gmtId, true),
+      output)
   }
 
   test("from_json - input=array of single object, schema=struct, output=single row") {

From 86d2f20e7d6219809ea6411d0fcf8577e6fabf44 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Thu, 31 May 2018 17:11:46 +0200
Subject: [PATCH 04/20] Added case insensitive options for jsonToStruct

---
 .../spark/sql/catalyst/expressions/jsonExpressions.scala | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index a8c23e87a6d84..3dd12acc25298 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.expressions
 
-import java.io.{ByteArrayInputStream, ByteArrayOutputStream, CharArrayWriter, InputStreamReader, StringWriter}
+import java.io._
 
 import scala.util.parsing.combinator.RegexParsers
 
@@ -29,7 +29,7 @@ import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
 import org.apache.spark.sql.catalyst.json._
 import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
-import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, ArrayData, BadRecordException, FailFastMode, GenericArrayData, MapData}
+import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.UTF8String
@@ -523,7 +523,10 @@ case class JsonToStructs(
   // can generate incorrect files if values are missing in columns declared as non-nullable.
   val nullableSchema = if (forceNullableSchema) schema.asNullable else schema
 
-  val unpackArray: Boolean = options.get("unpackArray").map(_.toBoolean).getOrElse(false)
+  val caseInsensitiveOptions = CaseInsensitiveMap(options)
+  val unpackArray: Boolean = {
+    caseInsensitiveOptions.get("unpackArray").map(_.toBoolean).getOrElse(false)
+  }
 
   override def nullable: Boolean = true
 

From 9d0230aa9c335e69099a0ddfc280244bb42a360b Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Thu, 31 May 2018 17:16:01 +0200
Subject: [PATCH 05/20] Making added values private

---
 .../spark/sql/catalyst/expressions/jsonExpressions.scala      | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index 3dd12acc25298..caf32430878f2 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -523,8 +523,8 @@ case class JsonToStructs(
   // can generate incorrect files if values are missing in columns declared as non-nullable.
   val nullableSchema = if (forceNullableSchema) schema.asNullable else schema
 
-  val caseInsensitiveOptions = CaseInsensitiveMap(options)
-  val unpackArray: Boolean = {
+  private val caseInsensitiveOptions = CaseInsensitiveMap(options)
+  private val unpackArray: Boolean = {
     caseInsensitiveOptions.get("unpackArray").map(_.toBoolean).getOrElse(false)
   }
 

From 181dcae04393776600c0061691a3b3a6ab999915 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Mon, 11 Jun 2018 19:34:36 +0200
Subject: [PATCH 06/20] Unnecessary check of input params is removed

---
 .../apache/spark/sql/catalyst/expressions/jsonExpressions.scala | 2 --
 1 file changed, 2 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index caf32430878f2..111e2a8dce6b2 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -553,8 +553,6 @@ case class JsonToStructs(
       forceNullableSchema = SQLConf.get.getConf(SQLConf.FROM_JSON_FORCE_NULLABLE_SCHEMA))
 
   override def checkInputDataTypes(): TypeCheckResult = nullableSchema match {
-    case ArrayType(_: StructType, _) if unpackArray =>
-      super.checkInputDataTypes()
     case _: StructType | _: ArrayType | _: MapType =>
       super.checkInputDataTypes()
     case _ => TypeCheckResult.TypeCheckFailure(

From 6d54cf003ec397deedcfe25f45941bef47649e36 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Mon, 11 Jun 2018 20:22:23 +0200
Subject: [PATCH 07/20] Added comment for the unpackArray config

---
 .../spark/sql/catalyst/expressions/jsonExpressions.scala       | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index 111e2a8dce6b2..a782295883bca 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -524,6 +524,9 @@ case class JsonToStructs(
   val nullableSchema = if (forceNullableSchema) schema.asNullable else schema
 
   private val caseInsensitiveOptions = CaseInsensitiveMap(options)
+  // The flag enables old behavior which allows to specify array of struct as a schema
+  // for the json string `{...}`. For example, if the schema is ArrayType(StructType(...)),
+  // the array type will be unpacked and StructType will be applied for JSON string.
   private val unpackArray: Boolean = {
     caseInsensitiveOptions.get("unpackArray").map(_.toBoolean).getOrElse(false)
   }

From e321e37b0aaf2d17ab267b685ea14ee778f06192 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Mon, 11 Jun 2018 20:27:53 +0200
Subject: [PATCH 08/20] Added a test when schema is array type but json input
 is {}

---
 .../sql/catalyst/expressions/JsonExpressionsSuite.scala  | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
index 3cad76eefcfa3..4a124e7e3a981 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
@@ -428,6 +428,15 @@ class JsonExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper with
       output)
   }
 
+  test("from_json - input=object, schema=array, output=null") {
+    val input = """{"a": 1}"""
+    val schema = ArrayType(StructType(StructField("a", IntegerType) :: Nil))
+    val output = null
+    checkEvaluation(
+      JsonToStructs(schema, Map("unpackArray" -> "false"), Literal(input), gmtId, true),
+      output)
+  }
+
   test("from_json - input=empty array, schema=array, output=empty array") {
     val input = "[ ]"
     val schema = ArrayType(StructType(StructField("a", IntegerType) :: Nil))

From b5b0d9c4c8b1ea97e110519f27d55a55e3ce8cad Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Fri, 15 Jun 2018 01:14:18 +0200
Subject: [PATCH 09/20] Making imports shorter

---
 .../spark/sql/catalyst/expressions/jsonExpressions.scala      | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index a609ba8d3933a..f4ef32aa7d3d1 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.expressions
 
-import java.io.{ByteArrayInputStream, ByteArrayOutputStream, CharArrayWriter, InputStreamReader, StringWriter}
+import java.io._
 
 import scala.util.parsing.combinator.RegexParsers
 
@@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
 import org.apache.spark.sql.catalyst.json._
-import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, ArrayData, BadRecordException, FailFastMode, GenericArrayData, MapData}
+import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.UTF8String

From ce9918b2598a4d70a95d2e8269a32afe2db2ec26 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Fri, 15 Jun 2018 01:24:58 +0200
Subject: [PATCH 10/20] SQL tests for arrays

---
 .../sql-tests/inputs/json-functions.sql       |  3 +++
 .../sql-tests/results/json-functions.sql.out  | 26 ++++++++++++++++++-
 2 files changed, 28 insertions(+), 1 deletion(-)

diff --git a/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql b/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql
index dc15d13cd1dd3..762068c6fbff4 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql
@@ -35,3 +35,6 @@ DROP VIEW IF EXISTS jsonTable;
 -- from_json - complex types
 select from_json('{"a":1, "b":2}', 'map<string, int>');
 select from_json('{"a":1, "b":"2"}', 'struct<a:int,b:string>');
+select from_json('[1, 2, 3]', 'array<int>');
+select from_json('[{"a": 1}, {"a":2}]', 'array<struct<a:int>>');
+select from_json('[{"a": 1}, {"b":2}]', 'array<map<string,int>>');
diff --git a/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out b/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out
index 2b3288dc5a137..b2e02a3e11f85 100644
--- a/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out
@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 28
+-- Number of queries: 31
 
 
 -- !query 0
@@ -274,3 +274,27 @@ select from_json('{"a":1, "b":"2"}', 'struct<a:int,b:string>')
 struct<jsontostructs({"a":1, "b":"2"}):struct<a:int,b:string>>
 -- !query 27 output
 {"a":1,"b":"2"}
+
+
+-- !query 28
+select from_json('[1, 2, 3]', 'array<int>')
+-- !query 28 schema
+struct<jsontostructs([1, 2, 3]):array<int>>
+-- !query 28 output
+[1,2,3]
+
+
+-- !query 29
+select from_json('[{"a": 1}, {"a":2}]', 'array<struct<a:int>>')
+-- !query 29 schema
+struct<jsontostructs([{"a": 1}, {"a":2}]):array<struct<a:int>>>
+-- !query 29 output
+[{"a":1},{"a":2}]
+
+
+-- !query 30
+select from_json('[{"a": 1}, {"b":2}]', 'array<map<string,int>>')
+-- !query 30 schema
+struct<jsontostructs([{"a": 1}, {"b":2}]):array<map<string,int>>>
+-- !query 30 output
+[{"a":1},{"b":2}]

From fced8ec9557ad8173a6e1be07d7f4f11369e5727 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Fri, 29 Jun 2018 14:33:36 +0200
Subject: [PATCH 11/20] Enable unpackArray by default to keep backward
 compatibility

---
 .../spark/sql/catalyst/expressions/jsonExpressions.scala | 9 +++++----
 .../sql/catalyst/expressions/JsonExpressionsSuite.scala  | 4 +---
 2 files changed, 6 insertions(+), 7 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index f4ef32aa7d3d1..ccecf8ca46e0d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -523,11 +523,12 @@ case class JsonToStructs(
   val nullableSchema = if (forceNullableSchema) schema.asNullable else schema
 
   private val caseInsensitiveOptions = CaseInsensitiveMap(options)
-  // The flag enables old behavior which allows to specify array of struct as a schema
-  // for the json string `{...}`. For example, if the schema is ArrayType(StructType(...)),
-  // the array type will be unpacked and StructType will be applied for JSON string.
+  // The flag allows to specify a schema as an array of structs for an json structs like `{...}`.
+  // For example, if the schema is ArrayType(StructType(...)),
+  // the array type will be unpacked and StructType will be applied to JSON strings.
+  // The behavior is turned on by default.
   private val unpackArray: Boolean = {
-    caseInsensitiveOptions.get("unpackArray").map(_.toBoolean).getOrElse(false)
+    caseInsensitiveOptions.get("unpackArray").map(_.toBoolean).getOrElse(true)
   }
 
   override def nullable: Boolean = true
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
index 4a124e7e3a981..3a069de1f2dbe 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
@@ -448,9 +448,7 @@ class JsonExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper with
     val input = "{ }"
     val schema = ArrayType(StructType(StructField("a", IntegerType) :: Nil))
     val output = InternalRow(null) :: Nil
-    checkEvaluation(
-      JsonToStructs(schema, Map("unpackArray" -> "true"), Literal(input), gmtId, true),
-      output)
+    checkEvaluation(JsonToStructs(schema, Map.empty, Literal(input), gmtId, true), output)
   }
 
   test("from_json - input=array of single object, schema=struct, output=single row") {

From e49ee9d0b1ae8bb4a1b61968dae0f0450891de06 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <max.gekk@gmail.com>
Date: Sat, 7 Jul 2018 14:17:39 +0200
Subject: [PATCH 12/20] Updating of sql tests

---
 .../sql-tests/inputs/json-functions.sql       |  8 +++---
 .../sql-tests/results/json-functions.sql.out  | 26 ++++++++++++++++++-
 2 files changed, 30 insertions(+), 4 deletions(-)

diff --git a/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql b/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql
index 2a0b4a890edb0..a9c537a62d04f 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql
@@ -35,10 +35,12 @@ DROP VIEW IF EXISTS jsonTable;
 -- from_json - complex types
 select from_json('{"a":1, "b":2}', 'map<string, int>');
 select from_json('{"a":1, "b":"2"}', 'struct<a:int,b:string>');
-select from_json('[1, 2, 3]', 'array<int>');
-select from_json('[{"a": 1}, {"a":2}]', 'array<struct<a:int>>');
-select from_json('[{"a": 1}, {"b":2}]', 'array<map<string,int>>');
 
 -- infer schema of json literal
 select schema_of_json('{"c1":0, "c2":[1]}');
 select from_json('{"c1":[1, 2, 3]}', schema_of_json('{"c1":[0]}'));
+
+-- from_json - array type
+select from_json('[1, 2, 3]', 'array<int>');
+select from_json('[{"a": 1}, {"a":2}]', 'array<struct<a:int>>');
+select from_json('[{"a": 1}, {"b":2}]', 'array<map<string,int>>');
diff --git a/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out b/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out
index 3d49323751a10..2fd1f6e5653c6 100644
--- a/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out
@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 30
+-- Number of queries: 33
 
 
 -- !query 0
@@ -290,3 +290,27 @@ select from_json('{"c1":[1, 2, 3]}', schema_of_json('{"c1":[0]}'))
 struct<jsontostructs({"c1":[1, 2, 3]}):struct<c1:array<bigint>>>
 -- !query 29 output
 {"c1":[1,2,3]}
+
+
+-- !query 30
+select from_json('[1, 2, 3]', 'array<int>')
+-- !query 30 schema
+struct<jsontostructs([1, 2, 3]):array<int>>
+-- !query 30 output
+[1,2,3]
+
+
+-- !query 31
+select from_json('[{"a": 1}, {"a":2}]', 'array<struct<a:int>>')
+-- !query 31 schema
+struct<jsontostructs([{"a": 1}, {"a":2}]):array<struct<a:int>>>
+-- !query 31 output
+[{"a":1},{"a":2}]
+
+
+-- !query 32
+select from_json('[{"a": 1}, {"b":2}]', 'array<map<string,int>>')
+-- !query 32 schema
+struct<jsontostructs([{"a": 1}, {"b":2}]):array<map<string,int>>>
+-- !query 32 output
+[{"a":1},{"b":2}]

From 2bca7e0f231ab5224db702d181ddf6a17ac9824a Mon Sep 17 00:00:00 2001
From: Maxim Gekk <max.gekk@gmail.com>
Date: Sat, 7 Jul 2018 14:24:14 +0200
Subject: [PATCH 13/20] Fix python tests

---
 python/pyspark/sql/functions.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/python/pyspark/sql/functions.py b/python/pyspark/sql/functions.py
index 749e6a54639ad..7c69c94de1414 100644
--- a/python/pyspark/sql/functions.py
+++ b/python/pyspark/sql/functions.py
@@ -2209,14 +2209,14 @@ def from_json(col, schema, options={}):
     >>> df = spark.createDataFrame(data, ("key", "value"))
     >>> df.select(from_json(df.value, schema).alias("json")).collect()
     [Row(json=[Row(a=1)])]
+    >>> schema = schema_of_json(lit('''{"a": 0}'''))
+    >>> df.select(from_json(df.value, schema).alias("json")).collect()
+    [Row(json=Row(a=1))]
     >>> data = [(1, '''[1, 2, 3]''')]
     >>> schema = ArrayType(IntegerType())
     >>> df = spark.createDataFrame(data, ("key", "value"))
     >>> df.select(from_json(df.value, schema).alias("json")).collect()
     [Row(json=[1, 2, 3])]
-    >>> schema = schema_of_json(lit('''{"a": 0}'''))
-    >>> df.select(from_json(df.value, schema).alias("json")).collect()
-    [Row(json=Row(a=1))]
     """
 
     sc = SparkContext._active_spark_context

From 758d1dfebc604715aac826c88c7ff8421095b05e Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Fri, 13 Jul 2018 21:18:26 +0200
Subject: [PATCH 14/20] Fix tests - removing unused parameter

---
 .../spark/sql/catalyst/expressions/JsonExpressionsSuite.scala | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
index cdbb1378de452..0caecf0aff7d6 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
@@ -424,7 +424,7 @@ class JsonExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper with
     val schema = ArrayType(StructType(StructField("a", IntegerType) :: Nil))
     val output = InternalRow(1) :: Nil
     checkEvaluation(
-      JsonToStructs(schema, Map("unpackArray" -> "true"), Literal(input), gmtId, true),
+      JsonToStructs(schema, Map("unpackArray" -> "true"), Literal(input), gmtId),
       output)
   }
 
@@ -433,7 +433,7 @@ class JsonExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper with
     val schema = ArrayType(StructType(StructField("a", IntegerType) :: Nil))
     val output = null
     checkEvaluation(
-      JsonToStructs(schema, Map("unpackArray" -> "false"), Literal(input), gmtId, true),
+      JsonToStructs(schema, Map("unpackArray" -> "false"), Literal(input), gmtId),
       output)
   }
 

From 2746d35e567ac2eb8e9947e90d3ecf055177b0c4 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Sat, 21 Jul 2018 21:04:16 +0200
Subject: [PATCH 15/20] Removing unpackArray option

---
 .../sql/catalyst/expressions/jsonExpressions.scala   | 12 ------------
 .../spark/sql/catalyst/json/JacksonParser.scala      |  4 ++++
 .../catalyst/expressions/JsonExpressionsSuite.scala  | 11 +----------
 3 files changed, 5 insertions(+), 22 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index 4ad2892161407..73d92f16245b9 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -524,15 +524,6 @@ case class JsonToStructs(
   // can generate incorrect files if values are missing in columns declared as non-nullable.
   val nullableSchema = if (forceNullableSchema) schema.asNullable else schema
 
-  private val caseInsensitiveOptions = CaseInsensitiveMap(options)
-  // The flag allows to specify a schema as an array of structs for an json structs like `{...}`.
-  // For example, if the schema is ArrayType(StructType(...)),
-  // the array type will be unpacked and StructType will be applied to JSON strings.
-  // The behavior is turned on by default.
-  private val unpackArray: Boolean = {
-    caseInsensitiveOptions.get("unpackArray").map(_.toBoolean).getOrElse(true)
-  }
-
   override def nullable: Boolean = true
 
   // Used in `FunctionRegistry`
@@ -562,7 +553,6 @@ case class JsonToStructs(
   @transient
   lazy val rowSchema = nullableSchema match {
     case st: StructType => st
-    case ArrayType(st: StructType, _) if unpackArray => st
     case at: ArrayType => at
     case mt: MapType => mt
   }
@@ -572,8 +562,6 @@ case class JsonToStructs(
   lazy val converter = nullableSchema match {
     case _: StructType =>
       (rows: Seq[InternalRow]) => if (rows.length == 1) rows.head else null
-    case ArrayType(_: StructType, _) if unpackArray =>
-      (rows: Seq[InternalRow]) => new GenericArrayData(rows)
     case _: ArrayType =>
       (rows: Seq[InternalRow]) => rows.head.getArray(0)
     case _: MapType =>
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
index bf00512c5db02..97ae2f6ef8b18 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
@@ -106,6 +106,10 @@ class JacksonParser(
     val elemConverter = makeConverter(at.elementType)
     (parser: JsonParser) => parseJsonToken[Seq[InternalRow]](parser, at) {
       case START_ARRAY => Seq(InternalRow(convertArray(parser, elemConverter)))
+      case START_OBJECT if at.elementType.isInstanceOf[StructType] =>
+        val st = at.elementType.asInstanceOf[StructType]
+        val fieldConverters = st.map(_.dataType).map(makeConverter).toArray
+        Seq(InternalRow(new GenericArrayData(Seq(convertObject(parser, st, fieldConverters)))))
     }
   }
 
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
index 0caecf0aff7d6..af2aaf4778eb1 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
@@ -424,16 +424,7 @@ class JsonExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper with
     val schema = ArrayType(StructType(StructField("a", IntegerType) :: Nil))
     val output = InternalRow(1) :: Nil
     checkEvaluation(
-      JsonToStructs(schema, Map("unpackArray" -> "true"), Literal(input), gmtId),
-      output)
-  }
-
-  test("from_json - input=object, schema=array, output=null") {
-    val input = """{"a": 1}"""
-    val schema = ArrayType(StructType(StructField("a", IntegerType) :: Nil))
-    val output = null
-    checkEvaluation(
-      JsonToStructs(schema, Map("unpackArray" -> "false"), Literal(input), gmtId),
+      JsonToStructs(schema, Map.empty, Literal(input), gmtId),
       output)
   }
 

From bc3a2ddb870a220cc075337b6ada31614901f591 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Sat, 21 Jul 2018 22:30:34 +0200
Subject: [PATCH 16/20] Removing unused val

---
 .../spark/sql/catalyst/expressions/jsonExpressions.scala | 9 +--------
 1 file changed, 1 insertion(+), 8 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index 73d92f16245b9..392d2db39c608 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -550,13 +550,6 @@ case class JsonToStructs(
       s"Input schema ${nullableSchema.catalogString} must be a struct or an array of structs.")
   }
 
-  @transient
-  lazy val rowSchema = nullableSchema match {
-    case st: StructType => st
-    case at: ArrayType => at
-    case mt: MapType => mt
-  }
-
   // This converts parsed rows to the desired output by the given schema.
   @transient
   lazy val converter = nullableSchema match {
@@ -571,7 +564,7 @@ case class JsonToStructs(
   @transient
   lazy val parser =
     new JacksonParser(
-      rowSchema,
+      nullableSchema,
       new JSONOptions(options + ("mode" -> FailFastMode.name), timeZoneId.get))
 
   override def dataType: DataType = nullableSchema

From 39a0a4e96266747674814e5a19821d994e438ea1 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Tue, 24 Jul 2018 15:13:19 +0200
Subject: [PATCH 17/20] Reverting unrelated changes

---
 .../spark/sql/catalyst/expressions/JsonExpressionsSuite.scala | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
index af2aaf4778eb1..04f1c8ce0b83d 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
@@ -423,9 +423,7 @@ class JsonExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper with
     val input = """{"a": 1}"""
     val schema = ArrayType(StructType(StructField("a", IntegerType) :: Nil))
     val output = InternalRow(1) :: Nil
-    checkEvaluation(
-      JsonToStructs(schema, Map.empty, Literal(input), gmtId),
-      output)
+    checkEvaluation(JsonToStructs(schema, Map.empty, Literal(input), gmtId), output)
   }
 
   test("from_json - input=empty array, schema=array, output=empty array") {

From 021350b29c11c19c5a1b8343553c4a531abe39a2 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Sat, 28 Jul 2018 22:49:40 +0200
Subject: [PATCH 18/20] Addressing Liang-Chi Hsieh's review comments

---
 .../spark/sql/catalyst/expressions/jsonExpressions.scala   | 4 ++--
 .../org/apache/spark/sql/catalyst/json/JacksonParser.scala | 4 ++++
 .../scala/org/apache/spark/sql/JsonFunctionsSuite.scala    | 7 ++-----
 3 files changed, 8 insertions(+), 7 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index 392d2db39c608..ca99100b6d64f 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -495,7 +495,7 @@ case class JsonTuple(children: Seq[Expression])
 }
 
 /**
- * Converts an json input string to a [[StructType]] or [[ArrayType]] of [[StructType]]s
+ * Converts an json input string to a [[StructType]], [[ArrayType]] or [[MapType]]
  * with the specified schema.
  */
 // scalastyle:off line.size.limit
@@ -547,7 +547,7 @@ case class JsonToStructs(
     case _: StructType | _: ArrayType | _: MapType =>
       super.checkInputDataTypes()
     case _ => TypeCheckResult.TypeCheckFailure(
-      s"Input schema ${nullableSchema.catalogString} must be a struct or an array of structs.")
+      s"Input schema ${nullableSchema.catalogString} must be a struct, an array or a map.")
   }
 
   // This converts parsed rows to the desired output by the given schema.
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
index 97ae2f6ef8b18..b21863d151a8c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
@@ -107,6 +107,10 @@ class JacksonParser(
     (parser: JsonParser) => parseJsonToken[Seq[InternalRow]](parser, at) {
       case START_ARRAY => Seq(InternalRow(convertArray(parser, elemConverter)))
       case START_OBJECT if at.elementType.isInstanceOf[StructType] =>
+        // This handles the case when an input JSON object is a structure but
+        // the specified schema is an array of structures. In that case, the input JSON is
+        // considered as an array of only one element of struct type.
+        // This behavior was introduced by changes for SPARK-19595.
         val st = at.elementType.asInstanceOf[StructType]
         val fieldConverters = st.map(_.dataType).map(makeConverter).toArray
         Seq(InternalRow(new GenericArrayData(Seq(convertObject(parser, st, fieldConverters)))))
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
index 7be7a844c92e3..f321ab86e9b7f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
@@ -133,14 +133,11 @@ class JsonFunctionsSuite extends QueryTest with SharedSQLContext {
       Row(null) :: Nil)
   }
 
-  test("from_json invalid schema") {
+  test("from_json - json doesn't conform to the array type") {
     val df = Seq("""{"a" 1}""").toDS()
     val schema = ArrayType(StringType)
 
-    checkAnswer(
-      df.select(from_json($"value", schema)),
-      Seq(Row(null))
-    )
+    checkAnswer(df.select(from_json($"value", schema)), Seq(Row(null)))
   }
 
   test("from_json array support") {

From bdfd8a1afd3a8ccc04c8a48d8a5fc77548881de1 Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Sun, 12 Aug 2018 22:53:16 +0200
Subject: [PATCH 19/20] Added an example

---
 .../spark/sql/catalyst/json/JacksonParser.scala    | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
index b21863d151a8c..6feea500b2aa0 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
@@ -111,6 +111,20 @@ class JacksonParser(
         // the specified schema is an array of structures. In that case, the input JSON is
         // considered as an array of only one element of struct type.
         // This behavior was introduced by changes for SPARK-19595.
+        //
+        // For example, if the specified schema is ArrayType(new StructType().add("i", IntegerType))
+        // and JSON input as below:
+        //
+        // [{"i": 1}, {"i": 2}]
+        // [{"i": 3}]
+        // {"i": 4}
+        //
+        // The last row is considered as an array with one element, and result of conversion:
+        //
+        // Seq(Row(1), Row(2))
+        // Seq(Row(3))
+        // Seq(Row(4))
+        //
         val st = at.elementType.asInstanceOf[StructType]
         val fieldConverters = st.map(_.dataType).map(makeConverter).toArray
         Seq(InternalRow(new GenericArrayData(Seq(convertObject(parser, st, fieldConverters)))))

From 74a779964b666b36b36a65b2cdd4b47d9df1e04c Mon Sep 17 00:00:00 2001
From: Maxim Gekk <maxim.gekk@databricks.com>
Date: Sun, 12 Aug 2018 23:22:14 +0200
Subject: [PATCH 20/20] A few negative SQL tests

---
 .../sql-tests/inputs/json-functions.sql       |  7 +++
 .../sql-tests/results/json-functions.sql.out  | 52 ++++++++++++++++---
 2 files changed, 53 insertions(+), 6 deletions(-)

diff --git a/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql b/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql
index a9c537a62d04f..0cf370c13e8c0 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/json-functions.sql
@@ -42,5 +42,12 @@ select from_json('{"c1":[1, 2, 3]}', schema_of_json('{"c1":[0]}'));
 
 -- from_json - array type
 select from_json('[1, 2, 3]', 'array<int>');
+select from_json('[1, "2", 3]', 'array<int>');
+select from_json('[1, 2, null]', 'array<int>');
+
 select from_json('[{"a": 1}, {"a":2}]', 'array<struct<a:int>>');
+select from_json('{"a": 1}', 'array<struct<a:int>>');
+select from_json('[null, {"a":2}]', 'array<struct<a:int>>');
+
 select from_json('[{"a": 1}, {"b":2}]', 'array<map<string,int>>');
+select from_json('[{"a": 1}, 2]', 'array<map<string,int>>');
diff --git a/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out b/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out
index 95257960b1899..b44883b070663 100644
--- a/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/json-functions.sql.out
@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 33
+-- Number of queries: 38
 
 
 -- !query 0
@@ -301,16 +301,56 @@ struct<jsontostructs([1, 2, 3]):array<int>>
 
 
 -- !query 31
-select from_json('[{"a": 1}, {"a":2}]', 'array<struct<a:int>>')
+select from_json('[1, "2", 3]', 'array<int>')
 -- !query 31 schema
-struct<jsontostructs([{"a": 1}, {"a":2}]):array<struct<a:int>>>
+struct<jsontostructs([1, "2", 3]):array<int>>
 -- !query 31 output
-[{"a":1},{"a":2}]
+NULL
 
 
 -- !query 32
-select from_json('[{"a": 1}, {"b":2}]', 'array<map<string,int>>')
+select from_json('[1, 2, null]', 'array<int>')
 -- !query 32 schema
-struct<jsontostructs([{"a": 1}, {"b":2}]):array<map<string,int>>>
+struct<jsontostructs([1, 2, null]):array<int>>
 -- !query 32 output
+[1,2,null]
+
+
+-- !query 33
+select from_json('[{"a": 1}, {"a":2}]', 'array<struct<a:int>>')
+-- !query 33 schema
+struct<jsontostructs([{"a": 1}, {"a":2}]):array<struct<a:int>>>
+-- !query 33 output
+[{"a":1},{"a":2}]
+
+
+-- !query 34
+select from_json('{"a": 1}', 'array<struct<a:int>>')
+-- !query 34 schema
+struct<jsontostructs({"a": 1}):array<struct<a:int>>>
+-- !query 34 output
+[{"a":1}]
+
+
+-- !query 35
+select from_json('[null, {"a":2}]', 'array<struct<a:int>>')
+-- !query 35 schema
+struct<jsontostructs([null, {"a":2}]):array<struct<a:int>>>
+-- !query 35 output
+[null,{"a":2}]
+
+
+-- !query 36
+select from_json('[{"a": 1}, {"b":2}]', 'array<map<string,int>>')
+-- !query 36 schema
+struct<jsontostructs([{"a": 1}, {"b":2}]):array<map<string,int>>>
+-- !query 36 output
 [{"a":1},{"b":2}]
+
+
+-- !query 37
+select from_json('[{"a": 1}, 2]', 'array<map<string,int>>')
+-- !query 37 schema
+struct<jsontostructs([{"a": 1}, 2]):array<map<string,int>>>
+-- !query 37 output
+NULL