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 04a4eb0ffc03..0c14be75203e 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
@@ -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._
@@ -615,7 +616,7 @@ case class JsonToStructs(
     }
   }
 
-  override def inputTypes: Seq[AbstractDataType] = StringType :: Nil
+  override def inputTypes: Seq[AbstractDataType] = Seq(StringType, StringType)
 
   override def sql: String = schema match {
     case _: MapType => "entries"
@@ -747,8 +748,13 @@ 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) =>
+      try {
+        DataType.fromJson(s.toString)
+      } catch {
+        case NonFatal(_) => StructType.fromDDL(s.toString)
+      }
     case e => throw new AnalysisException(s"Expected a string literal instead of $e")
   }
 
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 fea069eac4d4..7d7d55f09ab2 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
@@ -25,6 +25,10 @@ select from_json('{"a":1}', 'a InvalidType');
 select from_json('{"a":1}', 'a INT', named_struct('mode', 'PERMISSIVE'));
 select from_json('{"a":1}', 'a INT', map('mode', 1));
 select from_json();
+-- from_json - schema in json format
+select from_json('{"a":1}', '{"type":"struct","fields":[{"name":"a","type":"integer", "nullable":true}]}');
+select from_json('{"a":1}', '{"type":"map", "keyType":"string", "valueType":"integer","valueContainsNull":false}');
+
 -- json_tuple
 SELECT json_tuple('{"a" : 1, "b" : 2}', CAST(NULL AS STRING), 'b', CAST(NULL AS STRING), 'a');
 CREATE TEMPORARY VIEW jsonTable(jsonField, a) AS SELECT * FROM VALUES ('{"a": 1, "b": 2}', 'a');
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 14a69128ffb4..76ddb38774f3 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: 26
+-- Number of queries: 28
 
 
 -- !query 0
@@ -229,32 +229,48 @@ Invalid number of arguments for function from_json. Expected: one of 2 and 3; Fo
 
 
 -- !query 22
-SELECT json_tuple('{"a" : 1, "b" : 2}', CAST(NULL AS STRING), 'b', CAST(NULL AS STRING), 'a')
+select from_json('{"a":1}', '{"type":"struct","fields":[{"name":"a","type":"integer", "nullable":true}]}')
 -- !query 22 schema
-struct<c0:string,c1:string,c2:string,c3:string>
+struct<jsontostructs({"a":1}):struct<a:int>>
 -- !query 22 output
-NULL	2	NULL	1
+{"a":1}
 
 
 -- !query 23
-CREATE TEMPORARY VIEW jsonTable(jsonField, a) AS SELECT * FROM VALUES ('{"a": 1, "b": 2}', 'a')
+select from_json('{"a":1}', '{"type":"map", "keyType":"string", "valueType":"integer","valueContainsNull":false}')
 -- !query 23 schema
-struct<>
+struct<entries:map<string,int>>
 -- !query 23 output
-
+{"a":1}
 
 
 -- !query 24
-SELECT json_tuple(jsonField, 'b', CAST(NULL AS STRING), a) FROM jsonTable
+SELECT json_tuple('{"a" : 1, "b" : 2}', CAST(NULL AS STRING), 'b', CAST(NULL AS STRING), 'a')
 -- !query 24 schema
-struct<c0:string,c1:string,c2:string>
+struct<c0:string,c1:string,c2:string,c3:string>
 -- !query 24 output
-2	NULL	1
+NULL	2	NULL	1
 
 
 -- !query 25
-DROP VIEW IF EXISTS jsonTable
+CREATE TEMPORARY VIEW jsonTable(jsonField, a) AS SELECT * FROM VALUES ('{"a": 1, "b": 2}', 'a')
 -- !query 25 schema
 struct<>
 -- !query 25 output
 
+
+
+-- !query 26
+SELECT json_tuple(jsonField, 'b', CAST(NULL AS STRING), a) FROM jsonTable
+-- !query 26 schema
+struct<c0:string,c1:string,c2:string>
+-- !query 26 output
+2	NULL	1
+
+
+-- !query 27
+DROP VIEW IF EXISTS jsonTable
+-- !query 27 schema
+struct<>
+-- !query 27 output
+