SPARK-6245 [SQL] jsonRDD() of empty RDD results in exception

sql/core/src/main/scala/org/apache/spark/sql/json/JsonRDD.scala

@@ -48,7 +48,11 @@ private[sql] object JsonRDD extends Logging {
     require(samplingRatio > 0, s"samplingRatio ($samplingRatio) should be greater than 0")
     val schemaData = if (samplingRatio > 0.99) json else json.sample(false, samplingRatio, 1)
     val allKeys =
-      parseJson(schemaData, columnNameOfCorruptRecords).map(allKeysWithValueTypes).reduce(_ ++ _)
+      if (schemaData.isEmpty()) {
+        Set.empty[(String,DataType)]
+      } else {
+        parseJson(schemaData, columnNameOfCorruptRecords).map(allKeysWithValueTypes).reduce(_ ++ _)
+      }
     createSchema(allKeys)
   }
 

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

@@ -1033,4 +1033,11 @@ class JsonSuite extends QueryTest {
     assert(!logicalRelation2.sameResult(logicalRelation3),
       s"$logicalRelation2 and $logicalRelation3 should be considered not having the same result.")
   }
+
+  test("SPARK-6245 JsonRDD.inferSchema on empty RDD") {
+    // This is really a test that it doesn't throw an exception
+    val emptySchema = JsonRDD.inferSchema(empty, 1.0, "")
+    assert(StructType(Seq()) === emptySchema)
+  }
+
 }

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

@@ -185,4 +185,7 @@ object TestJsonData {
       """{"a":{, b:3}""" ::
       """{"b":"str_b_4", "a":"str_a_4", "c":"str_c_4"}""" ::
       """]""" :: Nil)
+
+  val empty =
+    TestSQLContext.sparkContext.parallelize(Seq[String]())
 }