[SPARK-24543][SQL] Support any type as DDL string for from_json's schema

python/pyspark/sql/functions.py

@@ -2168,8 +2168,7 @@ def from_json(col, schema, options={}):
     [Row(json=Row(a=1))]
     >>> df.select(from_json(df.value, "a INT").alias("json")).collect()
     [Row(json=Row(a=1))]
-    >>> schema = MapType(StringType(), IntegerType())
-    >>> df.select(from_json(df.value, schema).alias("json")).collect()
+    >>> df.select(from_json(df.value, "MAP<STRING,INT>").alias("json")).collect()
     [Row(json={u'a': 1})]
     >>> data = [(1, '''[{"a": 1}]''')]
     >>> schema = ArrayType(StructType([StructField("a", IntegerType())]))

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala

@@ -19,6 +19,7 @@ package org.apache.spark.sql.catalyst.expressions
 
 import java.io.{ByteArrayInputStream, ByteArrayOutputStream, CharArrayWriter, InputStreamReader, StringWriter}
 
+import scala.util.control.NonFatal
 import scala.util.parsing.combinator.RegexParsers
 
 import com.fasterxml.jackson.core._
@@ -747,8 +748,14 @@ case class StructsToJson(
 
 object JsonExprUtils {
 
-  def validateSchemaLiteral(exp: Expression): StructType = exp match {
-    case Literal(s, StringType) => CatalystSqlParser.parseTableSchema(s.toString)
+  def validateSchemaLiteral(exp: Expression): DataType = exp match {
+    case Literal(s, StringType) =>
+      val schema = s.toString
+      try {
+        CatalystSqlParser.parseDataType(schema)
+      } catch {
+        case NonFatal(_) => CatalystSqlParser.parseTableSchema(schema)
+      }
     case e => throw new AnalysisException(s"Expected a string literal instead of $e")
   }
 

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

@@ -26,6 +26,7 @@ import org.json4s.jackson.JsonMethods._
 
 import org.apache.spark.annotation.InterfaceStability
 import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.util.Utils
 
@@ -110,6 +111,8 @@ abstract class DataType extends AbstractDataType {
 @InterfaceStability.Stable
 object DataType {
 
+  def fromDDL(ddl: String): DataType = CatalystSqlParser.parseDataType(ddl)
+
   def fromJson(json: String): DataType = parseDataType(parse(json))
 
   private val nonDecimalNameToType = {

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

@@ -3369,7 +3369,12 @@ object functions {
     val dataType = try {
       DataType.fromJson(schema)
     } catch {
-      case NonFatal(_) => StructType.fromDDL(schema)
+      case NonFatal(_) =>
+        try {
+          StructType.fromDDL(schema)
+        } catch {
+          case NonFatal(_) => DataType.fromDDL(schema)
+        }
     }
     from_json(e, dataType, options)
   }

sql/core/src/test/resources/sql-tests/inputs/json-functions.sql

@@ -31,3 +31,7 @@ CREATE TEMPORARY VIEW jsonTable(jsonField, a) AS SELECT * FROM VALUES ('{"a": 1,
 SELECT json_tuple(jsonField, 'b', CAST(NULL AS STRING), a) FROM jsonTable;
 -- Clean up
 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>');

sql/core/src/test/resources/sql-tests/results/json-functions.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 26
+-- Number of queries: 28
 
 
 -- !query 0
@@ -258,3 +258,19 @@ DROP VIEW IF EXISTS jsonTable
 struct<>
 -- !query 25 output
 
+
+
+-- !query 26
+select from_json('{"a":1, "b":2}', 'map<string, int>')
+-- !query 26 schema
+struct<entries:map<string,int>>
+-- !query 26 output
+{"a":1,"b":2}
+
+
+-- !query 27
+select from_json('{"a":1, "b":"2"}', 'struct<a:int,b:string>')
+-- !query 27 schema
+struct<jsontostructs({"a":1, "b":"2"}):struct<a:int,b:string>>
+-- !query 27 output
+{"a":1,"b":"2"}

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

@@ -354,8 +354,8 @@ class JsonFunctionsSuite extends QueryTest with SharedSQLContext {
 
   test("SPARK-24027: from_json - map<string, map<string, int>>") {
     val in = Seq("""{"a": {"b": 1}}""").toDS()
-    val schema = MapType(StringType, MapType(StringType, IntegerType))
-    val out = in.select(from_json($"value", schema))
+    val schema = "map<string, map<string, int>>"
+    val out = in.select(from_json($"value", schema, Map.empty[String, String]))
 
     checkAnswer(out, Row(Map("a" -> Map("b" -> 1))))
   }