init import: create database and create function.

Author: viirya

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

@@ -23,7 +23,7 @@ import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 
-import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.{AnalysisException, Row}
 import org.apache.spark.sql.catalyst.{CatalystConf, EmptyConf, TableIdentifier}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Subquery}
 
@@ -50,6 +50,10 @@ trait Catalog {
     throw new UnsupportedOperationException
   }
 
+  def runNativeCommand(sql: String): Seq[Row] = {
+    throw new UnsupportedOperationException
+  }
+
   /**
    * Returns tuples of (tableName, isTemporary) for all tables in the given database.
    * isTemporary is a Boolean value indicates if a table is a temporary or not.

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkQl.scala

@@ -62,6 +62,61 @@ private[sql] class SparkQl(conf: ParserConf = SimpleParserConf()) extends Cataly
         val tableIdent = extractTableIdent(nameParts)
         RefreshTable(tableIdent)
 
+      case Token("TOK_CREATEDATABASE", Token(databaseName, Nil) :: createDatabaseArgs) =>
+        val Seq(
+          allowExisting,
+          dbLocation,
+          databaseComment,
+          dbprops) = getClauses(Seq(
+          "TOK_IFNOTEXISTS",
+          "TOK_DATABASELOCATION",
+          "TOK_DATABASECOMMENT",
+          "TOK_DATABASEPROPERTIES"), createDatabaseArgs)
+
+        val location = dbLocation.map {
+          case Token("TOK_DATABASELOCATION", Token(loc, Nil) :: Nil) => unquoteString(loc)
+        }
+        val comment = databaseComment.map {
+          case Token("TOK_DATABASECOMMENT", Token(comment, Nil) :: Nil) => unquoteString(comment)
+        }
+        val props: Map[String, String] = dbprops.toSeq.flatMap {
+          case Token("TOK_DATABASEPROPERTIES", propList) =>
+            propList.flatMap {
+              case Token("TOK_DBPROPLIST", props) =>
+                props.map {
+                  case Token("TOK_TABLEPROPERTY", keysAndValue) =>
+                    val key = keysAndValue.init.map(x => unquoteString(x.text)).mkString(".")
+                    val value = unquoteString(keysAndValue.last.text)
+                    (key, value)
+                }
+            }
+        }.toMap
+
+        CreateDataBase(databaseName, allowExisting.isDefined, location, comment, props)(node.source)
+
+      case Token("TOK_CREATEFUNCTION", func :: as :: createFuncArgs) =>
+        val funcName = func.map(x => unquoteString(x.text)).mkString(".")
+        val asName = unquoteString(as.text)
+        val Seq(
+          rList,
+          temp) = getClauses(Seq(
+          "TOK_RESOURCE_LIST",
+          "TOK_TEMPORARY"), createFuncArgs)
+
+        val resourcesMap: Map[String, String] = rList.toSeq.flatMap {
+          case Token("TOK_RESOURCE_LIST", resources) =>
+            resources.map {
+              case Token("TOK_RESOURCE_URI", rType :: Token(rPath, Nil) :: Nil) =>
+                val resourceType = rType match {
+                  case Token("TOK_JAR", Nil) => "jar"
+                  case Token("TOK_FILE", Nil) => "file"
+                  case Token("TOK_ARCHIVE", Nil) => "archive"
+                }
+                (resourceType, unquoteString(rPath))
+            }
+        }.toMap
+        CreateFunction(funcName, asName, resourcesMap, temp.isDefined)(node.source)
+
       case Token("TOK_CREATETABLEUSING", createTableArgs) =>
         val Seq(
           temp,

sql/core/src/main/scala/org/apache/spark/sql/execution/commands.scala

@@ -418,3 +418,32 @@ case class SetDatabaseCommand(databaseName: String) extends RunnableCommand {
 
   override val output: Seq[Attribute] = Seq.empty
 }
+
+case class CreateDataBase(
+    databaseName: String,
+    allowExisting: Boolean,
+    path: Option[String],
+    comment: Option[String],
+    props: Map[String, String])(sql: String) extends RunnableCommand with Logging {
+
+  override def run(sqlContext: SQLContext): Seq[Row] = {
+    sqlContext.catalog.runNativeCommand(sql)
+  }
+
+  override val output: Seq[Attribute] =
+    Seq(AttributeReference("result", StringType, nullable = false)())
+}
+
+case class CreateFunction(
+    functionName: String,
+    asName: String,
+    resourcesMap: Map[String, String],
+    isTemp: Boolean)(sql: String) extends RunnableCommand with Logging {
+
+  override def run(sqlContext: SQLContext): Seq[Row] = {
+    sqlContext.catalog.runNativeCommand(sql)
+  }
+
+  override val output: Seq[Attribute] =
+    Seq(AttributeReference("result", StringType, nullable = false)())
+}

sql/core/src/test/scala/org/apache/spark/sql/execution/SparkQlSuite.scala

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution
+
+import org.apache.spark.sql.catalyst.plans.PlanTest
+
+class SparkQlSuite extends PlanTest {
+  val parser = new SparkQl()
+
+  test("create database") {
+    parser.parsePlan("CREATE DATABASE IF NOT EXISTS database_name " +
+      "COMMENT 'database_comment' LOCATION '/home/user/db' " +
+      "WITH DBPROPERTIES ('a'='a', 'b'='b', 'c'='c')")
+  }
+
+  test("create function") {
+    parser.parsePlan("CREATE TEMPORARY FUNCTION helloworld as " +
+      "'com.matthewrathbone.example.SimpleUDFExample' USING JAR '/path/to/jar', " +
+      "FILE 'path/to/file'")
+  }
+}

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala

@@ -31,7 +31,7 @@ import org.apache.hadoop.hive.ql.metadata._
 import org.apache.hadoop.hive.ql.plan.TableDesc
 
 import org.apache.spark.Logging
-import org.apache.spark.sql.{AnalysisException, SaveMode, SQLContext}
+import org.apache.spark.sql.{AnalysisException, SaveMode, Row, SQLContext}
 import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
 import org.apache.spark.sql.catalyst.analysis.{Catalog, MultiInstanceRelation, OverrideCatalog}
 import org.apache.spark.sql.catalyst.expressions._
@@ -715,6 +715,10 @@ private[hive] class HiveMetastoreCatalog(val client: HiveClient, hive: HiveConte
   override def setCurrentDatabase(databaseName: String): Unit = {
     client.setCurrentDatabase(databaseName)
   }
+
+  override def runNativeCommand(sql: String): Seq[Row] = {
+    hive.runSqlHive(sql).map(Row(_))
+  }
 }
 
 /**

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveQl.scala

@@ -107,8 +107,6 @@ private[hive] class HiveQl(conf: ParserConf) extends SparkQl(conf) with Logging
     "TOK_ALTERVIEW_PROPERTIES",
     "TOK_ALTERVIEW_RENAME",
 
-    "TOK_CREATEDATABASE",
-    "TOK_CREATEFUNCTION",
     "TOK_CREATEINDEX",
     "TOK_CREATEMACRO",
     "TOK_CREATEROLE",