[SPARK-12057] [SQL] Prevent failure on corrupt JSON records

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

@@ -266,11 +266,7 @@ object JacksonParser {
                 } else {
                   array.toArray[InternalRow](schema)
                 }
-              case _ =>
-                sys.error(
-                  s"Failed to parse record $record. Please make sure that each line of " +
-                    "the file (or each string in the RDD) is a valid JSON object or " +
-                    "an array of JSON objects.")
+              case _ => failedRecord(record)
             }
           }
         } catch {

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

@@ -1427,4 +1427,31 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData {
       }
     }
   }
+
+  test("SPARK-12057 additional corrupt records do not throw exceptions") {
+    // Test if we can query corrupt records.
+    withSQLConf(SQLConf.COLUMN_NAME_OF_CORRUPT_RECORD.key -> "_unparsed") {
+      withTempTable("jsonTable") {
+        val schema = StructType(
+          StructField("_unparsed", StringType, true) ::
+            StructField("dummy", StringType, true) :: Nil)
+        val jsonDF = sqlContext.read.schema(schema).json(additionalCorruptRecords)
+        jsonDF.registerTempTable("jsonTable")
+
+        // In HiveContext, backticks should be used to access columns starting with a underscore.
+        checkAnswer(
+          sql(
+            """
+              |SELECT dummy, _unparsed
+              |FROM jsonTable
+            """.stripMargin),
+          Row("test", null) ::
+          Row(null, """42""") ::
+          Row(null, """     ","ian":"test"}""") :: Nil
+        )
+
+      }
+    }
+  }
+
 }

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

@@ -188,6 +188,12 @@ private[json] trait TestJsonData {
       """{"b":"str_b_4", "a":"str_a_4", "c":"str_c_4"}""" ::
       """]""" :: Nil)
 
+  def additionalCorruptRecords: RDD[String] =
+    sqlContext.sparkContext.parallelize(
+      """{"dummy":"test"}""" ::
+      """42""" ::
+      """     ","ian":"test"}""" :: Nil)
+
   def emptyRecords: RDD[String] =
     sqlContext.sparkContext.parallelize(
       """{""" ::