Use user's charset or autodetect it if the charset is not specified

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

@@ -39,8 +39,13 @@ private[sql] object CreateJacksonParser extends Serializable {
     jsonFactory.createParser(new InputStreamReader(bain, "UTF-8"))
   }
 
-  def text(jsonFactory: JsonFactory, record: Text): JsonParser = {
-    jsonFactory.createParser(record.getBytes, 0, record.getLength)
+  def text(charset: Option[String])(jsonFactory: JsonFactory, record: Text): JsonParser = {
+    charset.map {cs =>
+      val bain = new ByteArrayInputStream(record.getBytes, 0, record.getLength)
+      jsonFactory.createParser(new InputStreamReader(bain, cs))
+    }.getOrElse {
+      jsonFactory.createParser(record.getBytes, 0, record.getLength)
+    }
   }
 
   def inputStream(jsonFactory: JsonFactory, record: InputStream): JsonParser = {

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

@@ -85,6 +85,9 @@ private[sql] class JSONOptions(
 
   val multiLine = parameters.get("multiLine").map(_.toBoolean).getOrElse(false)
 
+  /** Standard charset name. For example UTF-8, UTF-16 and UTF-32 */
+  val charset = parameters.get("charset")
+
   /** Sets config options on a Jackson [[JsonFactory]]. */
   def setJacksonOptions(factory: JsonFactory): Unit = {
     factory.configure(JsonParser.Feature.ALLOW_COMMENTS, allowComments)

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

@@ -122,8 +122,10 @@ object TextInputJsonDataSource extends JsonDataSource {
       schema: StructType): Iterator[InternalRow] = {
     val linesReader = new HadoopFileLinesReader(file, conf)
     Option(TaskContext.get()).foreach(_.addTaskCompletionListener(_ => linesReader.close()))
+    val charset = parser.options.charset
+
     val safeParser = new FailureSafeParser[Text](
-      input => parser.parse(input, CreateJacksonParser.text, textToUTF8String),
+      input => parser.parse(input, CreateJacksonParser.text(charset), textToUTF8String),
       parser.options.parseMode,
       schema,
       parser.options.columnNameOfCorruptRecord)