Merge pull request apache#41 from marmbrus/splitComponents

Break catalyst into 3 major components and move everything into org.apache.spark.sql

Author: rxin

.travis.yml

@@ -9,4 +9,4 @@
      - $HOME/.ivy2
      - $HOME/.sbt
  script:
-   - "GIT_AUTHOR_NAME=\"Michael Armbrust\" GIT_AUTHOR_EMAIL=\"[email protected]\" GIT_COMMITTER_NAME=\"Michael Armbrust\" GIT_COMMITTER_EMAIL=\"[email protected]\" sbt ++$TRAVIS_SCALA_VERSION 'set scalacOptions += \"-Xfatal-warnings\"' test:compile scalastyle test ghpages-push-site"
+   - "GIT_AUTHOR_NAME=\"Michael Armbrust\" GIT_AUTHOR_EMAIL=\"[email protected]\" GIT_COMMITTER_NAME=\"Michael Armbrust\" GIT_COMMITTER_EMAIL=\"[email protected]\" sbt ++$TRAVIS_SCALA_VERSION 'set scalacOptions += \"-Xfatal-warnings\"' test:compile test scalastyle"

README.md

@@ -61,30 +61,36 @@ Welcome to Scala version 2.10.3 (Java HotSpot(TM) 64-Bit Server VM, Java 1.7.0_4
 Type in expressions to have them evaluated.
 Type :help for more information.
 
-scala> val query = "SELECT * FROM (SELECT * FROM src) a".q
-query: catalyst.execution.TestShark.SharkSqlQuery =
+scala> scala> val query = sql("SELECT * FROM (SELECT * FROM src) a")
+query: org.apache.spark.sql.ExecutedQuery =
 SELECT * FROM (SELECT * FROM src) a
-== Logical Plan ==
-Project {key#0,value#1}
- Subquery a
-  Project {key#0,value#1}
-   MetastoreRelation default, src, None
+=== Query Plan ===
+Project [key#6:0.0,value#7:0.1]
+ HiveTableScan [key#6,value#7], (MetastoreRelation default, src, None), None
+```
 
-== Physical Plan ==
-Project {key#0,value#1}
- HiveTableScan {key#0,value#1}, (MetastoreRelation default, src, None)
+Query results are RDDs and can be operated as such.
+```
+scala> query.collect()
+res8: Array[org.apache.spark.sql.execution.Row] = Array([238,val_238], [86,val_86], [311,val_311]...
+```
+
+You can also build further queries on top of these RDDs using the query DSL.
+```
+scala> query.where('key === 100).toRdd.collect()
+res11: Array[org.apache.spark.sql.execution.Row] = Array([100,val_100], [100,val_100])
 ```
 
 From the console you can even write rules that transform query plans.  For example, the above query has redundant project operators that aren't doing anything.  This redundancy can be eliminated using the `transform` function that is available on all [`TreeNode`](http://databricks.github.io/catalyst/latest/api/#catalyst.trees.TreeNode) objects.
 ```scala
-scala> query.optimizedPlan
+scala> query.logicalPlan
 res1: catalyst.plans.logical.LogicalPlan = 
 Project {key#0,value#1}
  Project {key#0,value#1}
   MetastoreRelation default, src, None
 
 
-scala> res0.optimizedPlan transform {
+scala> query.logicalPlan transform {
      |   case Project(projectList, child) if projectList == child.output => child
      | }
 res2: catalyst.plans.logical.LogicalPlan = 

build.sbt

@@ -1,82 +1,52 @@
-import AssemblyKeys._ // put this at the top of the file
 
-name := "catalyst"
-
-organization := "com.databricks"
-
-version := "0.1-SNAPSHOT"
-
-scalaVersion := "2.10.3"
-
-scalacOptions ++= Seq("-deprecation", "-feature", "-unchecked")
-
-resolvers += "Local Maven Repository" at "file://"+Path.userHome.absolutePath+"/.m2/repository"
-
-// TODO: Remove when Spark 0.9.0 is released for real.
-resolvers += "SparkStaging" at "https://repository.apache.org/content/repositories/orgapachespark-1006/"
-
-libraryDependencies += "org.apache.spark" %% "spark-core" % "0.9.0-incubating"
-
-// Hive 0.10.0 relies on a weird version of jdo that is not published anywhere... Remove when we upgrade to 0.11.0
-libraryDependencies += "javax.jdo" % "jdo2-api" % "2.3-ec" from "http://www.datanucleus.org/downloads/maven2/javax/jdo/jdo2-api/2.3-ec/jdo2-api-2.3-ec.jar"
-
-libraryDependencies ++= Seq(
- "org.apache.hadoop" % "hadoop-client" % "1.0.4",
- "org.scalatest" %% "scalatest" % "1.9.1" % "test",
- //"net.hydromatic" % "optiq-core" % "0.4.16-SNAPSHOT",
- "org.apache.hive" % "hive-metastore" % "0.12.0",
- "org.apache.hive" % "hive-exec" % "0.12.0",
- "org.apache.hive" % "hive-serde" % "0.12.0",
-  "com.typesafe" %% "scalalogging-slf4j" % "1.0.1")
-
-org.scalastyle.sbt.ScalastylePlugin.Settings
-
-// Multiple queries rely on the TestShark singleton.  See comments there for more details.
-parallelExecution in Test := false
-
-resolvers ++= Seq(
-    // For Optiq
-    "Conjars Repository" at "http://conjars.org/repo/",
-    // For jdo-2 required by Hive < 0.12.0
-    "Datanucleus Repository" at "http://www.datanucleus.org/downloads/maven2")
-
-resolvers += "Databees" at "http://repository-databricks.forge.cloudbees.com/snapshot/"
-
-initialCommands in console := """
-import catalyst.analysis._
-import catalyst.dsl._
-import catalyst.errors._
-import catalyst.expressions._
-import catalyst.frontend._
-import catalyst.plans.logical._
-import catalyst.rules._
-import catalyst.types._
-import catalyst.util._
-import catalyst.execution.TestShark._"""
-
-site.settings
-
-ghpages.settings
-
-git.remoteRepo := "[email protected]:databricks/catalyst.git"
-
-site.settings
-
-site.includeScaladoc()
-
-assemblySettings
-
-test in assembly := {}
-
-mergeStrategy in assembly := {
-  case m if m.toLowerCase.endsWith("manifest.mf") => MergeStrategy.discard
-  case m if m.toLowerCase.matches("meta-inf.*\\.sf$") => MergeStrategy.discard
-  case "log4j.properties" => MergeStrategy.discard
-  case m if m.toLowerCase.startsWith("meta-inf/services/") => MergeStrategy.filterDistinctLines
-  case "reference.conf" => MergeStrategy.concat
-  case _ => MergeStrategy.first
-}
-
-scalacOptions in (Compile, doc)  <++= (baseDirectory) map {
-  bd => Seq("-sourcepath", bd.getAbsolutePath, "-doc-source-url","https://github.com/databricks/catalyst/blob/master/€{FILE_PATH}.scala")
-}
+lazy val catalyst = Project("catalyst", file("catalyst"), settings = catalystSettings)
+lazy val core = Project("core", file("core"), settings = coreSettings).dependsOn(catalyst)
+lazy val shark = Project("shark", file("shark"), settings = sharkSettings).dependsOn(core)
+
+def sharedSettings = Defaults.defaultSettings ++ Seq(
+  organization := "org.apache.spark.sql",
+  version := "0.1-SNAPSHOT",
+  scalaVersion := "2.10.3",
+  scalacOptions ++= Seq("-deprecation", "-feature", "-unchecked"),
+  // Common Dependencies.
+  libraryDependencies ++= Seq(
+    "org.scalatest" %% "scalatest" % "1.9.1" % "test",
+    "com.typesafe" %% "scalalogging-slf4j" % "1.0.1")
+) ++ org.scalastyle.sbt.ScalastylePlugin.Settings
+
+def catalystSettings = sharedSettings ++ Seq(
+  name := "catalyst",
+  // The mechanics of rewriting expression ids to compare trees in some test cases makes
+  // assumptions about the the expression ids being contiguious.  Running tests in parallel breaks
+  // this non-deterministically.  TODO: FIX THIS.
+  parallelExecution in Test := false
+)
+
+def coreSettings = sharedSettings ++ Seq(
+  name := "core",
+  libraryDependencies += "org.apache.spark" %% "spark-core" % "0.9.0-incubating"
+)
+
+def sharkSettings = sharedSettings ++ Seq(
+  name := "shark",
+  libraryDependencies ++= Seq(
+    "org.apache.hadoop" % "hadoop-client" % "1.0.4",
+    "org.apache.hive" % "hive-metastore" % "0.12.0",
+    "org.apache.hive" % "hive-exec" % "0.12.0",
+    "org.apache.hive" % "hive-serde" % "0.12.0"),
+  // Multiple queries rely on the TestShark singleton.  See comments there for more details.
+  parallelExecution in Test := false,
+  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.shark._
+      |import org.apache.spark.sql.shark.TestShark._""".stripMargin
+)

src/main/scala/catalyst/analysis/Analyzer.scala renamed to catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1,10 +1,10 @@
+package org.apache.spark.sql
 package catalyst
 package analysis
 
 import expressions._
 import plans.logical._
 import rules._
-import catalyst.execution.MetastoreRelation
 
 /**
  * A trivial [[Analyzer]] with an [[EmptyCatalog]] and [[EmptyFunctionRegistry]]. Used for testing
@@ -36,7 +36,6 @@ class Analyzer(catalog: Catalog, registry: FunctionRegistry, caseSensitive: Bool
       StarExpansion ::
       ResolveFunctions ::
       GlobalAggregates ::
-      PreInsertionCasts ::
       typeCoercionRules :_*)
   )
 
@@ -165,34 +164,4 @@ class Analyzer(catalog: Catalog, registry: FunctionRegistry, caseSensitive: Bool
     protected def containsStar(exprs: Seq[Expression]): Boolean =
       exprs.collect { case _: Star => true }.nonEmpty
   }
-
-  /**
-   * Casts input data to correct data types according to table definition before inserting into
-   * that table.
-   */
-  object PreInsertionCasts extends Rule[LogicalPlan] {
-    def apply(plan: LogicalPlan): LogicalPlan = plan.transform {
-      // Wait until children are resolved
-      case p: LogicalPlan if !p.childrenResolved => p
-
-      case p @ InsertIntoTable(table: MetastoreRelation, _, child) =>
-        val childOutputDataTypes = child.output.map(_.dataType)
-        // Only check attributes, not partitionKeys since they are always strings.
-        // TODO: Fully support inserting into partitioned tables.
-        val tableOutputDataTypes = table.attributes.map(_.dataType)
-
-        if (childOutputDataTypes == tableOutputDataTypes) {
-          p
-        } else {
-          // Only do the casting when child output data types differ from table output data types.
-          val castedChildOutput = child.output.zip(table.output).map {
-            case (input, table) if input.dataType != table.dataType =>
-              Alias(Cast(input, table.dataType), input.name)()
-            case (input, _) => input
-          }
-
-          p.copy(child = Project(castedChildOutput, child))
-        }
-    }
-  }
 }

src/main/scala/catalyst/analysis/Catalog.scala renamed to catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Catalog.scala

@@ -1,7 +1,9 @@
+package org.apache.spark.sql
 package catalyst
 package analysis
 
 import plans.logical.LogicalPlan
+import scala.collection.mutable
 
 /**
  * An interface for looking up relations by name.  Used by an [[Analyzer]].
@@ -13,6 +15,24 @@ trait Catalog {
     alias: Option[String] = None): LogicalPlan
 }
 
+trait OverrideCatalog extends Catalog {
+
+  // TODO: This doesn't work when the database changes...
+  val overrides = new mutable.HashMap[(Option[String],String), LogicalPlan]()
+
+  abstract override def lookupRelation(
+    databaseName: Option[String],
+    tableName: String,
+    alias: Option[String] = None): LogicalPlan = {
+
+    overrides.get((databaseName, tableName))
+      .getOrElse(super.lookupRelation(databaseName, tableName, alias))
+  }
+
+  def overrideTable(databaseName: Option[String], tableName: String, plan: LogicalPlan) =
+    overrides.put((databaseName, tableName), plan)
+}
+
 /**
  * A trivial catalog that returns an error when a relation is requested.  Used for testing when all
  * relations are already filled in and the analyser needs only to resolve attribute references.

src/main/scala/catalyst/analysis/FunctionRegistry.scala renamed to catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala

@@ -1,3 +1,4 @@
+package org.apache.spark.sql
 package catalyst
 package analysis
 

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

@@ -1,3 +1,4 @@
+package org.apache.spark.sql
 package catalyst
 package analysis
 

src/main/scala/catalyst/analysis/package.scala renamed to catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/package.scala

@@ -1,3 +1,4 @@
+package org.apache.spark.sql
 package catalyst
 package analysis
 
@@ -10,14 +11,14 @@ import trees.TreeNode
  * resolved.
  */
 class UnresolvedException[TreeType <: TreeNode[_]](tree: TreeType, function: String) extends
-  errors.TreeNodeException(tree, s"Invalid call to $function on unresolved object")
+  errors.TreeNodeException(tree, s"Invalid call to $function on unresolved object", null)
 
 /**
  * Holds the name of a relation that has yet to be looked up in a [[Catalog]].
  */
 case class UnresolvedRelation(
     databaseName: Option[String],
-    name: String,
+    tableName: String,
     alias: Option[String] = None) extends BaseRelation {
   def output = Nil
   override lazy val resolved = false

src/main/scala/catalyst/analysis/unresolved.scala renamed to catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala

@@ -1,3 +1,4 @@
+package org.apache.spark.sql
 package catalyst
 
 import scala.language.implicitConversions
@@ -118,10 +119,10 @@ package object dsl {
 
     def unionAll(otherPlan: LogicalPlan) = Union(plan, otherPlan)
 
-    def filter[T1](arg1: Symbol)(udf: (T1) => Boolean) =
+    def sfilter[T1](arg1: Symbol)(udf: (T1) => Boolean) =
       Filter(ScalaUdf(udf, BooleanType, Seq(UnresolvedAttribute(arg1.name))), plan)
 
-    def filter(dynamicUdf: (DynamicRow) => Boolean) =
+    def sfilter(dynamicUdf: (DynamicRow) => Boolean) =
       Filter(ScalaUdf(dynamicUdf, BooleanType, Seq(WrapDynamic(plan.output))), plan)
 
     def sample(