[SPARK-2060][SQL] Querying JSON Datasets with SQL and DSL in Spark SQL

project/SparkBuild.scala

@@ -486,9 +486,23 @@ object SparkBuild extends Build {
   def sqlCoreSettings = sharedSettings ++ Seq(
     name := "spark-sql",
     libraryDependencies ++= Seq(
-      "com.twitter" % "parquet-column" % parquetVersion,
-      "com.twitter" % "parquet-hadoop" % parquetVersion
-    )
+      "com.twitter"                  % "parquet-column"             % parquetVersion,
+      "com.twitter"                  % "parquet-hadoop"             % parquetVersion,
+      "com.fasterxml.jackson.core"   % "jackson-core"               % "2.3.2"
+    ),
+    initialCommands in console :=
+      """
+        |import org.apache.spark.sql.catalyst.analysis._
+        |import org.apache.spark.sql.catalyst.dsl._
+        |import org.apache.spark.sql.catalyst.errors._
+        |import org.apache.spark.sql.catalyst.expressions._
+        |import org.apache.spark.sql.catalyst.plans.logical._
+        |import org.apache.spark.sql.catalyst.rules._
+        |import org.apache.spark.sql.catalyst.types._
+        |import org.apache.spark.sql.catalyst.util._
+        |import org.apache.spark.sql.execution
+        |import org.apache.spark.sql.test.TestSQLContext._
+        |import org.apache.spark.sql.parquet.ParquetTestData""".stripMargin
   )
 
   // Since we don't include hive in the main assembly this project also acts as an alternative

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/HiveTypeCoercion.scala

@@ -108,6 +108,9 @@ trait HiveTypeCoercion {
    *
    * Additionally, all types when UNION-ed with strings will be promoted to strings.
    * Other string conversions are handled by PromoteStrings.
+   *
+   * A widening conversion of a value with IntegerType and LongType to FloatType,
+   * or of a value with LongType to DoubleType, may result in loss of precision.
    */
   object WidenTypes extends Rule[LogicalPlan] {
     // See https://cwiki.apache.org/confluence/display/Hive/LanguageManual+Types.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala

@@ -19,6 +19,7 @@ package org.apache.spark.sql.catalyst.plans
 
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
 import org.apache.spark.sql.catalyst.trees.TreeNode
+import org.apache.spark.sql.catalyst.plans
 
 abstract class QueryPlan[PlanType <: TreeNode[PlanType]] extends TreeNode[PlanType] {
   self: PlanType with Product =>
@@ -123,4 +124,8 @@ abstract class QueryPlan[PlanType <: TreeNode[PlanType]] extends TreeNode[PlanTy
       case other => Nil
     }.toSeq
   }
+
+  def printSchema(): Unit = {
+    println(plans.generateSchemaTreeString(output))
+  }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/package.scala

@@ -17,8 +17,56 @@
 
 package org.apache.spark.sql.catalyst
 
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.types.{StructField, DataType, ArrayType, StructType}
+
 /**
  * A a collection of common abstractions for query plans as well as
  * a base logical plan representation.
  */
-package object plans
+package object plans {
+  def generateSchemaTreeString(schema: Seq[Attribute]): String = {
+    val builder = new StringBuilder
+    builder.append("root\n")
+    val prefix = " |"
+    schema.foreach {
+      attribute => {
+        val name = attribute.name
+        val dataType = attribute.dataType
+        dataType match {
+          case fields: StructType =>
+            builder.append(s"$prefix-- $name: $StructType\n")
+            generateSchemaTreeString(fields, s"$prefix    |", builder)
+          case ArrayType(fields: StructType) =>
+            builder.append(s"$prefix-- $name: $ArrayType[$StructType]\n")
+            generateSchemaTreeString(fields, s"$prefix    |", builder)
+          case ArrayType(elementType: DataType) =>
+            builder.append(s"$prefix-- $name: $ArrayType[$elementType]\n")
+          case _ => builder.append(s"$prefix-- $name: $dataType\n")
+        }
+      }
+    }
+
+    builder.toString()
+  }
+
+  def generateSchemaTreeString(
+      schema: StructType,
+      prefix: String,
+      builder: StringBuilder): StringBuilder = {
+    schema.fields.foreach {
+      case StructField(name, fields: StructType, _) =>
+        builder.append(s"$prefix-- $name: $StructType\n")
+        generateSchemaTreeString(fields, s"$prefix    |", builder)
+      case StructField(name, ArrayType(fields: StructType), _) =>
+        builder.append(s"$prefix-- $name: $ArrayType[$StructType]\n")
+        generateSchemaTreeString(fields, s"$prefix    |", builder)
+      case StructField(name, ArrayType(elementType: DataType), _) =>
+        builder.append(s"$prefix-- $name: $ArrayType[$elementType]\n")
+      case StructField(name, fieldType: DataType, _) =>
+        builder.append(s"$prefix-- $name: $fieldType\n")
+    }
+
+    builder
+  }
+}

sql/core/pom.xml

@@ -53,6 +53,11 @@
       <artifactId>parquet-hadoop</artifactId>
       <version>${parquet.version}</version>
     </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-core</artifactId>
+      <version>2.3.2</version>
+    </dependency>
     <dependency>
       <groupId>org.scalatest</groupId>
       <artifactId>scalatest_${scala.binary.version}</artifactId>

sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala

@@ -40,6 +40,7 @@ import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.SparkStrategies
 
 import org.apache.spark.sql.parquet.ParquetRelation
+import org.apache.spark.sql.json._
 
 /**
  * :: AlphaComponent ::
@@ -97,6 +98,41 @@ class SQLContext(@transient val sparkContext: SparkContext)
   def parquetFile(path: String): SchemaRDD =
     new SchemaRDD(this, parquet.ParquetRelation(path))
 
+  /**
+   * Loads a JSON file, returning the result as a [[SchemaRDD]].
+   * Right now, we only do eager schema resolution.
+   */
+  def jsonFile(
+      path: String,
+      mode: SchemaResolutionMode = EAGER_SCHEMA_RESOLUTION): SchemaRDD = {
+    logger.info(s"Loads a JSON file $path.")
+    val json = sparkContext.textFile(path)
+    jsonRDD(json, mode)
+  }
+
+  /**
+   * Loads a RDD[String] storing JSON objects (one object per record),
+   * returning the result as a [[SchemaRDD]].
+   * Right now, we only do eager schema resolution.
+   */
+  def jsonRDD(
+      json: RDD[String],
+      mode: SchemaResolutionMode = EAGER_SCHEMA_RESOLUTION): SchemaRDD = {
+    mode match {
+      case EAGER_SCHEMA_RESOLUTION =>
+        logger.info(s"Eagerly resolve the schema without sampling.")
+        val logicalPlan = JsonTable.inferSchema(json)
+        logicalPlanToSparkQuery(logicalPlan)
+      case EAGER_SCHEMA_RESOLUTION_WITH_SAMPLING(fraction) =>
+        logger.info(s"Eagerly resolve the schema with sampling " +
+          s"(sampling fraction: $fraction).")
+        val logicalPlan = JsonTable.inferSchema(json, Some(fraction))
+        logicalPlanToSparkQuery(logicalPlan)
+      case LAZY_SCHEMA_RESOLUTION =>
+        throw new UnsupportedOperationException("Lazy schema resolution has not been implemented.")
+    }
+  }
+
   /**
    * :: Experimental ::
    * Creates an empty parquet file with the schema of class `A`, which can be registered as a table.

sql/core/src/main/scala/org/apache/spark/sql/SchemaRDDLike.scala

@@ -111,4 +111,10 @@ private[sql] trait SchemaRDDLike {
   @Experimental
   def saveAsTable(tableName: String): Unit =
     sqlContext.executePlan(InsertIntoCreatedTable(None, tableName, logicalPlan)).toRdd
+
+
+  /**
+   * Print the schema of this SchemaRDD.
+   */
+  def printSchema = queryExecution.analyzed.printSchema()
 }