[SPARK-11753][SQL][test-hadoop2.2] Make allowNonNumericNumbers option work

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

@@ -31,7 +31,7 @@ case class JSONOptions(
     allowUnquotedFieldNames: Boolean = false,
     allowSingleQuotes: Boolean = true,
     allowNumericLeadingZeros: Boolean = false,
-    allowNonNumericNumbers: Boolean = false) {
+    allowNonNumericNumbers: Boolean = true) {
 
   /** Sets config options on a Jackson [[JsonFactory]]. */
   def setJacksonOptions(factory: JsonFactory): Unit = {

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

@@ -19,7 +19,7 @@ package org.apache.spark.sql.execution.datasources.json
 
 import java.io.CharArrayWriter
 
-import com.fasterxml.jackson.core.JsonFactory
+import com.fasterxml.jackson.core.{JsonGenerator, JsonFactory}
 import com.google.common.base.Objects
 import org.apache.hadoop.fs.{FileStatus, Path}
 import org.apache.hadoop.io.{LongWritable, NullWritable, Text}
@@ -161,12 +161,13 @@ private[sql] class JSONRelation(
   }
 
   override def prepareJobForWrite(job: Job): OutputWriterFactory = {
+    val quoteNonNumeric = !options.allowNonNumericNumbers
     new OutputWriterFactory {
       override def newInstance(
           path: String,
           dataSchema: StructType,
           context: TaskAttemptContext): OutputWriter = {
-        new JsonOutputWriter(path, dataSchema, context)
+        new JsonOutputWriter(path, dataSchema, context, quoteNonNumeric)
       }
     }
   }
@@ -175,12 +176,15 @@ private[sql] class JSONRelation(
 private[json] class JsonOutputWriter(
     path: String,
     dataSchema: StructType,
-    context: TaskAttemptContext)
+    context: TaskAttemptContext,
+    quoteNonNumeric: Boolean)
   extends OutputWriter with SparkHadoopMapRedUtil with Logging {
 
   private[this] val writer = new CharArrayWriter()
   // create the Generator without separator inserted between 2 records
-  private[this] val gen = new JsonFactory().createGenerator(writer).setRootValueSeparator(null)
+  private[this] val factory = new JsonFactory()
+  factory.configure(JsonGenerator.Feature.QUOTE_NON_NUMERIC_NUMBERS, quoteNonNumeric)
+  private[this] val gen = factory.createGenerator(writer).setRootValueSeparator(null)
   private[this] val result = new Text()
 
   private val recordWriter: RecordWriter[NullWritable, Text] = {

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

@@ -100,35 +100,13 @@ object JacksonParser {
         parser.getFloatValue
 
       case (VALUE_STRING, FloatType) =>
-        // Special case handling for NaN and Infinity.
-        val value = parser.getText
-        val lowerCaseValue = value.toLowerCase()
-        if (lowerCaseValue.equals("nan") ||
-          lowerCaseValue.equals("infinity") ||
-          lowerCaseValue.equals("-infinity") ||
-          lowerCaseValue.equals("inf") ||
-          lowerCaseValue.equals("-inf")) {
-          value.toFloat
-        } else {
-          sys.error(s"Cannot parse $value as FloatType.")
-        }
+        parser.getFloatValue
 
       case (VALUE_NUMBER_INT | VALUE_NUMBER_FLOAT, DoubleType) =>
         parser.getDoubleValue
 
       case (VALUE_STRING, DoubleType) =>
-        // Special case handling for NaN and Infinity.
-        val value = parser.getText
-        val lowerCaseValue = value.toLowerCase()
-        if (lowerCaseValue.equals("nan") ||
-          lowerCaseValue.equals("infinity") ||
-          lowerCaseValue.equals("-infinity") ||
-          lowerCaseValue.equals("inf") ||
-          lowerCaseValue.equals("-inf")) {
-          value.toDouble
-        } else {
-          sys.error(s"Cannot parse $value as DoubleType.")
-        }
+        parser.getDoubleValue
 
       case (VALUE_NUMBER_INT | VALUE_NUMBER_FLOAT, dt: DecimalType) =>
         Decimal(parser.getDecimalValue, dt.precision, dt.scale)

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonParsingOptionsSuite.scala

@@ -93,22 +93,29 @@ class JsonParsingOptionsSuite extends QueryTest with SharedSQLContext {
     assert(df.first().getLong(0) == 18)
   }
 
-  // The following two tests are not really working - need to look into Jackson's
-  // JsonParser.Feature.ALLOW_NON_NUMERIC_NUMBERS.
-  ignore("allowNonNumericNumbers off") {
-    val str = """{"age": NaN}"""
-    val rdd = sqlContext.sparkContext.parallelize(Seq(str))
-    val df = sqlContext.read.json(rdd)
+  test("allowNonNumericNumbers off") {
+    val testCases: Seq[String] = Seq("""{"age": NaN}""", """{"age": Infinity}""",
+      """{"age": -Infinity}""")
 
-    assert(df.schema.head.name == "_corrupt_record")
+    testCases.foreach { str =>
+      val rdd = sqlContext.sparkContext.parallelize(Seq(str))
+      val df = sqlContext.read.option("allowNonNumericNumbers", "false").json(rdd)
+
+      assert(df.schema.head.name == "_corrupt_record")
+    }
   }
 
-  ignore("allowNonNumericNumbers on") {
-    val str = """{"age": NaN}"""
-    val rdd = sqlContext.sparkContext.parallelize(Seq(str))
-    val df = sqlContext.read.option("allowNonNumericNumbers", "true").json(rdd)
+  test("allowNonNumericNumbers on") {
+    val testCases: Seq[String] = Seq("""{"age": NaN}""", """{"age": Infinity}""",
+      """{"age": -Infinity}""")
+    val tests: Seq[Double => Boolean] = Seq(_.isNaN, _.isPosInfinity, _.isNegInfinity)
 
-    assert(df.schema.head.name == "age")
-    assert(df.first().getDouble(0).isNaN)
+    testCases.zipWithIndex.foreach { case (str, idx) =>
+      val rdd = sqlContext.sparkContext.parallelize(Seq(str))
+      val df = sqlContext.read.json(rdd)
+
+      assert(df.schema.head.name == "age")
+      assert(tests(idx)(df.first().getDouble(0)))
+    }
   }
 }