Addressing Hyukjin Kwon's review comments

Author: MaxGekk

python/pyspark/sql/readwriter.py

@@ -238,8 +238,9 @@ def json(self, path, schema=None, primitivesAsString=None, prefersDecimal=None,
         :param allowUnquotedControlChars: allows JSON Strings to contain unquoted control
                                           characters (ASCII characters with value less than 32,
                                           including tab and line feed characters) or not.
-        :param encoding: standard encoding (charset) name, for example UTF-8, UTF-16LE and UTF-32BE.
-                         If None is set, the encoding of input JSON will be detected automatically
+        :param encoding: allows to forcibly set one of standard basic or extended encoding for
+                         the JSON files. For example UTF-16BE, UTF-32LE. If None is set,
+                         the encoding of input JSON will be detected automatically
                          when the multiLine option is set to ``true``.
         :param lineSep: defines the line separator that should be used for parsing. If None is
                         set, it covers all ``\\r``, ``\\r\\n`` and ``\\n``.

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

@@ -47,7 +47,17 @@ private[sql] object CreateJacksonParser extends Serializable {
     jsonFactory.createParser(record.getBytes, 0, record.getLength)
   }
 
-  def getStreamDecoder(enc: String, in: Array[Byte], length: Int): StreamDecoder = {
+  // Jackson parsers can be ranked according to their performance:
+  // 1. Array based with actual encoding UTF-8 in the array. This is the fastest parser
+  //    but it doesn't allow to set encoding explicitly. Actual encoding is detected automatically
+  //    by checking leading bytes of the array.
+  // 2. InputStream based with actual encoding UTF-8 in the stream. Encoding is detected
+  //    automatically by analyzing first bytes of the input stream.
+  // 3. Reader based parser. This is the slowest parser used here but it allows to create
+  //    a reader with specific encoding.
+  // The method creates a reader for an array with given encoding and sets size of internal
+  // decoding buffer according to size of input array.
+  private def getStreamDecoder(enc: String, in: Array[Byte], length: Int): StreamDecoder = {
     val bais = new ByteArrayInputStream(in, 0, length)
     val byteChannel = Channels.newChannel(bais)
     val decodingBufferSize = Math.min(length, 8192)

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

@@ -113,9 +113,9 @@ private[sql] class JSONOptions(
         s"""The ${enc} encoding must not be included in the blacklist when multiLine is disabled:
            | ${blacklist.mkString(", ")}""".stripMargin)
 
-      val forcingLineSep = !(multiLine == false &&
+      val isLineSepRequired = !(multiLine == false &&
         Charset.forName(enc) != StandardCharsets.UTF_8 && lineSeparator.isEmpty)
-      require(forcingLineSep, s"The lineSep option must be specified for the $enc encoding")
+      require(isLineSepRequired, s"The lineSep option must be specified for the $enc encoding")
 
       enc
   }

…c/test/resources/json-tests/utf16LE.json …rc/test/resources/test-data/utf16LE.jsonsql/core/src/test/resources/json-tests/utf16LE.json renamed to sql/core/src/test/resources/test-data/utf16LE.json sql/core/src/test/resources/json-tests/utf16LE.json renamed to sql/core/src/test/resources/test-data/utf16LE.json

@@ -24,7 +24,7 @@ import org.apache.spark.sql.types.{LongType, StringType, StructType}
 import org.apache.spark.util.{Benchmark, Utils}
 
 /**
- * Benchmark to measure JSON read/write performance.
+ * The benchmarks aims to measure performance of JSON parsing when encoding is set and isn't.
  * To run this:
  *  spark-submit --class <this class> --jars <spark sql test jar>
  */
@@ -49,9 +49,9 @@ object JSONBenchmarks {
     val benchmark = new Benchmark("JSON schema inferring", rowsNum)
 
     withTempPath { path =>
-      // scalastyle:off
+      // scalastyle:off println
       benchmark.out.println("Preparing data for benchmarking ...")
-      // scalastyle:on
+      // scalastyle:on println
 
       spark.sparkContext.range(0, rowsNum, 1)
         .map(_ => "a")
@@ -86,9 +86,9 @@ object JSONBenchmarks {
     val benchmark = new Benchmark("JSON per-line parsing", rowsNum)
 
     withTempPath { path =>
-      // scalastyle:off
+      // scalastyle:off println
       benchmark.out.println("Preparing data for benchmarking ...")
-      // scalastyle:on
+      // scalastyle:on println
 
       spark.sparkContext.range(0, rowsNum, 1)
         .map(_ => "a")
@@ -127,9 +127,9 @@ object JSONBenchmarks {
     val benchmark = new Benchmark("JSON parsing of wide lines", rowsNum)
 
     withTempPath { path =>
-      // scalastyle:off
+      // scalastyle:off println
       benchmark.out.println("Preparing data for benchmarking ...")
-      // scalastyle:on
+      // scalastyle:on println
 
       spark.sparkContext.range(0, rowsNum, 1)
         .map { i =>

…t/resources/json-tests/utf16WithBOM.json …st/resources/test-data/utf16WithBOM.jsonsql/core/src/test/resources/json-tests/utf16WithBOM.json renamed to sql/core/src/test/resources/test-data/utf16WithBOM.json sql/core/src/test/resources/json-tests/utf16WithBOM.json renamed to sql/core/src/test/resources/test-data/utf16WithBOM.json

@@ -48,6 +48,10 @@ class TestFileFilter extends PathFilter {
 class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
   import testImplicits._
 
+  def testFile(fileName: String): String = {
+    Thread.currentThread().getContextClassLoader.getResource(fileName).toString
+  }
+
   test("Type promotion") {
     def checkTypePromotion(expected: Any, actual: Any) {
       assert(expected.getClass == actual.getClass,
@@ -2168,12 +2172,8 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
     assert(sampled.count() == ds.count())
   }
 
-  def testFile(fileName: String): String = {
-    Thread.currentThread().getContextClassLoader.getResource(fileName).toString
-  }
-
   test("SPARK-23723: json in UTF-16 with BOM") {
-    val fileName = "json-tests/utf16WithBOM.json"
+    val fileName = "test-data/utf16WithBOM.json"
     val schema = new StructType().add("firstName", StringType).add("lastName", StringType)
     val jsonDF = spark.read.schema(schema)
       .option("multiline", "true")
@@ -2184,7 +2184,7 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
   }
 
   test("SPARK-23723: multi-line json in UTF-32BE with BOM") {
-    val fileName = "json-tests/utf32BEWithBOM.json"
+    val fileName = "test-data/utf32BEWithBOM.json"
     val schema = new StructType().add("firstName", StringType).add("lastName", StringType)
     val jsonDF = spark.read.schema(schema)
       .option("multiline", "true")
@@ -2194,7 +2194,7 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
   }
 
   test("SPARK-23723: Use user's encoding in reading of multi-line json in UTF-16LE") {
-    val fileName = "json-tests/utf16LE.json"
+    val fileName = "test-data/utf16LE.json"
     val schema = new StructType().add("firstName", StringType).add("lastName", StringType)
     val jsonDF = spark.read.schema(schema)
       .option("multiline", "true")
@@ -2209,15 +2209,15 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
     val exception = intercept[UnsupportedCharsetException] {
       spark.read
         .options(Map("encoding" -> invalidCharset, "lineSep" -> "\n"))
-        .json(testFile("json-tests/utf16LE.json"))
+        .json(testFile("test-data/utf16LE.json"))
         .count()
     }
 
     assert(exception.getMessage.contains(invalidCharset))
   }
 
   test("SPARK-23723: checking that the encoding option is case agnostic") {
-    val fileName = "json-tests/utf16LE.json"
+    val fileName = "test-data/utf16LE.json"
     val schema = new StructType().add("firstName", StringType).add("lastName", StringType)
     val jsonDF = spark.read.schema(schema)
       .option("multiline", "true")
@@ -2229,7 +2229,7 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
 
 
   test("SPARK-23723: specified encoding is not matched to actual encoding") {
-    val fileName = "json-tests/utf16LE.json"
+    val fileName = "test-data/utf16LE.json"
     val schema = new StructType().add("firstName", StringType).add("lastName", StringType)
     val exception = intercept[SparkException] {
       spark.read.schema(schema)
@@ -2244,17 +2244,15 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
     assert(errMsg.contains("Malformed records are detected in record parsing"))
   }
 
-  def checkEncoding(
-      expectedEncoding: String,
-      pathToJsonFiles: String,
+  def checkEncoding(expectedEncoding: String, pathToJsonFiles: String,
       expectedContent: String): Unit = {
     val jsonFiles = new File(pathToJsonFiles)
       .listFiles()
       .filter(_.isFile)
       .filter(_.getName.endsWith("json"))
     val actualContent = jsonFiles.map { file =>
       new String(Files.readAllBytes(file.toPath), expectedEncoding)
-    }.mkString.trim.replaceAll(" ", "")
+    }.mkString.trim
 
     assert(actualContent == expectedContent)
   }
@@ -2265,8 +2263,7 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
       val df = spark.createDataset(Seq(("Dog", 42)))
       df.write
         .options(Map("encoding" -> encoding, "lineSep" -> "\n"))
-        .format("json").mode("overwrite")
-        .save(path.getCanonicalPath)
+        .json(path.getCanonicalPath)
 
       checkEncoding(
         expectedEncoding = encoding,
@@ -2278,9 +2275,7 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
   test("SPARK-23723: save json in default encoding - UTF-8") {
     withTempPath { path =>
       val df = spark.createDataset(Seq(("Dog", 42)))
-      df.write
-        .format("json").mode("overwrite")
-        .save(path.getCanonicalPath)
+      df.write.json(path.getCanonicalPath)
 
       checkEncoding(
         expectedEncoding = "UTF-8",
@@ -2296,24 +2291,19 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
         val df = spark.createDataset(Seq((0)))
         df.write
           .options(Map("encoding" -> encoding, "lineSep" -> "\n"))
-          .format("json").mode("overwrite")
-          .save(path.getCanonicalPath)
+          .json(path.getCanonicalPath)
       }
     }
 
     assert(exception.getMessage == encoding)
   }
 
-  test("SPARK-23723: read written json in UTF-16LE") {
+  test("SPARK-23723: read back json in UTF-16LE") {
     val options = Map("encoding" -> "UTF-16LE", "lineSep" -> "\n")
     withTempPath { path =>
-      val ds = spark.createDataset(Seq(
-        ("a", 1), ("b", 2), ("c", 3))
-      ).repartition(2)
-      ds.write
-        .options(options)
-        .format("json").mode("overwrite")
-        .save(path.getCanonicalPath)
+      val ds = spark.createDataset(Seq(("a", 1), ("b", 2), ("c", 3))).repartition(2)
+      ds.write.options(options).json(path.getCanonicalPath)
+
       val readBack = spark
         .read
         .options(options)
@@ -2325,20 +2315,12 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
 
   def checkReadJson(lineSep: String, encoding: String, inferSchema: Boolean, id: Int): Unit = {
     test(s"SPARK-23724: checks reading json in ${encoding} #${id}") {
-      val lineSepInBytes = {
-        if (lineSep.startsWith("x")) {
-          lineSep.replaceAll("[^0-9A-Fa-f]", "")
-            .sliding(2, 2).toArray.map(Integer.parseInt(_, 16).toByte)
-        } else {
-          lineSep.getBytes(encoding)
-        }
-      }
       val schema = new StructType().add("f1", StringType).add("f2", IntegerType)
       withTempPath { path =>
         val records = List(("a", 1), ("b", 2))
         val data = records
           .map(rec => s"""{"f1":"${rec._1}", "f2":${rec._2}}""".getBytes(encoding))
-          .reduce((a1, a2) => a1 ++ lineSepInBytes ++ a2)
+          .reduce((a1, a2) => a1 ++ lineSep.getBytes(encoding) ++ a2)
         val os = new FileOutputStream(path)
         os.write(data)
         os.close()
@@ -2372,15 +2354,17 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
     (11, "\u000a\u000d", "UTF-8", true),
     (12, "===", "US-ASCII", false),
     (13, "$^+", "utf-32le", true)
-  ).foreach{case (i, d, c, s) => checkReadJson(d, c, s, i)}
+  ).foreach {
+    case (testNum, sep, encoding, inferSchema) => checkReadJson(sep, encoding, inferSchema, testNum)
+  }
   // scalastyle:on nonascii
 
   test("SPARK-23724: lineSep should be set if encoding if different from UTF-8") {
     val encoding = "UTF-16LE"
     val exception = intercept[IllegalArgumentException] {
       spark.read
         .options(Map("encoding" -> encoding))
-        .json(testFile("json-tests/utf16LE.json"))
+        .json(testFile("test-data/utf16LE.json"))
         .count()
     }
 
@@ -2390,36 +2374,38 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
 
   private val badJson = "\u0000\u0000\u0000A\u0001AAA"
 
-  test("SPARK-23094: invalid json with leading nulls - from file (multiLine=true)") {
-    withTempDir { tempDir =>
+  test("SPARK-23094: permissively read JSON file with leading nulls when multiLine is enabled") {
+    withTempPath { tempDir =>
       val path = tempDir.getAbsolutePath
-      Seq(badJson + """{"a":1}""").toDS().write.mode("overwrite").text(path)
+      Seq(badJson + """{"a":1}""").toDS().write.text(path)
       val expected = s"""${badJson}{"a":1}\n"""
       val schema = new StructType().add("a", IntegerType).add("_corrupt_record", StringType)
       val df = spark.read.format("json")
+        .option("mode", "PERMISSIVE")
         .option("multiLine", true)
         .option("encoding", "UTF-8")
         .schema(schema).load(path)
       checkAnswer(df, Row(null, expected))
     }
   }
 
-  test("SPARK-23094: invalid json with leading nulls - from file (multiLine=false)") {
-    withTempDir { tempDir =>
+  test("SPARK-23094: permissively read JSON file with leading nulls when multiLine is disabled") {
+    withTempPath { tempDir =>
       val path = tempDir.getAbsolutePath
-      Seq(badJson, """{"a":1}""").toDS().write.mode("overwrite").text(path)
+      Seq(badJson, """{"a":1}""").toDS().write.text(path)
       val schema = new StructType().add("a", IntegerType).add("_corrupt_record", StringType)
       val df = spark.read.format("json")
+        .option("mode", "PERMISSIVE")
         .option("multiLine", false)
         .option("encoding", "UTF-8")
         .schema(schema).load(path)
       checkAnswer(df, Seq(Row(1, null), Row(null, badJson)))
     }
   }
 
-  test("SPARK-23094: invalid json with leading nulls - from dataset") {
+  test("SPARK-23094: permissively parse a dataset contains JSON with leading nulls") {
     checkAnswer(
-      spark.read.option("encoding", "UTF-8").json(Seq(badJson).toDS()),
+      spark.read.option("mode", "PERMISSIVE").option("encoding", "UTF-8").json(Seq(badJson).toDS()),
       Row(badJson))
   }
 }