From 014c38e28e8f4545f926ef60ccb2ee4acae07b59 Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Fri, 8 Apr 2016 14:32:39 -0700
Subject: [PATCH 01/18] Parse various parts of the CREATE TABLE command

We need to reconcile the differences between what's added here in
SparkSqlParser and HiveSqlParser. That will come in the next
commit.

This currently still fails tests, obviously because create table
is not implemented yet!
---
 .../spark/sql/catalyst/parser/SqlBase.g4      |   3 +-
 .../sql/catalyst/catalog/interface.scala      |  52 +++++++
 .../spark/sql/execution/SparkSqlParser.scala  | 143 +++++++++++++++---
 .../spark/sql/execution/command/ddl.scala     |  28 +---
 .../spark/sql/execution/command/tables.scala  |  94 ++++++++++++
 .../execution/command/DDLCommandSuite.scala   |  26 +---
 .../sql/execution/command/DDLSuite.scala      |   9 +-
 .../sql/hive/execution/HiveSqlParser.scala    |  23 +--
 8 files changed, 290 insertions(+), 88 deletions(-)
 create mode 100644 sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala

diff --git a/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4 b/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
index 85cb585919da0..55720caa8ec20 100644
--- a/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
+++ b/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
@@ -271,8 +271,7 @@ createFileFormat
     ;
 
 fileFormat
-    : INPUTFORMAT inFmt=STRING OUTPUTFORMAT outFmt=STRING (SERDE serdeCls=STRING)?
-      (INPUTDRIVER inDriver=STRING OUTPUTDRIVER outDriver=STRING)?                         #tableFileFormat
+    : INPUTFORMAT inFmt=STRING OUTPUTFORMAT outFmt=STRING (SERDE serdeCls=STRING)?         #tableFileFormat
     | identifier                                                                           #genericFileFormat
     ;
 
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
index e29d6bd8b09ea..7c928549862c0 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
@@ -291,3 +291,55 @@ case class CatalogRelation(
   require(metadata.identifier.database == Some(db),
     "provided database does not match the one specified in the table definition")
 }
+
+
+/**
+ * Skew specifications for a table.
+ */
+case class SkewSpec(
+    // e.g. ['datetime', 'country']
+    columns: Seq[String],
+    // e.g. [['2008-08-08', 'us], ['2009-09-09', 'uk']]
+    values: Seq[Seq[String]],
+    storedAsDirs: Boolean) {
+
+  require(values.forall(_.size == columns.size),
+    "number of columns in skewed values do not match number of skewed columns provided")
+}
+
+
+/**
+ * Row format for creating tables, specified with ROW FORMAT [...].
+ */
+sealed trait RowFormat
+
+case class RowFormatSerde(serde: String, serdeProperties: Map[String, String])
+  extends RowFormat
+
+case class RowFormatDelimited(
+    fieldsTerminatedBy: Option[String],
+    fieldsEscapedBy: Option[String],
+    mapKeysTerminatedBy: Option[String],
+    linesTerminatedBy: Option[String],
+    nullDefinedAs: Option[String])
+  extends RowFormat
+
+
+/**
+ * File format for creating tables, specified with STORED AS [...].
+ */
+sealed trait FileFormat
+
+case class TableFileFormat(
+    inputFormat: String,
+    outputFormat: String,
+    serde: Option[String] = None)
+  extends FileFormat
+
+case class GenericFileFormat(format: String) extends FileFormat
+
+
+/**
+ * Storage handler for creating tables, specified with STORED BY [...].
+ */
+case class StorageHandler(handlerClass: String, serdeProps: Map[String, String])
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 3de8aa02766dc..87198e0777587 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -20,6 +20,7 @@ import scala.collection.JavaConverters._
 
 import org.apache.spark.sql.{AnalysisException, SaveMode}
 import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.parser._
 import org.apache.spark.sql.catalyst.parser.SqlBaseParser._
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, OneRowRelation}
@@ -182,9 +183,61 @@ class SparkSqlAstBuilder extends AstBuilder {
     }
   }
 
-  /** Type to keep track of a table header. */
+  /**
+   * Type to keep track of a table header: (identifier, isTemporary, ifNotExists, isExternal).
+   */
   type TableHeader = (TableIdentifier, Boolean, Boolean, Boolean)
 
+  /**
+   * Create a [[CreateTable]] command.
+   *
+   * For example:
+   * {{{
+   *   CREATE [TEMPORARY] [EXTERNAL] TABLE [IF NOT EXISTS] [db_name.]table_name
+   *   [(col1 data_type [COMMENT col_comment], ...)]
+   *   [COMMENT table_comment]
+   *   [PARTITIONED BY (col3 data_type [COMMENT col_comment], ...)]
+   *   [CLUSTERED BY (col1, ...) [SORTED BY (col1 [ASC|DESC], ...)] INTO num_buckets BUCKETS]
+   *   [SKEWED BY (col1, col2, ...) ON ((col_value, col_value, ...), ...)
+   *   [STORED AS DIRECTORIES]
+   *   [ROW FORMAT row_format]
+   *   [STORED AS file_format | STORED BY storage_handler_class [WITH SERDEPROPERTIES (...)]]
+   *   [LOCATION path]
+   *   [TBLPROPERTIES (property_name=property_value, ...)]
+   *   [AS select_statement];
+   * }}}
+   */
+  override def visitCreateTable(ctx: CreateTableContext): LogicalPlan = withOrigin(ctx) {
+    val (name, temp, ifNotExists, external) = visitCreateTableHeader(ctx.createTableHeader)
+    val comment = Option(ctx.STRING).map(string)
+    val columns = Option(ctx.columns).map(visitColTypeList).getOrElse(Seq())
+    val partitionColumns = Option(ctx.partitionColumns).map(visitColTypeList).getOrElse(Seq())
+    val bucketSpec = Option(ctx.bucketSpec).map(visitBucketSpec)
+    val skewSpec = Option(ctx.skewSpec).map(visitSkewSpec)
+    val rowFormat = Option(ctx.rowFormat).map(visitRowFormat)
+    val fileFormat = Option(ctx.createFileFormat).map(_.fileFormat).map(visitFileFormat)
+    val storageHandler = Option(ctx.createFileFormat).map(_.storageHandler).map(visitStorageHandler)
+    val location = Option(ctx.locationSpec).map(_.STRING).map(string)
+    val properties = Option(ctx.tablePropertyList).map(visitTablePropertyList).getOrElse(Map())
+    val selectQuery = Option(ctx.query).map(plan)
+    CreateTable(
+      name,
+      temp,
+      ifNotExists,
+      external,
+      comment,
+      columns,
+      partitionColumns,
+      bucketSpec,
+      skewSpec,
+      rowFormat,
+      fileFormat,
+      storageHandler,
+      location,
+      properties,
+      selectQuery)
+  }
+
   /**
    * Validate a create table statement and return the [[TableIdentifier]].
    */
@@ -599,25 +652,10 @@ class SparkSqlAstBuilder extends AstBuilder {
    */
   override def visitSetTableFileFormat(
       ctx: SetTableFileFormatContext): LogicalPlan = withOrigin(ctx) {
-    // AlterTableSetFileFormat currently takes both a GenericFileFormat and a
-    // TableFileFormatContext. This is a bit weird because it should only take one. It also should
-    // use a CatalogFileFormat instead of either a String or a Sequence of Strings. We will address
-    // this in a follow-up PR.
-    val (fileFormat, genericFormat) = ctx.fileFormat match {
-      case s: GenericFileFormatContext =>
-        (Seq.empty[String], Option(s.identifier.getText))
-      case s: TableFileFormatContext =>
-        val elements = Seq(s.inFmt, s.outFmt) ++
-          Option(s.serdeCls).toSeq ++
-          Option(s.inDriver).toSeq ++
-          Option(s.outDriver).toSeq
-        (elements.map(string), None)
-    }
     AlterTableSetFileFormat(
       visitTableIdentifier(ctx.tableIdentifier),
       Option(ctx.partitionSpec).map(visitNonOptionalPartitionSpec),
-      fileFormat,
-      genericFormat)(
+      visitFileFormat(ctx.fileFormat))(
       command(ctx))
   }
 
@@ -780,14 +818,79 @@ class SparkSqlAstBuilder extends AstBuilder {
    * - Values for which are skewed. The size of each entry must match the number of skewed columns.
    * - A store in directory flag.
    */
-  override def visitSkewSpec(
-      ctx: SkewSpecContext): (Seq[String], Seq[Seq[String]], Boolean) = withOrigin(ctx) {
+  override def visitSkewSpec(ctx: SkewSpecContext): SkewSpec = withOrigin(ctx) {
     val skewedValues = if (ctx.constantList != null) {
       Seq(visitConstantList(ctx.constantList))
     } else {
       visitNestedConstantList(ctx.nestedConstantList)
     }
-    (visitIdentifierList(ctx.identifierList), skewedValues, ctx.DIRECTORIES != null)
+    SkewSpec(visitIdentifierList(ctx.identifierList), skewedValues, ctx.DIRECTORIES != null)
+  }
+
+  /**
+   * Create a [[RowFormat]] used for creating tables.
+   *
+   * Example format:
+   * {{{
+   *   SERDE serde_name [WITH SERDEPROPERTIES (k1=v1, k2=v2, ...)]
+   * }}}
+   *
+   * OR
+   *
+   * {{{
+   *   DELIMITED [FIELDS TERMINATED BY char [ESCAPED BY char]]
+   *   [COLLECTION ITEMS TERMINATED BY char]
+   *   [MAP KEYS TERMINATED BY char]
+   *   [LINES TERMINATED BY char]
+   *   [NULL DEFINED AS char]
+   * }}}
+   */
+  private def visitRowFormat(ctx: RowFormatContext): RowFormat = withOrigin(ctx) {
+    ctx match {
+      case serde: RowFormatSerdeContext =>
+        RowFormatSerde(
+          string(serde.name),
+          visitTablePropertyList(serde.props))
+      case delimited: RowFormatDelimitedContext =>
+        RowFormatDelimited(
+          fieldsTerminatedBy = Option(delimited.fieldsTerminatedBy).map(string),
+          fieldsEscapedBy = Option(delimited.escapedBy).map(string),
+          mapKeysTerminatedBy = Option(delimited.keysTerminatedBy).map(string),
+          linesTerminatedBy = Option(delimited.linesSeparatedBy).map(string),
+          nullDefinedAs = Option(delimited.nullDefinedAs).map(string))
+    }
+  }
+
+  /**
+   * Create a [[FileFormat]] for creating and altering tables.
+   *
+   * Example format:
+   * {{{
+   *   SEQUENCEFILE |
+   *   TEXTFILE |
+   *   RCFILE |
+   *   ORC |
+   *   PARQUET |
+   *   AVRO |
+   *   INPUTFORMAT input_format_classname OUTPUTFORMAT output_format_classname
+   * }}}
+   */
+  private def visitFileFormat(ctx: FileFormatContext): FileFormat = withOrigin(ctx) {
+    ctx match {
+      case s: GenericFileFormatContext =>
+        GenericFileFormat(s.identifier.getText)
+      case s: TableFileFormatContext =>
+        TableFileFormat(string(s.inFmt), string(s.outFmt), Option(s.serdeCls).map(string))
+    }
+  }
+
+  /**
+   * Create a [[StorageHandler]] for creating tables.
+   */
+  override def visitStorageHandler(ctx: StorageHandlerContext): StorageHandler = withOrigin(ctx) {
+    val handlerClassName = string(ctx.STRING)
+    val serdeProps = Option(ctx.tablePropertyList).map(visitTablePropertyList).getOrElse(Map())
+    StorageHandler(handlerClassName, serdeProps)
   }
 
   /**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
index 20779d68e0fdd..d14db2c6f8af2 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.execution.command
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.{AnalysisException, Row, SQLContext}
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, CatalogTable}
+import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, CatalogTable, FileFormat}
 import org.apache.spark.sql.catalyst.catalog.ExternalCatalog.TablePartitionSpec
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
 import org.apache.spark.sql.types._
@@ -175,29 +175,6 @@ case class DescribeDatabase(
   }
 }
 
-/**
- * A command that renames a table/view.
- *
- * The syntax of this command is:
- * {{{
- *    ALTER TABLE table1 RENAME TO table2;
- *    ALTER VIEW view1 RENAME TO view2;
- * }}}
- */
-case class AlterTableRename(
-    oldName: TableIdentifier,
-    newName: TableIdentifier)
-  extends RunnableCommand {
-
-  override def run(sqlContext: SQLContext): Seq[Row] = {
-    val catalog = sqlContext.sessionState.catalog
-    catalog.invalidateTable(oldName)
-    catalog.renameTable(oldName, newName)
-    Seq.empty[Row]
-  }
-
-}
-
 /**
  * A command that sets table/view properties.
  *
@@ -354,8 +331,7 @@ case class AlterTableUnarchivePartition(
 case class AlterTableSetFileFormat(
     tableName: TableIdentifier,
     partitionSpec: Option[TablePartitionSpec],
-    fileFormat: Seq[String],
-    genericFormat: Option[String])(sql: String)
+    fileFormat: FileFormat)(sql: String)
   extends NativeDDLCommand(sql) with Logging
 
 /**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
new file mode 100644
index 0000000000000..2ed9a108d4ae8
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -0,0 +1,94 @@
+/*
+ * 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.command
+
+import org.apache.spark.sql.{Row, SQLContext}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{FileFormat, RowFormat, SkewSpec, StorageHandler}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.execution.datasources.BucketSpec
+import org.apache.spark.sql.types.StructField
+
+
+// TODO: move the rest of the table commands from ddl.scala to this file
+
+/**
+ * A command to create a table.
+ *
+ * The syntax of using this command in SQL is:
+ * {{{
+ *   CREATE [TEMPORARY] [EXTERNAL] TABLE [IF NOT EXISTS] [db_name.]table_name
+ *   [(col1 data_type [COMMENT col_comment], ...)]
+ *   [COMMENT table_comment]
+ *   [PARTITIONED BY (col3 data_type [COMMENT col_comment], ...)]
+ *   [CLUSTERED BY (col1, ...) [SORTED BY (col1 [ASC|DESC], ...)] INTO num_buckets BUCKETS]
+ *   [SKEWED BY (col1, col2, ...) ON ((col_value, col_value, ...), ...)
+ *   [STORED AS DIRECTORIES]
+ *   [ROW FORMAT row_format]
+ *   [STORED AS file_format | STORED BY storage_handler_class [WITH SERDEPROPERTIES (...)]]
+ *   [LOCATION path]
+ *   [TBLPROPERTIES (property_name=property_value, ...)]
+ *   [AS select_statement];
+ * }}}
+ */
+case class CreateTable(
+    name: TableIdentifier,
+    isTemp: Boolean,
+    ifNotExists: Boolean,
+    isExternal: Boolean,
+    comment: Option[String],
+    columns: Seq[StructField],
+    partitionedColumns: Seq[StructField],
+    bucketSpec: Option[BucketSpec],
+    skewSpec: Option[SkewSpec],
+    rowFormat: Option[RowFormat],
+    fileFormat: Option[FileFormat],
+    storageHandler: Option[StorageHandler],
+    location: Option[String],
+    properties: Map[String, String],
+    selectQuery: Option[LogicalPlan])
+  extends RunnableCommand {
+
+  override def run(sqlContext: SQLContext): Seq[Row] = {
+    Seq.empty[Row]
+  }
+
+}
+
+/**
+ * A command that renames a table/view.
+ *
+ * The syntax of this command is:
+ * {{{
+ *    ALTER TABLE table1 RENAME TO table2;
+ *    ALTER VIEW view1 RENAME TO view2;
+ * }}}
+ */
+case class AlterTableRename(
+    oldName: TableIdentifier,
+    newName: TableIdentifier)
+  extends RunnableCommand {
+
+  override def run(sqlContext: SQLContext): Seq[Row] = {
+    val catalog = sqlContext.sessionState.catalog
+    catalog.invalidateTable(oldName)
+    catalog.renameTable(oldName, newName)
+    Seq.empty[Row]
+  }
+
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
index 8e63b69876501..eeb9a61d2c13b 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
@@ -18,11 +18,11 @@
 package org.apache.spark.sql.execution.command
 
 import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.catalyst.plans.PlanTest
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.plans.logical.Project
 import org.apache.spark.sql.execution.SparkSqlParser
-import org.apache.spark.sql.execution.datasources.BucketSpec
 import org.apache.spark.sql.types._
 
 class DDLCommandSuite extends PlanTest {
@@ -454,37 +454,23 @@ class DDLCommandSuite extends PlanTest {
   }
 
   test("alter table: set file format") {
-    val sql1 =
-      """
-       |ALTER TABLE table_name SET FILEFORMAT INPUTFORMAT 'test'
-       |OUTPUTFORMAT 'test' SERDE 'test' INPUTDRIVER 'test' OUTPUTDRIVER 'test'
-      """.stripMargin
-    val sql2 = "ALTER TABLE table_name SET FILEFORMAT INPUTFORMAT 'test' " +
+    val sql1 = "ALTER TABLE table_name SET FILEFORMAT INPUTFORMAT 'test' " +
       "OUTPUTFORMAT 'test' SERDE 'test'"
-    val sql3 = "ALTER TABLE table_name PARTITION (dt='2008-08-08', country='us') " +
+    val sql2 = "ALTER TABLE table_name PARTITION (dt='2008-08-08', country='us') " +
       "SET FILEFORMAT PARQUET"
     val parsed1 = parser.parsePlan(sql1)
     val parsed2 = parser.parsePlan(sql2)
-    val parsed3 = parser.parsePlan(sql3)
     val tableIdent = TableIdentifier("table_name", None)
     val expected1 = AlterTableSetFileFormat(
       tableIdent,
       None,
-      List("test", "test", "test", "test", "test"),
-      None)(sql1)
+      TableFileFormat("test", "test", Some("test")))(sql1)
     val expected2 = AlterTableSetFileFormat(
-      tableIdent,
-      None,
-      List("test", "test", "test"),
-      None)(sql2)
-    val expected3 = AlterTableSetFileFormat(
       tableIdent,
       Some(Map("dt" -> "2008-08-08", "country" -> "us")),
-      Seq(),
-      Some("PARQUET"))(sql3)
+      GenericFileFormat("PARQUET"))(sql2)
     comparePlans(parsed1, expected1)
     comparePlans(parsed2, expected2)
-    comparePlans(parsed3, expected3)
   }
 
   test("alter table: set location") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
index 7084665b3b802..b3ccfa3ac4740 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
@@ -196,6 +196,13 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
 
   // TODO: test drop database in restrict mode
 
+  test("create table") {
+    val catalog = sqlContext.sessionState.catalog
+    val tableIdent1 = TableIdentifier("tab1", Some("dbx"))
+    createDatabase(catalog, "dbx")
+    sql("CREATE TABLE dbx.tab1 (id int, name string) partitioned by (nickname string)")
+  }
+
   test("alter table: rename") {
     val catalog = sqlContext.sessionState.catalog
     val tableIdent1 = TableIdentifier("tab1", Some("dbx"))
@@ -320,8 +327,6 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
     assertUnsupported("ALTER TABLE dbx.tab1 NOT STORED AS DIRECTORIES")
   }
 
-  // TODO: ADD a testcase for Drop Database in Restric when we can create tables in SQLContext
-
   test("show tables") {
     withTempTable("show1a", "show2b") {
       sql(
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
index ab69d3502e938..4391356495d45 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
@@ -141,7 +141,7 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
     if (ctx.storageHandler == null) {
       typedVisit[CatalogStorageFormat](ctx.fileFormat)
     } else {
-      visitStorageHandler(ctx.storageHandler)
+      throw new ParseException("Storage Handlers are currently unsupported.", ctx)
     }
   }
 
@@ -363,19 +363,14 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
   private val EmptyStorageFormat = CatalogStorageFormat(None, None, None, None, Map.empty)
 
   /**
-   * Create a [[CatalogStorageFormat]]. The INPUTDRIVER and OUTPUTDRIVER clauses are currently
-   * ignored.
+   * Create a [[CatalogStorageFormat]].
    */
   override def visitTableFileFormat(
       ctx: TableFileFormatContext): CatalogStorageFormat = withOrigin(ctx) {
-    import ctx._
-    if (inDriver != null || outDriver != null) {
-      logWarning("INPUTDRIVER ... OUTPUTDRIVER ... clauses are ignored.")
-    }
     EmptyStorageFormat.copy(
-      inputFormat = Option(string(inFmt)),
-      outputFormat = Option(string(outFmt)),
-      serde = Option(serdeCls).map(string)
+      inputFormat = Option(string(ctx.inFmt)),
+      outputFormat = Option(string(ctx.outFmt)),
+      serde = Option(ctx.serdeCls).map(string)
     )
   }
 
@@ -396,14 +391,6 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
     }
   }
 
-  /**
-   * Storage Handlers are currently not supported in the statements we support (CTAS).
-   */
-  override def visitStorageHandler(
-      ctx: StorageHandlerContext): CatalogStorageFormat = withOrigin(ctx) {
-    throw new ParseException("Storage Handlers are currently unsupported.", ctx)
-  }
-
   /**
    * Create SERDE row format name and properties pair.
    */

From 15bb3b6c76e61d708538bee5d797981689ab6a8f Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Fri, 1 Apr 2016 14:20:37 -0700
Subject: [PATCH 02/18] Refactor CatalogTable column semantics

Before: CatalogTable has schema, partitionColumns and sortColumns.
There are no constraints between the 3. However, Hive will
complain if schema and partitionColumns overlap.

After: CatalogTable has schema, partitionColumnNames,
sortColumnNames, bucketColumnNames and skewColumnNames. All the
columns must be a subset of schema. This means splitting up
schema into (schema, partitionCols) before passing it to Hive.

This allows us to store the columns more uniformly. Otherwise
partition columns would be the odd one out. This commit also
fixes "alter table bucketing", which was incorrectly using
partition columns as bucket columns.
---
 .../sql/catalyst/catalog/interface.scala      | 24 +++++++++++++++++--
 .../catalyst/catalog/CatalogTestCases.scala   |  8 +++++--
 .../spark/sql/hive/HiveMetastoreCatalog.scala |  9 +++++--
 .../sql/hive/client/HiveClientImpl.scala      | 24 +++++++++++++++----
 .../sql/hive/execution/HiveSqlParser.scala    | 10 ++++----
 .../sql/hive/HiveMetastoreCatalogSuite.scala  |  4 ++--
 .../sql/hive/execution/PruningSuite.scala     |  2 +-
 7 files changed, 63 insertions(+), 18 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
index 7c928549862c0..c60e8fab8aaca 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
@@ -218,8 +218,13 @@ case class CatalogTable(
     tableType: CatalogTableType,
     storage: CatalogStorageFormat,
     schema: Seq[CatalogColumn],
-    partitionColumns: Seq[CatalogColumn] = Seq.empty,
-    sortColumns: Seq[CatalogColumn] = Seq.empty,
+    partitionColumnNames: Seq[String] = Seq.empty,
+    sortColumnNames: Seq[String] = Seq.empty,
+    bucketColumnNames: Seq[String] = Seq.empty,
+    // e.g. (date, country)
+    skewColumnNames: Seq[String] = Seq.empty,
+    // e.g. ('2008-08-08', 'us), ('2009-09-09', 'uk')
+    skewColumnValues: Seq[Seq[String]] = Seq.empty,
     numBuckets: Int = 0,
     createTime: Long = System.currentTimeMillis,
     lastAccessTime: Long = System.currentTimeMillis,
@@ -227,6 +232,21 @@ case class CatalogTable(
     viewOriginalText: Option[String] = None,
     viewText: Option[String] = None) {
 
+  // Verify that the provided columns are part of the schema
+  private val colNames = schema.map(_.name).toSet
+  private def requireSubsetOfSchema(cols: Seq[String], colType: String): Unit = {
+    require(cols.toSet.subsetOf(colNames), s"$colType columns (${cols.mkString(", ")}) " +
+      s"must be a subset of schema (${colNames.mkString(", ")}) in table '$identifier'")
+  }
+  requireSubsetOfSchema(partitionColumnNames, "partition")
+  requireSubsetOfSchema(sortColumnNames, "sort")
+  requireSubsetOfSchema(bucketColumnNames, "bucket")
+  requireSubsetOfSchema(skewColumnNames, "skew")
+
+  /** Columns this table is partitioned by. */
+  def partitionColumns: Seq[CatalogColumn] =
+    schema.filter { c => partitionColumnNames.contains(c.name) }
+
   /** Return the database this table was specified to belong to, assuming it exists. */
   def database: String = identifier.database.getOrElse {
     throw new AnalysisException(s"table $identifier did not specify database")
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/CatalogTestCases.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/CatalogTestCases.scala
index fbcac09ce223f..c0e61fc7bfbf3 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/CatalogTestCases.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/CatalogTestCases.scala
@@ -538,8 +538,12 @@ abstract class CatalogTestUtils {
       identifier = TableIdentifier(name, database),
       tableType = CatalogTableType.EXTERNAL_TABLE,
       storage = storageFormat,
-      schema = Seq(CatalogColumn("col1", "int"), CatalogColumn("col2", "string")),
-      partitionColumns = Seq(CatalogColumn("a", "int"), CatalogColumn("b", "string")))
+      schema = Seq(
+        CatalogColumn("col1", "int"),
+        CatalogColumn("col2", "string"),
+        CatalogColumn("a", "int"),
+        CatalogColumn("b", "string")),
+      partitionColumnNames = Seq("a", "b"))
   }
 
   def newFunc(name: String, database: Option[String] = None): CatalogFunction = {
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
index 14f331961ef4a..be1db95ad43b5 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
@@ -905,8 +905,13 @@ private[hive] case class MetastoreRelation(
 
     val sd = new org.apache.hadoop.hive.metastore.api.StorageDescriptor()
     tTable.setSd(sd)
-    sd.setCols(table.schema.map(toHiveColumn).asJava)
-    tTable.setPartitionKeys(table.partitionColumns.map(toHiveColumn).asJava)
+
+    // Note: In Hive the schema and partition columns must be disjoint sets
+    val (partCols, schema) = table.schema.map(toHiveColumn).partition { c =>
+      table.partitionColumnNames.contains(c.getName)
+    }
+    sd.setCols(schema.asJava)
+    tTable.setPartitionKeys(partCols.asJava)
 
     table.storage.locationUri.foreach(sd.setLocation)
     table.storage.inputFormat.foreach(sd.setInputFormat)
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index d0eb9ddf50ae8..ca5ac494e31a6 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -299,6 +299,10 @@ private[hive] class HiveClientImpl(
       tableName: String): Option[CatalogTable] = withHiveState {
     logDebug(s"Looking up $dbName.$tableName")
     Option(client.getTable(dbName, tableName, false)).map { h =>
+      // Note: Hive separates partition columns and the schema, but for us the
+      // partition columns are part of the schema
+      val partCols = h.getPartCols.asScala.map(fromHiveColumn)
+      val schema = h.getCols.asScala.map(fromHiveColumn) ++ partCols
       CatalogTable(
         identifier = TableIdentifier(h.getTableName, Option(h.getDbName)),
         tableType = h.getTableType match {
@@ -307,9 +311,12 @@ private[hive] class HiveClientImpl(
           case HiveTableType.INDEX_TABLE => CatalogTableType.INDEX_TABLE
           case HiveTableType.VIRTUAL_VIEW => CatalogTableType.VIRTUAL_VIEW
         },
-        schema = h.getCols.asScala.map(fromHiveColumn),
-        partitionColumns = h.getPartCols.asScala.map(fromHiveColumn),
-        sortColumns = Seq(),
+        schema = schema,
+        partitionColumnNames = partCols.map(_.name),
+        sortColumnNames = Seq(), // TODO: populate this
+        bucketColumnNames = h.getBucketCols.asScala,
+        skewColumnNames = h.getSkewedColNames.asScala,
+        skewColumnValues = h.getSkewedColValues.asScala.map(_.asScala),
         numBuckets = h.getNumBuckets,
         createTime = h.getTTable.getCreateTime.toLong * 1000,
         lastAccessTime = h.getLastAccessTime.toLong * 1000,
@@ -665,9 +672,16 @@ private[hive] class HiveClientImpl(
       case CatalogTableType.INDEX_TABLE => HiveTableType.INDEX_TABLE
       case CatalogTableType.VIRTUAL_VIEW => HiveTableType.VIRTUAL_VIEW
     })
-    hiveTable.setFields(table.schema.map(toHiveColumn).asJava)
-    hiveTable.setPartCols(table.partitionColumns.map(toHiveColumn).asJava)
+    // Note: In Hive the schema and partition columns must be disjoint sets
+    val (partCols, schema) = table.schema.map(toHiveColumn).partition { c =>
+      table.partitionColumnNames.contains(c.getName)
+    }
+    hiveTable.setFields(schema.asJava)
+    hiveTable.setPartCols(partCols.asJava)
     // TODO: set sort columns here too
+    hiveTable.setBucketCols(table.bucketColumnNames.asJava)
+    hiveTable.setSkewedColNames(table.skewColumnNames.asJava)
+    hiveTable.setSkewedColValues(table.skewColumnValues.map(_.asJava).asJava)
     hiveTable.setOwner(conf.getUser)
     hiveTable.setNumBuckets(table.numBuckets)
     hiveTable.setCreateTime((table.createTime / 1000).toInt)
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
index 4391356495d45..585bc77a0fbe9 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
@@ -172,12 +172,14 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
         logWarning("SKEWED BY ... ON ... [STORED AS DIRECTORIES] clause is ignored.")
       }
 
-      // Create the schema.
-      val schema = Option(ctx.columns).toSeq.flatMap(visitCatalogColumns(_, _.toLowerCase))
-
       // Get the column by which the table is partitioned.
       val partitionCols = Option(ctx.partitionColumns).toSeq.flatMap(visitCatalogColumns(_))
 
+      // Note: Hive requires partition columns to be distinct from the schema, so we need
+      // to include the partition columns here explicitly
+      val schema =
+        Option(ctx.columns).toSeq.flatMap(visitCatalogColumns(_, _.toLowerCase)) ++ partitionCols
+
       // Create the storage.
       def format(fmt: ParserRuleContext): CatalogStorageFormat = {
         Option(fmt).map(typedVisit[CatalogStorageFormat]).getOrElse(EmptyStorageFormat)
@@ -204,7 +206,7 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
         identifier = table,
         tableType = tableType,
         schema = schema,
-        partitionColumns = partitionCols,
+        partitionColumnNames = partitionCols.map(_.name),
         storage = storage,
         properties = Option(ctx.tablePropertyList).map(visitTablePropertyList).getOrElse(Map.empty),
         // TODO support the sql text - have a proper location for this!
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveMetastoreCatalogSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveMetastoreCatalogSuite.scala
index ada8621d07579..8648834f0d881 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveMetastoreCatalogSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveMetastoreCatalogSuite.scala
@@ -88,7 +88,7 @@ class DataSourceWithHiveMetastoreCatalogSuite
         assert(hiveTable.storage.outputFormat === Some(outputFormat))
         assert(hiveTable.storage.serde === Some(serde))
 
-        assert(hiveTable.partitionColumns.isEmpty)
+        assert(hiveTable.partitionColumnNames.isEmpty)
         assert(hiveTable.tableType === CatalogTableType.MANAGED_TABLE)
 
         val columns = hiveTable.schema
@@ -151,7 +151,7 @@ class DataSourceWithHiveMetastoreCatalogSuite
           assert(hiveTable.storage.outputFormat === Some(outputFormat))
           assert(hiveTable.storage.serde === Some(serde))
 
-          assert(hiveTable.partitionColumns.isEmpty)
+          assert(hiveTable.partitionColumnNames.isEmpty)
           assert(hiveTable.tableType === CatalogTableType.EXTERNAL_TABLE)
 
           val columns = hiveTable.schema
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/PruningSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/PruningSuite.scala
index 37c01792d9c3f..97cb9d972081c 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/PruningSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/PruningSuite.scala
@@ -149,7 +149,7 @@ class PruningSuite extends HiveComparisonTest with BeforeAndAfter {
       val (actualScannedColumns, actualPartValues) = plan.collect {
         case p @ HiveTableScan(columns, relation, _) =>
           val columnNames = columns.map(_.name)
-          val partValues = if (relation.table.partitionColumns.nonEmpty) {
+          val partValues = if (relation.table.partitionColumnNames.nonEmpty) {
             p.prunePartitions(relation.getHiveQlPartitions()).map(_.getValues)
           } else {
             Seq.empty

From b6b4d293c2efeb537110ef56fa9ffdcad90c9bb0 Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Fri, 8 Apr 2016 17:53:18 -0700
Subject: [PATCH 03/18] Implement CREATE TABLE in Hive parser

This involves reverting part of the changes in an earlier commit,
where we tried to implement the parsing logic in the general SQL
parser and introduced a bunch of case classes that we won't end
up using.

As of this commit the actual CREATE TABLE logic is not there yet.
It will come in a future commit.
---
 .../sql/catalyst/catalog/interface.scala      |  43 +---
 .../spark/sql/execution/SparkSqlParser.scala  | 130 ++----------
 .../spark/sql/execution/command/ddl.scala     |   5 +-
 .../spark/sql/execution/command/tables.scala  |  30 +--
 .../execution/command/DDLCommandSuite.scala   |  22 +-
 .../sql/execution/command/DDLSuite.scala      |   7 -
 .../sql/hive/client/HiveClientImpl.scala      |   1 +
 .../sql/hive/execution/HiveSqlParser.scala    | 196 ++++++++++++------
 8 files changed, 175 insertions(+), 259 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
index c60e8fab8aaca..ef96acd33e6bd 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
@@ -181,7 +181,8 @@ case class CatalogStorageFormat(
     inputFormat: Option[String],
     outputFormat: Option[String],
     serde: Option[String],
-    serdeProperties: Map[String, String])
+    serdeProperties: Map[String, String],
+    storedAsDirs: Boolean = false)
 
 
 /**
@@ -230,7 +231,8 @@ case class CatalogTable(
     lastAccessTime: Long = System.currentTimeMillis,
     properties: Map[String, String] = Map.empty,
     viewOriginalText: Option[String] = None,
-    viewText: Option[String] = None) {
+    viewText: Option[String] = None,
+    comment: Option[String] = None) {
 
   // Verify that the provided columns are part of the schema
   private val colNames = schema.map(_.name).toSet
@@ -326,40 +328,3 @@ case class SkewSpec(
   require(values.forall(_.size == columns.size),
     "number of columns in skewed values do not match number of skewed columns provided")
 }
-
-
-/**
- * Row format for creating tables, specified with ROW FORMAT [...].
- */
-sealed trait RowFormat
-
-case class RowFormatSerde(serde: String, serdeProperties: Map[String, String])
-  extends RowFormat
-
-case class RowFormatDelimited(
-    fieldsTerminatedBy: Option[String],
-    fieldsEscapedBy: Option[String],
-    mapKeysTerminatedBy: Option[String],
-    linesTerminatedBy: Option[String],
-    nullDefinedAs: Option[String])
-  extends RowFormat
-
-
-/**
- * File format for creating tables, specified with STORED AS [...].
- */
-sealed trait FileFormat
-
-case class TableFileFormat(
-    inputFormat: String,
-    outputFormat: String,
-    serde: Option[String] = None)
-  extends FileFormat
-
-case class GenericFileFormat(format: String) extends FileFormat
-
-
-/**
- * Storage handler for creating tables, specified with STORED BY [...].
- */
-case class StorageHandler(handlerClass: String, serdeProps: Map[String, String])
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 87198e0777587..0fa209b8feaae 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -188,56 +188,6 @@ class SparkSqlAstBuilder extends AstBuilder {
    */
   type TableHeader = (TableIdentifier, Boolean, Boolean, Boolean)
 
-  /**
-   * Create a [[CreateTable]] command.
-   *
-   * For example:
-   * {{{
-   *   CREATE [TEMPORARY] [EXTERNAL] TABLE [IF NOT EXISTS] [db_name.]table_name
-   *   [(col1 data_type [COMMENT col_comment], ...)]
-   *   [COMMENT table_comment]
-   *   [PARTITIONED BY (col3 data_type [COMMENT col_comment], ...)]
-   *   [CLUSTERED BY (col1, ...) [SORTED BY (col1 [ASC|DESC], ...)] INTO num_buckets BUCKETS]
-   *   [SKEWED BY (col1, col2, ...) ON ((col_value, col_value, ...), ...)
-   *   [STORED AS DIRECTORIES]
-   *   [ROW FORMAT row_format]
-   *   [STORED AS file_format | STORED BY storage_handler_class [WITH SERDEPROPERTIES (...)]]
-   *   [LOCATION path]
-   *   [TBLPROPERTIES (property_name=property_value, ...)]
-   *   [AS select_statement];
-   * }}}
-   */
-  override def visitCreateTable(ctx: CreateTableContext): LogicalPlan = withOrigin(ctx) {
-    val (name, temp, ifNotExists, external) = visitCreateTableHeader(ctx.createTableHeader)
-    val comment = Option(ctx.STRING).map(string)
-    val columns = Option(ctx.columns).map(visitColTypeList).getOrElse(Seq())
-    val partitionColumns = Option(ctx.partitionColumns).map(visitColTypeList).getOrElse(Seq())
-    val bucketSpec = Option(ctx.bucketSpec).map(visitBucketSpec)
-    val skewSpec = Option(ctx.skewSpec).map(visitSkewSpec)
-    val rowFormat = Option(ctx.rowFormat).map(visitRowFormat)
-    val fileFormat = Option(ctx.createFileFormat).map(_.fileFormat).map(visitFileFormat)
-    val storageHandler = Option(ctx.createFileFormat).map(_.storageHandler).map(visitStorageHandler)
-    val location = Option(ctx.locationSpec).map(_.STRING).map(string)
-    val properties = Option(ctx.tablePropertyList).map(visitTablePropertyList).getOrElse(Map())
-    val selectQuery = Option(ctx.query).map(plan)
-    CreateTable(
-      name,
-      temp,
-      ifNotExists,
-      external,
-      comment,
-      columns,
-      partitionColumns,
-      bucketSpec,
-      skewSpec,
-      rowFormat,
-      fileFormat,
-      storageHandler,
-      location,
-      properties,
-      selectQuery)
-  }
-
   /**
    * Validate a create table statement and return the [[TableIdentifier]].
    */
@@ -652,10 +602,22 @@ class SparkSqlAstBuilder extends AstBuilder {
    */
   override def visitSetTableFileFormat(
       ctx: SetTableFileFormatContext): LogicalPlan = withOrigin(ctx) {
+    // AlterTableSetFileFormat currently takes both a GenericFileFormat and a
+    // TableFileFormatContext. This is a bit weird because it should only take one. It also should
+    // use a CatalogFileFormat instead of either a String or a Sequence of Strings. We will address
+    // this in a follow-up PR.
+    val (fileFormat, genericFormat) = ctx.fileFormat match {
+      case s: GenericFileFormatContext =>
+        (Seq.empty[String], Option(s.identifier.getText))
+      case s: TableFileFormatContext =>
+        val elements = Seq(s.inFmt, s.outFmt) ++ Option(s.serdeCls).toSeq
+        (elements.map(string), None)
+    }
     AlterTableSetFileFormat(
       visitTableIdentifier(ctx.tableIdentifier),
       Option(ctx.partitionSpec).map(visitNonOptionalPartitionSpec),
-      visitFileFormat(ctx.fileFormat))(
+      fileFormat,
+      genericFormat)(
       command(ctx))
   }
 
@@ -827,72 +789,6 @@ class SparkSqlAstBuilder extends AstBuilder {
     SkewSpec(visitIdentifierList(ctx.identifierList), skewedValues, ctx.DIRECTORIES != null)
   }
 
-  /**
-   * Create a [[RowFormat]] used for creating tables.
-   *
-   * Example format:
-   * {{{
-   *   SERDE serde_name [WITH SERDEPROPERTIES (k1=v1, k2=v2, ...)]
-   * }}}
-   *
-   * OR
-   *
-   * {{{
-   *   DELIMITED [FIELDS TERMINATED BY char [ESCAPED BY char]]
-   *   [COLLECTION ITEMS TERMINATED BY char]
-   *   [MAP KEYS TERMINATED BY char]
-   *   [LINES TERMINATED BY char]
-   *   [NULL DEFINED AS char]
-   * }}}
-   */
-  private def visitRowFormat(ctx: RowFormatContext): RowFormat = withOrigin(ctx) {
-    ctx match {
-      case serde: RowFormatSerdeContext =>
-        RowFormatSerde(
-          string(serde.name),
-          visitTablePropertyList(serde.props))
-      case delimited: RowFormatDelimitedContext =>
-        RowFormatDelimited(
-          fieldsTerminatedBy = Option(delimited.fieldsTerminatedBy).map(string),
-          fieldsEscapedBy = Option(delimited.escapedBy).map(string),
-          mapKeysTerminatedBy = Option(delimited.keysTerminatedBy).map(string),
-          linesTerminatedBy = Option(delimited.linesSeparatedBy).map(string),
-          nullDefinedAs = Option(delimited.nullDefinedAs).map(string))
-    }
-  }
-
-  /**
-   * Create a [[FileFormat]] for creating and altering tables.
-   *
-   * Example format:
-   * {{{
-   *   SEQUENCEFILE |
-   *   TEXTFILE |
-   *   RCFILE |
-   *   ORC |
-   *   PARQUET |
-   *   AVRO |
-   *   INPUTFORMAT input_format_classname OUTPUTFORMAT output_format_classname
-   * }}}
-   */
-  private def visitFileFormat(ctx: FileFormatContext): FileFormat = withOrigin(ctx) {
-    ctx match {
-      case s: GenericFileFormatContext =>
-        GenericFileFormat(s.identifier.getText)
-      case s: TableFileFormatContext =>
-        TableFileFormat(string(s.inFmt), string(s.outFmt), Option(s.serdeCls).map(string))
-    }
-  }
-
-  /**
-   * Create a [[StorageHandler]] for creating tables.
-   */
-  override def visitStorageHandler(ctx: StorageHandlerContext): StorageHandler = withOrigin(ctx) {
-    val handlerClassName = string(ctx.STRING)
-    val serdeProps = Option(ctx.tablePropertyList).map(visitTablePropertyList).getOrElse(Map())
-    StorageHandler(handlerClassName, serdeProps)
-  }
-
   /**
    * Convert a nested constants list into a sequence of string sequences.
    */
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
index d14db2c6f8af2..f5bd8f806c531 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.execution.command
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.{AnalysisException, Row, SQLContext}
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, CatalogTable, FileFormat}
+import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, CatalogTable}
 import org.apache.spark.sql.catalyst.catalog.ExternalCatalog.TablePartitionSpec
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
 import org.apache.spark.sql.types._
@@ -331,7 +331,8 @@ case class AlterTableUnarchivePartition(
 case class AlterTableSetFileFormat(
     tableName: TableIdentifier,
     partitionSpec: Option[TablePartitionSpec],
-    fileFormat: FileFormat)(sql: String)
+    fileFormat: Seq[String],
+    genericFormat: Option[String])(sql: String)
   extends NativeDDLCommand(sql) with Logging
 
 /**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 2ed9a108d4ae8..a813fc38ac315 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -19,10 +19,7 @@ package org.apache.spark.sql.execution.command
 
 import org.apache.spark.sql.{Row, SQLContext}
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.catalog.{FileFormat, RowFormat, SkewSpec, StorageHandler}
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.execution.datasources.BucketSpec
-import org.apache.spark.sql.types.StructField
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
 
 
 // TODO: move the rest of the table commands from ddl.scala to this file
@@ -30,9 +27,12 @@ import org.apache.spark.sql.types.StructField
 /**
  * A command to create a table.
  *
+ * Note: This is currently used only for creating Hive tables.
+ * This is not intended for temporary tables.
+ *
  * The syntax of using this command in SQL is:
  * {{{
- *   CREATE [TEMPORARY] [EXTERNAL] TABLE [IF NOT EXISTS] [db_name.]table_name
+ *   CREATE [EXTERNAL] TABLE [IF NOT EXISTS] [db_name.]table_name
  *   [(col1 data_type [COMMENT col_comment], ...)]
  *   [COMMENT table_comment]
  *   [PARTITIONED BY (col3 data_type [COMMENT col_comment], ...)]
@@ -46,30 +46,16 @@ import org.apache.spark.sql.types.StructField
  *   [AS select_statement];
  * }}}
  */
-case class CreateTable(
-    name: TableIdentifier,
-    isTemp: Boolean,
-    ifNotExists: Boolean,
-    isExternal: Boolean,
-    comment: Option[String],
-    columns: Seq[StructField],
-    partitionedColumns: Seq[StructField],
-    bucketSpec: Option[BucketSpec],
-    skewSpec: Option[SkewSpec],
-    rowFormat: Option[RowFormat],
-    fileFormat: Option[FileFormat],
-    storageHandler: Option[StorageHandler],
-    location: Option[String],
-    properties: Map[String, String],
-    selectQuery: Option[LogicalPlan])
-  extends RunnableCommand {
+case class CreateTable(table: CatalogTable, ifNotExists: Boolean) extends RunnableCommand {
 
   override def run(sqlContext: SQLContext): Seq[Row] = {
+    // TODO: implement me
     Seq.empty[Row]
   }
 
 }
 
+
 /**
  * A command that renames a table/view.
  *
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
index eeb9a61d2c13b..fc95e620037f1 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
@@ -454,23 +454,37 @@ class DDLCommandSuite extends PlanTest {
   }
 
   test("alter table: set file format") {
-    val sql1 = "ALTER TABLE table_name SET FILEFORMAT INPUTFORMAT 'test' " +
+    val sql1 =
+      """
+        |ALTER TABLE table_name SET FILEFORMAT INPUTFORMAT 'test'
+        |OUTPUTFORMAT 'test' SERDE 'test' INPUTDRIVER 'test' OUTPUTDRIVER 'test'
+      """.stripMargin
+    val sql2 = "ALTER TABLE table_name SET FILEFORMAT INPUTFORMAT 'test' " +
       "OUTPUTFORMAT 'test' SERDE 'test'"
-    val sql2 = "ALTER TABLE table_name PARTITION (dt='2008-08-08', country='us') " +
+    val sql3 = "ALTER TABLE table_name PARTITION (dt='2008-08-08', country='us') " +
       "SET FILEFORMAT PARQUET"
     val parsed1 = parser.parsePlan(sql1)
     val parsed2 = parser.parsePlan(sql2)
+    val parsed3 = parser.parsePlan(sql3)
     val tableIdent = TableIdentifier("table_name", None)
     val expected1 = AlterTableSetFileFormat(
       tableIdent,
       None,
-      TableFileFormat("test", "test", Some("test")))(sql1)
+      List("test", "test", "test", "test", "test"),
+      None)(sql1)
     val expected2 = AlterTableSetFileFormat(
+      tableIdent,
+      None,
+      List("test", "test", "test"),
+      None)(sql2)
+    val expected3 = AlterTableSetFileFormat(
       tableIdent,
       Some(Map("dt" -> "2008-08-08", "country" -> "us")),
-      GenericFileFormat("PARQUET"))(sql2)
+      Seq(),
+      Some("PARQUET"))(sql3)
     comparePlans(parsed1, expected1)
     comparePlans(parsed2, expected2)
+    comparePlans(parsed3, expected3)
   }
 
   test("alter table: set location") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
index b3ccfa3ac4740..0507722f17b8a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
@@ -196,13 +196,6 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
 
   // TODO: test drop database in restrict mode
 
-  test("create table") {
-    val catalog = sqlContext.sessionState.catalog
-    val tableIdent1 = TableIdentifier("tab1", Some("dbx"))
-    createDatabase(catalog, "dbx")
-    sql("CREATE TABLE dbx.tab1 (id int, name string) partitioned by (nickname string)")
-  }
-
   test("alter table: rename") {
     val catalog = sqlContext.sessionState.catalog
     val tableIdent1 = TableIdentifier("tab1", Some("dbx"))
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index ca5ac494e31a6..a7d25c040477e 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -672,6 +672,7 @@ private[hive] class HiveClientImpl(
       case CatalogTableType.INDEX_TABLE => HiveTableType.INDEX_TABLE
       case CatalogTableType.VIRTUAL_VIEW => HiveTableType.VIRTUAL_VIEW
     })
+    // TODO: pass the comment to Hive somehow
     // Note: In Hive the schema and partition columns must be disjoint sets
     val (partCols, schema) = table.schema.map(toHiveColumn).partition { c =>
       table.partitionColumnNames.contains(c.getName)
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
index 585bc77a0fbe9..54f2b4b138e76 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
@@ -26,6 +26,7 @@ import org.apache.hadoop.hive.ql.session.SessionState
 import org.apache.hadoop.hive.serde.serdeConstants
 import org.apache.hadoop.hive.serde2.`lazy`.LazySimpleSerDe
 
+import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.analysis.UnresolvedGenerator
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.expressions._
@@ -33,6 +34,7 @@ import org.apache.spark.sql.catalyst.parser._
 import org.apache.spark.sql.catalyst.parser.SqlBaseParser._
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.execution.SparkSqlAstBuilder
+import org.apache.spark.sql.execution.command.CreateTable
 import org.apache.spark.sql.hive.{CreateTableAsSelect => CTAS, CreateViewAsSelect => CreateView}
 import org.apache.spark.sql.hive.{HiveGenericUDTF, HiveSerDe}
 import org.apache.spark.sql.hive.HiveShim.HiveFunctionWrapper
@@ -134,84 +136,117 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
   }
 
   /**
-   * Create a [[CatalogStorageFormat]]. This is part of the [[CreateTableAsSelect]] command.
+   * Create a [[CatalogStorageFormat]] for creating tables.
    */
   override def visitCreateFileFormat(
       ctx: CreateFileFormatContext): CatalogStorageFormat = withOrigin(ctx) {
-    if (ctx.storageHandler == null) {
-      typedVisit[CatalogStorageFormat](ctx.fileFormat)
-    } else {
-      throw new ParseException("Storage Handlers are currently unsupported.", ctx)
+    (ctx.fileFormat, ctx.storageHandler) match {
+      case (fileFormat, null) if fileFormat != null =>
+        fileFormat match {
+          // Expected format: INPUTFORMAT input_format OUTPUTFORMAT output_format
+          case c: TableFileFormatContext => visitTableFileFormat(c)
+          // Expected format: SEQUENCEFILE | TEXTFILE | RCFILE | ORC | PARQUET | AVRO
+          case c: GenericFileFormatContext => visitGenericFileFormat(c)
+        }
+      case (null, storageHandler) if storageHandler != null =>
+        throw new ParseException("Operation not allowed: ... STORED BY storage_handler ...", ctx)
+      case (null, null) =>
+        throw new ParseException("expected one of STORED AS or STORED BY", ctx)
+      case _ =>
+        throw new ParseException("expected either STORED AS or STORED BY, not both", ctx)
     }
   }
 
   /**
-   * Create a [[CreateTableAsSelect]] command.
+   * Create a table. TODO: expand this comment!
+   *
+   * For example:
+   * {{{
+   *   CREATE [TEMPORARY] [EXTERNAL] TABLE [IF NOT EXISTS] [db_name.]table_name
+   *   [(col1 data_type [COMMENT col_comment], ...)]
+   *   [COMMENT table_comment]
+   *   [PARTITIONED BY (col3 data_type [COMMENT col_comment], ...)]
+   *   [CLUSTERED BY (col1, ...) [SORTED BY (col1 [ASC|DESC], ...)] INTO num_buckets BUCKETS]
+   *   [SKEWED BY (col1, col2, ...) ON ((col_value, col_value, ...), ...)
+   *   [STORED AS DIRECTORIES]
+   *   [ROW FORMAT row_format]
+   *   [STORED AS file_format | STORED BY storage_handler_class [WITH SERDEPROPERTIES (...)]]
+   *   [LOCATION path]
+   *   [TBLPROPERTIES (property_name=property_value, ...)]
+   *   [AS select_statement];
+   * }}}
    */
   override def visitCreateTable(ctx: CreateTableContext): LogicalPlan = {
-    if (ctx.query == null) {
-      HiveNativeCommand(command(ctx))
+    val (name, temp, ifNotExists, external) = visitCreateTableHeader(ctx.createTableHeader)
+    // TODO: implement temporary tables
+    if (temp) {
+      throw new AnalysisException(
+        "CREATE TEMPORARY TABLE is not supported yet. " +
+        "Please use registerTempTable as an alternative.")
+    }
+    val tableType = if (external) {
+      CatalogTableType.EXTERNAL_TABLE
     } else {
-      // Get the table header.
-      val (table, temp, ifNotExists, external) = visitCreateTableHeader(ctx.createTableHeader)
-      val tableType = if (external) {
-        CatalogTableType.EXTERNAL_TABLE
-      } else {
-        CatalogTableType.MANAGED_TABLE
-      }
-
-      // Unsupported clauses.
-      if (temp) {
-        logWarning("TEMPORARY clause is ignored.")
-      }
-      if (ctx.bucketSpec != null) {
-        // TODO add this - we need cluster columns in the CatalogTable for this to work.
-        logWarning("CLUSTERED BY ... [ORDERED BY ...] INTO ... BUCKETS clause is ignored.")
-      }
-      if (ctx.skewSpec != null) {
-        logWarning("SKEWED BY ... ON ... [STORED AS DIRECTORIES] clause is ignored.")
-      }
-
-      // Get the column by which the table is partitioned.
-      val partitionCols = Option(ctx.partitionColumns).toSeq.flatMap(visitCatalogColumns(_))
-
-      // Note: Hive requires partition columns to be distinct from the schema, so we need
-      // to include the partition columns here explicitly
-      val schema =
-        Option(ctx.columns).toSeq.flatMap(visitCatalogColumns(_, _.toLowerCase)) ++ partitionCols
+      CatalogTableType.MANAGED_TABLE
+    }
+    val comment = Option(ctx.STRING).map(string)
+    val partitionCols = Option(ctx.partitionColumns).toSeq.flatMap(visitCatalogColumns(_))
+    val cols = Option(ctx.columns).toSeq.flatMap(visitCatalogColumns(_, _.toLowerCase))
+    val bucketSpec = Option(ctx.bucketSpec).map(visitBucketSpec)
+    val sortColNames = bucketSpec.map(_.sortColumnNames).getOrElse(Seq())
+    val bucketColNames = bucketSpec.map(_.bucketColumnNames).getOrElse(Seq())
+    val numBuckets = bucketSpec.map(_.numBuckets).getOrElse(0)
+    val skewSpec = Option(ctx.skewSpec).map(visitSkewSpec)
+    val skewedColNames = skewSpec.map(_.columns).getOrElse(Seq())
+    val skewedColValues = skewSpec.map(_.values).getOrElse(Seq())
+    val storedAsDirs = skewSpec.exists(_.storedAsDirs)
+    val properties = Option(ctx.tablePropertyList).map(visitTablePropertyList).getOrElse(Map.empty)
+    val selectQuery = Option(ctx.query).map(plan)
+
+    // Note: Hive requires partition columns to be distinct from the schema, so we need
+    // to include the partition columns here explicitly
+    val schema = cols ++ partitionCols
+
+    // Storage format
+    val defaultStorageType = hiveConf.getVar(HiveConf.ConfVars.HIVEDEFAULTFILEFORMAT)
+    val defaultHiveSerde = HiveSerDe.sourceToSerDe(defaultStorageType, hiveConf).getOrElse {
+      HiveSerDe(
+        inputFormat = Option("org.apache.hadoop.mapred.TextInputFormat"),
+        outputFormat = Option("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat"))
+    }
+    val fileStorage = Option(ctx.createFileFormat).map(visitCreateFileFormat)
+    val rowStorage = Option(ctx.rowFormat).map(visitRowFormat)
+    val location = Option(ctx.locationSpec).map(visitLocationSpec)
+    val storage = CatalogStorageFormat(
+      locationUri = location,
+      inputFormat = fileStorage.map(_.inputFormat).getOrElse(defaultHiveSerde.inputFormat),
+      outputFormat = fileStorage.map(_.outputFormat).getOrElse(defaultHiveSerde.outputFormat),
+      serde = rowStorage.map(_.serde)
+        .orElse(fileStorage.map(_.serde))
+        .getOrElse(defaultHiveSerde.serde),
+      serdeProperties =
+        rowStorage.map(_.serdeProperties).getOrElse(Map()) ++
+        fileStorage.map(_.serdeProperties).getOrElse(Map()),
+      storedAsDirs = storedAsDirs)
 
-      // Create the storage.
-      def format(fmt: ParserRuleContext): CatalogStorageFormat = {
-        Option(fmt).map(typedVisit[CatalogStorageFormat]).getOrElse(EmptyStorageFormat)
-      }
-      // Default storage.
-      val defaultStorageType = hiveConf.getVar(HiveConf.ConfVars.HIVEDEFAULTFILEFORMAT)
-      val hiveSerDe = HiveSerDe.sourceToSerDe(defaultStorageType, hiveConf).getOrElse {
-        HiveSerDe(
-          inputFormat = Option("org.apache.hadoop.mapred.TextInputFormat"),
-          outputFormat = Option("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat"))
-      }
-      // Defined storage.
-      val fileStorage = format(ctx.createFileFormat)
-      val rowStorage = format(ctx.rowFormat)
-      val storage = CatalogStorageFormat(
-        Option(ctx.locationSpec).map(visitLocationSpec),
-        fileStorage.inputFormat.orElse(hiveSerDe.inputFormat),
-        fileStorage.outputFormat.orElse(hiveSerDe.outputFormat),
-        rowStorage.serde.orElse(hiveSerDe.serde).orElse(fileStorage.serde),
-        rowStorage.serdeProperties ++ fileStorage.serdeProperties
-      )
+    val tableDesc = CatalogTable(
+      identifier = name,
+      tableType = tableType,
+      storage = storage,
+      schema = schema,
+      partitionColumnNames = partitionCols.map(_.name),
+      sortColumnNames = sortColNames,
+      bucketColumnNames = bucketColNames,
+      skewColumnNames = skewedColNames,
+      skewColumnValues = skewedColValues,
+      numBuckets = numBuckets,
+      properties = properties,
+      // TODO support the sql text - have a proper location for this!
+      comment = comment)
 
-      val tableDesc = CatalogTable(
-        identifier = table,
-        tableType = tableType,
-        schema = schema,
-        partitionColumnNames = partitionCols.map(_.name),
-        storage = storage,
-        properties = Option(ctx.tablePropertyList).map(visitTablePropertyList).getOrElse(Map.empty),
-        // TODO support the sql text - have a proper location for this!
-        viewText = Option(ctx.STRING).map(string))
-      CTAS(tableDesc, plan(ctx.query), ifNotExists)
+    selectQuery match {
+      case Some(q) => CTAS(tableDesc, q, ifNotExists)
+      case None => CreateTable(tableDesc, ifNotExists)
     }
   }
 
@@ -322,7 +357,7 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
 
       case c: RowFormatSerdeContext =>
         // Use a serde format.
-        val CatalogStorageFormat(None, None, None, Some(name), props) = visitRowFormatSerde(c)
+        val CatalogStorageFormat(None, None, None, Some(name), props, _) = visitRowFormatSerde(c)
 
         // SPARK-10310: Special cases LazySimpleSerDe
         val recordHandler = if (name == classOf[LazySimpleSerDe].getCanonicalName) {
@@ -377,7 +412,7 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
   }
 
   /**
-   * Resolve a [[HiveSerDe]] based on the format name given.
+   * Resolve a [[HiveSerDe]] based on the name given and return it as a [[CatalogStorageFormat]].
    */
   override def visitGenericFileFormat(
       ctx: GenericFileFormatContext): CatalogStorageFormat = withOrigin(ctx) {
@@ -393,6 +428,31 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
     }
   }
 
+  /**
+   * Create a [[RowFormat]] used for creating tables.
+   *
+   * Example format:
+   * {{{
+   *   SERDE serde_name [WITH SERDEPROPERTIES (k1=v1, k2=v2, ...)]
+   * }}}
+   *
+   * OR
+   *
+   * {{{
+   *   DELIMITED [FIELDS TERMINATED BY char [ESCAPED BY char]]
+   *   [COLLECTION ITEMS TERMINATED BY char]
+   *   [MAP KEYS TERMINATED BY char]
+   *   [LINES TERMINATED BY char]
+   *   [NULL DEFINED AS char]
+   * }}}
+   */
+  private def visitRowFormat(ctx: RowFormatContext): CatalogStorageFormat = withOrigin(ctx) {
+    ctx match {
+      case serde: RowFormatSerdeContext => visitRowFormatSerde(serde)
+      case delimited: RowFormatDelimitedContext => visitRowFormatDelimited(delimited)
+    }
+  }
+
   /**
    * Create SERDE row format name and properties pair.
    */

From 5e0fe03bfa655c6de854bc8adaa73186a17a0b0c Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Fri, 8 Apr 2016 23:52:20 -0700
Subject: [PATCH 04/18] Implement it

---
 .../scala/org/apache/spark/sql/execution/command/tables.scala   | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index a813fc38ac315..9c6030502dc22 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -49,7 +49,7 @@ import org.apache.spark.sql.catalyst.catalog.CatalogTable
 case class CreateTable(table: CatalogTable, ifNotExists: Boolean) extends RunnableCommand {
 
   override def run(sqlContext: SQLContext): Seq[Row] = {
-    // TODO: implement me
+    sqlContext.sessionState.catalog.createTable(table, ifNotExists)
     Seq.empty[Row]
   }
 

From f7501d9ebc5c4f08374788a937de6a56689258b8 Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Sat, 9 Apr 2016 00:00:30 -0700
Subject: [PATCH 05/18] Revert unnecessary changes (small)

---
 .../org/apache/spark/sql/execution/SparkSqlParser.scala      | 2 +-
 .../apache/spark/sql/execution/command/DDLCommandSuite.scala | 5 ++---
 2 files changed, 3 insertions(+), 4 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 0fa209b8feaae..6fac8926529d8 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -20,7 +20,7 @@ import scala.collection.JavaConverters._
 
 import org.apache.spark.sql.{AnalysisException, SaveMode}
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.catalog._
+import org.apache.spark.sql.catalyst.catalog.SkewSpec
 import org.apache.spark.sql.catalyst.parser._
 import org.apache.spark.sql.catalyst.parser.SqlBaseParser._
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, OneRowRelation}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
index fc95e620037f1..2fb2605967b29 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
@@ -18,7 +18,6 @@
 package org.apache.spark.sql.execution.command
 
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.catalyst.plans.PlanTest
 import org.apache.spark.sql.catalyst.plans.logical.Project
@@ -456,8 +455,8 @@ class DDLCommandSuite extends PlanTest {
   test("alter table: set file format") {
     val sql1 =
       """
-        |ALTER TABLE table_name SET FILEFORMAT INPUTFORMAT 'test'
-        |OUTPUTFORMAT 'test' SERDE 'test' INPUTDRIVER 'test' OUTPUTDRIVER 'test'
+       |ALTER TABLE table_name SET FILEFORMAT INPUTFORMAT 'test'
+       |OUTPUTFORMAT 'test' SERDE 'test' INPUTDRIVER 'test' OUTPUTDRIVER 'test'
       """.stripMargin
     val sql2 = "ALTER TABLE table_name SET FILEFORMAT INPUTFORMAT 'test' " +
       "OUTPUTFORMAT 'test' SERDE 'test'"

From 3af954d355c3dc3c5fb982d3bcdf2a0a3e3c4580 Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Mon, 11 Apr 2016 15:37:00 -0700
Subject: [PATCH 06/18] Address comment

---
 .../sql/catalyst/catalog/interface.scala      | 23 +-----
 .../spark/sql/execution/SparkSqlParser.scala  | 16 ----
 .../execution/command/DDLCommandSuite.scala   | 20 +----
 .../sql/hive/client/HiveClientImpl.scala      |  6 +-
 .../sql/hive/execution/HiveSqlParser.scala    | 78 +++++++++----------
 .../spark/sql/hive/HiveDDLCommandSuite.scala  | 25 +++---
 6 files changed, 51 insertions(+), 117 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
index 809b29718458f..4601502e50e84 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
@@ -183,8 +183,7 @@ case class CatalogStorageFormat(
     inputFormat: Option[String],
     outputFormat: Option[String],
     serde: Option[String],
-    serdeProperties: Map[String, String],
-    storedAsDirs: Boolean = false)
+    serdeProperties: Map[String, String])
 
 
 /**
@@ -224,10 +223,6 @@ case class CatalogTable(
     partitionColumnNames: Seq[String] = Seq.empty,
     sortColumnNames: Seq[String] = Seq.empty,
     bucketColumnNames: Seq[String] = Seq.empty,
-    // e.g. (date, country)
-    skewColumnNames: Seq[String] = Seq.empty,
-    // e.g. ('2008-08-08', 'us), ('2009-09-09', 'uk')
-    skewColumnValues: Seq[Seq[String]] = Seq.empty,
     numBuckets: Int = 0,
     createTime: Long = System.currentTimeMillis,
     lastAccessTime: Long = System.currentTimeMillis,
@@ -245,7 +240,6 @@ case class CatalogTable(
   requireSubsetOfSchema(partitionColumnNames, "partition")
   requireSubsetOfSchema(sortColumnNames, "sort")
   requireSubsetOfSchema(bucketColumnNames, "bucket")
-  requireSubsetOfSchema(skewColumnNames, "skew")
 
   /** Columns this table is partitioned by. */
   def partitionColumns: Seq[CatalogColumn] =
@@ -315,18 +309,3 @@ case class CatalogRelation(
   require(metadata.identifier.database == Some(db),
     "provided database does not match the one specified in the table definition")
 }
-
-
-/**
- * Skew specifications for a table.
- */
-case class SkewSpec(
-    // e.g. ['datetime', 'country']
-    columns: Seq[String],
-    // e.g. [['2008-08-08', 'us], ['2009-09-09', 'uk']]
-    values: Seq[Seq[String]],
-    storedAsDirs: Boolean) {
-
-  require(values.forall(_.size == columns.size),
-    "number of columns in skewed values do not match number of skewed columns provided")
-}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 6dbdafa3c00a4..6f368d840c462 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -20,7 +20,6 @@ import scala.collection.JavaConverters._
 
 import org.apache.spark.sql.{AnalysisException, SaveMode}
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.catalog.SkewSpec
 import org.apache.spark.sql.catalyst.parser._
 import org.apache.spark.sql.catalyst.parser.SqlBaseParser._
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, OneRowRelation}
@@ -787,21 +786,6 @@ class SparkSqlAstBuilder extends AstBuilder {
         .map(_.identifier.getText))
   }
 
-  /**
-   * Create a skew specification. This contains three components:
-   * - The Skewed Columns
-   * - Values for which are skewed. The size of each entry must match the number of skewed columns.
-   * - A store in directory flag.
-   */
-  override def visitSkewSpec(ctx: SkewSpecContext): SkewSpec = withOrigin(ctx) {
-    val skewedValues = if (ctx.constantList != null) {
-      Seq(visitConstantList(ctx.constantList))
-    } else {
-      visitNestedConstantList(ctx.nestedConstantList)
-    }
-    SkewSpec(visitIdentifierList(ctx.identifierList), skewedValues, ctx.DIRECTORIES != null)
-  }
-
   /**
    * Convert a nested constants list into a sequence of string sequences.
    */
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
index ac69518ddf689..da9db9f5974ed 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
@@ -453,37 +453,25 @@ class DDLCommandSuite extends PlanTest {
   }
 
   test("alter table: set file format") {
-    val sql1 =
-      """
-       |ALTER TABLE table_name SET FILEFORMAT INPUTFORMAT 'test'
-       |OUTPUTFORMAT 'test' SERDE 'test' INPUTDRIVER 'test' OUTPUTDRIVER 'test'
-      """.stripMargin
-    val sql2 = "ALTER TABLE table_name SET FILEFORMAT INPUTFORMAT 'test' " +
+    val sql1 = "ALTER TABLE table_name SET FILEFORMAT INPUTFORMAT 'test' " +
       "OUTPUTFORMAT 'test' SERDE 'test'"
-    val sql3 = "ALTER TABLE table_name PARTITION (dt='2008-08-08', country='us') " +
+    val sql2 = "ALTER TABLE table_name PARTITION (dt='2008-08-08', country='us') " +
       "SET FILEFORMAT PARQUET"
     val parsed1 = parser.parsePlan(sql1)
     val parsed2 = parser.parsePlan(sql2)
-    val parsed3 = parser.parsePlan(sql3)
     val tableIdent = TableIdentifier("table_name", None)
     val expected1 = AlterTableSetFileFormat(
       tableIdent,
       None,
-      List("test", "test", "test", "test", "test"),
+      List("test", "test", "test"),
       None)(sql1)
     val expected2 = AlterTableSetFileFormat(
-      tableIdent,
-      None,
-      List("test", "test", "test"),
-      None)(sql2)
-    val expected3 = AlterTableSetFileFormat(
       tableIdent,
       Some(Map("dt" -> "2008-08-08", "country" -> "us")),
       Seq(),
-      Some("PARQUET"))(sql3)
+      Some("PARQUET"))(sql2)
     comparePlans(parsed1, expected1)
     comparePlans(parsed2, expected2)
-    comparePlans(parsed3, expected3)
   }
 
   test("alter table: set location") {
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index cc04c710f4c47..30565db644420 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -315,8 +315,6 @@ private[hive] class HiveClientImpl(
         partitionColumnNames = partCols.map(_.name),
         sortColumnNames = Seq(), // TODO: populate this
         bucketColumnNames = h.getBucketCols.asScala,
-        skewColumnNames = h.getSkewedColNames.asScala,
-        skewColumnValues = h.getSkewedColValues.asScala.map(_.asScala),
         numBuckets = h.getNumBuckets,
         createTime = h.getTTable.getCreateTime.toLong * 1000,
         lastAccessTime = h.getLastAccessTime.toLong * 1000,
@@ -681,7 +679,6 @@ private[hive] class HiveClientImpl(
       case CatalogTableType.INDEX_TABLE => HiveTableType.INDEX_TABLE
       case CatalogTableType.VIRTUAL_VIEW => HiveTableType.VIRTUAL_VIEW
     })
-    // TODO: pass the comment to Hive somehow
     // Note: In Hive the schema and partition columns must be disjoint sets
     val (partCols, schema) = table.schema.map(toHiveColumn).partition { c =>
       table.partitionColumnNames.contains(c.getName)
@@ -690,8 +687,6 @@ private[hive] class HiveClientImpl(
     hiveTable.setPartCols(partCols.asJava)
     // TODO: set sort columns here too
     hiveTable.setBucketCols(table.bucketColumnNames.asJava)
-    hiveTable.setSkewedColNames(table.skewColumnNames.asJava)
-    hiveTable.setSkewedColValues(table.skewColumnValues.map(_.asJava).asJava)
     hiveTable.setOwner(conf.getUser)
     hiveTable.setNumBuckets(table.numBuckets)
     hiveTable.setCreateTime((table.createTime / 1000).toInt)
@@ -702,6 +697,7 @@ private[hive] class HiveClientImpl(
     table.storage.serde.foreach(hiveTable.setSerializationLib)
     table.storage.serdeProperties.foreach { case (k, v) => hiveTable.setSerdeParam(k, v) }
     table.properties.foreach { case (k, v) => hiveTable.setProperty(k, v) }
+    table.comment.foreach { c => hiveTable.setProperty("comment", c) }
     table.viewOriginalText.foreach { t => hiveTable.setViewOriginalText(t) }
     table.viewText.foreach { t => hiveTable.setViewExpandedText(t) }
     hiveTable
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
index 28589eec046f0..d6faa90d47a74 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
@@ -128,17 +128,14 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
   override def visitCreateFileFormat(
       ctx: CreateFileFormatContext): CatalogStorageFormat = withOrigin(ctx) {
     (ctx.fileFormat, ctx.storageHandler) match {
-      case (fileFormat, null) if fileFormat != null =>
-        fileFormat match {
-          // Expected format: INPUTFORMAT input_format OUTPUTFORMAT output_format
-          case c: TableFileFormatContext => visitTableFileFormat(c)
-          // Expected format: SEQUENCEFILE | TEXTFILE | RCFILE | ORC | PARQUET | AVRO
-          case c: GenericFileFormatContext => visitGenericFileFormat(c)
-        }
-      case (null, storageHandler) if storageHandler != null =>
+      // Expected format: INPUTFORMAT input_format OUTPUTFORMAT output_format
+      case (c: TableFileFormatContext, null) =>
+          visitTableFileFormat(c)
+      // Expected format: SEQUENCEFILE | TEXTFILE | RCFILE | ORC | PARQUET | AVRO
+      case (c: GenericFileFormatContext, null) =>
+        visitGenericFileFormat(c)
+      case (null, storageHandler) =>
         throw new ParseException("Operation not allowed: ... STORED BY storage_handler ...", ctx)
-      case (null, null) =>
-        throw new ParseException("expected one of STORED AS or STORED BY", ctx)
       case _ =>
         throw new ParseException("expected either STORED AS or STORED BY, not both", ctx)
     }
@@ -163,13 +160,19 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
    *   [AS select_statement];
    * }}}
    */
-  override def visitCreateTable(ctx: CreateTableContext): LogicalPlan = {
+  override def visitCreateTable(ctx: CreateTableContext): LogicalPlan = withOrigin(ctx) {
     val (name, temp, ifNotExists, external) = visitCreateTableHeader(ctx.createTableHeader)
     // TODO: implement temporary tables
     if (temp) {
-      throw new AnalysisException(
+      throw new ParseException(
         "CREATE TEMPORARY TABLE is not supported yet. " +
-        "Please use registerTempTable as an alternative.")
+        "Please use registerTempTable as an alternative.", ctx)
+    }
+    if (ctx.skewSpec != null) {
+      throw new ParseException("Operation not allowed: CREATE TABLE ... SKEWED BY ...", ctx)
+    }
+    if (ctx.bucketSpec != null) {
+      throw new ParseException("Operation not allowed: CREATE TABLE ... CLUSTERED BY ...", ctx)
     }
     val tableType = if (external) {
       CatalogTableType.EXTERNAL_TABLE
@@ -179,14 +182,6 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
     val comment = Option(ctx.STRING).map(string)
     val partitionCols = Option(ctx.partitionColumns).toSeq.flatMap(visitCatalogColumns(_))
     val cols = Option(ctx.columns).toSeq.flatMap(visitCatalogColumns(_, _.toLowerCase))
-    val bucketSpec = Option(ctx.bucketSpec).map(visitBucketSpec)
-    val sortColNames = bucketSpec.map(_.sortColumnNames).getOrElse(Seq())
-    val bucketColNames = bucketSpec.map(_.bucketColumnNames).getOrElse(Seq())
-    val numBuckets = bucketSpec.map(_.numBuckets).getOrElse(0)
-    val skewSpec = Option(ctx.skewSpec).map(visitSkewSpec)
-    val skewedColNames = skewSpec.map(_.columns).getOrElse(Seq())
-    val skewedColValues = skewSpec.map(_.values).getOrElse(Seq())
-    val storedAsDirs = skewSpec.exists(_.storedAsDirs)
     val properties = Option(ctx.tablePropertyList).map(visitTablePropertyList).getOrElse(Map.empty)
     val selectQuery = Option(ctx.query).map(plan)
 
@@ -195,40 +190,37 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
     val schema = cols ++ partitionCols
 
     // Storage format
-    val defaultStorageType = hiveConf.getVar(HiveConf.ConfVars.HIVEDEFAULTFILEFORMAT)
-    val defaultHiveSerde = HiveSerDe.sourceToSerDe(defaultStorageType, hiveConf).getOrElse {
-      HiveSerDe(
-        inputFormat = Option("org.apache.hadoop.mapred.TextInputFormat"),
-        outputFormat = Option("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat"))
+    val defaultStorage: CatalogStorageFormat = {
+      val defaultStorageType = hiveConf.getVar(HiveConf.ConfVars.HIVEDEFAULTFILEFORMAT)
+      val defaultHiveSerde = HiveSerDe.sourceToSerDe(defaultStorageType, hiveConf)
+      CatalogStorageFormat(
+        locationUri = None,
+        inputFormat = defaultHiveSerde.flatMap(_.inputFormat)
+          .orElse(Option("org.apache.hadoop.mapred.TextInputFormat")),
+        outputFormat = defaultHiveSerde.flatMap(_.outputFormat)
+          .orElse(Option("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat")),
+        serde = defaultHiveSerde.flatMap(_.serde),
+        serdeProperties = Map())
     }
     val fileStorage = Option(ctx.createFileFormat).map(visitCreateFileFormat)
-    val rowStorage = Option(ctx.rowFormat).map(visitRowFormat)
+      .getOrElse(EmptyStorageFormat)
+    val rowStorage = Option(ctx.rowFormat).map(visitRowFormat).getOrElse(EmptyStorageFormat)
     val location = Option(ctx.locationSpec).map(visitLocationSpec)
     val storage = CatalogStorageFormat(
       locationUri = location,
-      inputFormat = fileStorage.map(_.inputFormat).getOrElse(defaultHiveSerde.inputFormat),
-      outputFormat = fileStorage.map(_.outputFormat).getOrElse(defaultHiveSerde.outputFormat),
-      serde = rowStorage.map(_.serde)
-        .orElse(fileStorage.map(_.serde))
-        .getOrElse(defaultHiveSerde.serde),
-      serdeProperties =
-        rowStorage.map(_.serdeProperties).getOrElse(Map()) ++
-        fileStorage.map(_.serdeProperties).getOrElse(Map()),
-      storedAsDirs = storedAsDirs)
+      inputFormat = fileStorage.inputFormat.orElse(defaultStorage.inputFormat),
+      outputFormat = fileStorage.outputFormat.orElse(defaultStorage.outputFormat),
+      serde = rowStorage.serde.orElse(fileStorage.serde).orElse(defaultStorage.serde),
+      serdeProperties = rowStorage.serdeProperties ++ fileStorage.serdeProperties)
 
+    // TODO support the sql text - have a proper location for this!
     val tableDesc = CatalogTable(
       identifier = name,
       tableType = tableType,
       storage = storage,
       schema = schema,
       partitionColumnNames = partitionCols.map(_.name),
-      sortColumnNames = sortColNames,
-      bucketColumnNames = bucketColNames,
-      skewColumnNames = skewedColNames,
-      skewColumnValues = skewedColValues,
-      numBuckets = numBuckets,
       properties = properties,
-      // TODO support the sql text - have a proper location for this!
       comment = comment)
 
     selectQuery match {
@@ -345,7 +337,7 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
 
       case c: RowFormatSerdeContext =>
         // Use a serde format.
-        val CatalogStorageFormat(None, None, None, Some(name), props, _) = visitRowFormatSerde(c)
+        val CatalogStorageFormat(None, None, None, Some(name), props) = visitRowFormatSerde(c)
 
         // SPARK-10310: Special cases LazySimpleSerDe
         val recordHandler = if (name == classOf[LazySimpleSerDe].getCanonicalName) {
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
index a144da49976ec..32ff4dffe5410 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
@@ -69,9 +69,12 @@ class HiveDDLCommandSuite extends PlanTest {
       CatalogColumn("page_url", "string") ::
       CatalogColumn("referrer_url", "string") ::
       CatalogColumn("ip", "string", comment = Some("IP Address of the User")) ::
-      CatalogColumn("country", "string", comment = Some("country of origination")) :: Nil)
+      CatalogColumn("country", "string", comment = Some("country of origination")) ::
+      CatalogColumn("dt", "string", comment = Some("date type")) ::
+      CatalogColumn("hour", "string", comment = Some("hour of the day")) :: Nil)
+    assert(desc.comment == Some("This is the staging page view table"))
     // TODO will be SQLText
-    assert(desc.viewText == Option("This is the staging page view table"))
+    assert(desc.viewText.isEmpty)
     assert(desc.viewOriginalText.isEmpty)
     assert(desc.partitionColumns ==
       CatalogColumn("dt", "string", comment = Some("date type")) ::
@@ -116,9 +119,12 @@ class HiveDDLCommandSuite extends PlanTest {
       CatalogColumn("page_url", "string") ::
       CatalogColumn("referrer_url", "string") ::
       CatalogColumn("ip", "string", comment = Some("IP Address of the User")) ::
-      CatalogColumn("country", "string", comment = Some("country of origination")) :: Nil)
+      CatalogColumn("country", "string", comment = Some("country of origination")) ::
+      CatalogColumn("dt", "string", comment = Some("date type")) ::
+      CatalogColumn("hour", "string", comment = Some("hour of the day")) :: Nil)
     // TODO will be SQLText
-    assert(desc.viewText == Option("This is the staging page view table"))
+    assert(desc.comment == Some("This is the staging page view table"))
+    assert(desc.viewText.isEmpty)
     assert(desc.viewOriginalText.isEmpty)
     assert(desc.partitionColumns ==
       CatalogColumn("dt", "string", comment = Some("date type")) ::
@@ -196,17 +202,6 @@ class HiveDDLCommandSuite extends PlanTest {
           |AS SELECT key, value FROM src ORDER BY key, value
         """.stripMargin)
     }
-    intercept[ParseException] {
-      parser.parsePlan(
-        """CREATE TABLE ctas2
-          |STORED AS
-          |INPUTFORMAT "org.apache.hadoop.mapred.TextInputFormat"
-          |OUTPUTFORMAT "org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat"
-          |INPUTDRIVER "org.apache.hadoop.hive.howl.rcfile.RCFileInputDriver"
-          |OUTPUTDRIVER "org.apache.hadoop.hive.howl.rcfile.RCFileOutputDriver"
-          |AS SELECT key, value FROM src ORDER BY key, value
-        """.stripMargin)
-    }
     intercept[ParseException] {
       parser.parsePlan(
         """

From 2e95ecf790dc5d5b12b6ec72c0bd2b4bca99b17d Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Mon, 11 Apr 2016 16:41:57 -0700
Subject: [PATCH 07/18] Add all the tests

---
 .../sql/hive/execution/HiveSqlParser.scala    |  13 +-
 .../spark/sql/hive/HiveDDLCommandSuite.scala  | 190 ++++++++++++++++++
 2 files changed, 199 insertions(+), 4 deletions(-)

diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
index d6faa90d47a74..98a3ba02125f0 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
@@ -142,17 +142,22 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
   }
 
   /**
-   * Create a table. TODO: expand this comment!
+   * Create a table, returning either a [[CreateTable]] or a [[CreateTableAsSelect]].
    *
-   * For example:
+   * This is not used to create datasource tables, which is handled through
+   * "CREATE TABLE ... USING ...".
+   *
+   * Note: several features are currently not supported - temporary tables, bucketing,
+   * skewed columns and storage handlers (STORED BY).
+   *
+   * Expected format:
    * {{{
    *   CREATE [TEMPORARY] [EXTERNAL] TABLE [IF NOT EXISTS] [db_name.]table_name
    *   [(col1 data_type [COMMENT col_comment], ...)]
    *   [COMMENT table_comment]
    *   [PARTITIONED BY (col3 data_type [COMMENT col_comment], ...)]
    *   [CLUSTERED BY (col1, ...) [SORTED BY (col1 [ASC|DESC], ...)] INTO num_buckets BUCKETS]
-   *   [SKEWED BY (col1, col2, ...) ON ((col_value, col_value, ...), ...)
-   *   [STORED AS DIRECTORIES]
+   *   [SKEWED BY (col1, col2, ...) ON ((col_value, col_value, ...), ...) [STORED AS DIRECTORIES]]
    *   [ROW FORMAT row_format]
    *   [STORED AS file_format | STORED BY storage_handler_class [WITH SERDEPROPERTIES (...)]]
    *   [LOCATION path]
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
index 32ff4dffe5410..bd8d736b761bb 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
@@ -29,6 +29,7 @@ import org.apache.spark.sql.catalyst.expressions.JsonTuple
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.catalyst.plans.PlanTest
 import org.apache.spark.sql.catalyst.plans.logical.{Generate, ScriptTransformation}
+import org.apache.spark.sql.execution.command.CreateTable
 import org.apache.spark.sql.hive.execution.{HiveNativeCommand, HiveSqlParser}
 
 class HiveDDLCommandSuite extends PlanTest {
@@ -36,6 +37,7 @@ class HiveDDLCommandSuite extends PlanTest {
 
   private def extractTableDesc(sql: String): (CatalogTable, Boolean) = {
     parser.parsePlan(sql).collect {
+      case CreateTable(desc, allowExisting) => (desc, allowExisting)
       case CreateTableAsSelect(desc, _, allowExisting) => (desc, allowExisting)
       case CreateViewAsSelect(desc, _, allowExisting, _, _) => (desc, allowExisting)
     }.head
@@ -312,6 +314,194 @@ class HiveDDLCommandSuite extends PlanTest {
       """.stripMargin)
   }
 
+  test("create table - basic") {
+    val query = "CREATE TABLE my_table (id int, name string)"
+    val (desc, allowExisting) = extractTableDesc(query)
+    assert(!allowExisting)
+    assert(desc.identifier.database.isEmpty)
+    assert(desc.identifier.table == "my_table")
+    assert(desc.tableType == CatalogTableType.MANAGED_TABLE)
+    assert(desc.schema == Seq(CatalogColumn("id", "int"), CatalogColumn("name", "string")))
+    assert(desc.partitionColumnNames.isEmpty)
+    assert(desc.sortColumnNames.isEmpty)
+    assert(desc.bucketColumnNames.isEmpty)
+    assert(desc.numBuckets == 0)
+    assert(desc.viewText.isEmpty)
+    assert(desc.viewOriginalText.isEmpty)
+    assert(desc.storage.locationUri.isEmpty)
+    assert(desc.storage.inputFormat ==
+      Some("org.apache.hadoop.mapred.TextInputFormat"))
+    assert(desc.storage.outputFormat ==
+      Some("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat"))
+    assert(desc.storage.serde.isEmpty)
+    assert(desc.storage.serdeProperties.isEmpty)
+    assert(desc.properties.isEmpty)
+    assert(desc.comment.isEmpty)
+  }
+
+  test("create table - with database name") {
+    val query = "CREATE TABLE dbx.my_table (id int, name string)"
+    val (desc, _) = extractTableDesc(query)
+    assert(desc.identifier.database == Some("dbx"))
+    assert(desc.identifier.table == "my_table")
+  }
+
+  test("create table - temporary") {
+    val query = "CREATE TEMPORARY TABLE tab1 (id int, name string)"
+    val e = intercept[ParseException] { parser.parsePlan(query) }
+    assert(e.message.contains("registerTempTable"))
+  }
+
+  test("create table - external") {
+    val query = "CREATE EXTERNAL TABLE tab1 (id int, name string)"
+    val (desc, _) = extractTableDesc(query)
+    assert(desc.tableType == CatalogTableType.EXTERNAL_TABLE)
+  }
+
+  test("create table - if not exists") {
+    val query = "CREATE TABLE IF NOT EXISTS tab1 (id int, name string)"
+    val (_, allowExisting) = extractTableDesc(query)
+    assert(allowExisting)
+  }
+
+  test("create table - comment") {
+    val query = "CREATE TABLE my_table (id int, name string) COMMENT 'its hot as hell below'"
+    val (desc, _) = extractTableDesc(query)
+    assert(desc.comment == Some("its hot as hell below"))
+  }
+
+  test("create table - partitioned columns") {
+    val query = "CREATE TABLE my_table (id int, name string) PARTITIONED BY (month int)"
+    val (desc, _) = extractTableDesc(query)
+    assert(desc.schema == Seq(
+      CatalogColumn("id", "int"),
+      CatalogColumn("name", "string"),
+      CatalogColumn("month", "int")))
+    assert(desc.partitionColumnNames == Seq("month"))
+  }
+
+  test("create table - clustered by") {
+    val baseQuery = "CREATE TABLE my_table (id int, name string) CLUSTERED BY(id)"
+    val query1 = s"$baseQuery INTO 10 BUCKETS"
+    val query2 = s"$baseQuery SORTED BY(id) INTO 10 BUCKETS"
+    val e1 = intercept[ParseException] { parser.parsePlan(query1) }
+    val e2 = intercept[ParseException] { parser.parsePlan(query2) }
+    assert(e1.getMessage.contains("Operation not allowed"))
+    assert(e2.getMessage.contains("Operation not allowed"))
+  }
+
+  test("create table - skewed by") {
+    val baseQuery = "CREATE TABLE my_table (id int, name string) SKEWED BY"
+    val query1 = s"$baseQuery(id) ON (1, 10, 100)"
+    val query2 = s"$baseQuery(id, name) ON ((1, 'x'), (2, 'y'), (3, 'z'))"
+    val query3 = s"$baseQuery(id, name) ON ((1, 'x'), (2, 'y'), (3, 'z')) STORED AS DIRECTORIES"
+    val e1 = intercept[ParseException] { parser.parsePlan(query1) }
+    val e2 = intercept[ParseException] { parser.parsePlan(query2) }
+    val e3 = intercept[ParseException] { parser.parsePlan(query3) }
+    assert(e1.getMessage.contains("Operation not allowed"))
+    assert(e2.getMessage.contains("Operation not allowed"))
+    assert(e3.getMessage.contains("Operation not allowed"))
+  }
+
+  test("create table - row format") {
+    val baseQuery = "CREATE TABLE my_table (id int, name string) ROW FORMAT"
+    val query1 = s"$baseQuery SERDE 'org.apache.poof.serde.Baff'"
+    val query2 = s"$baseQuery SERDE 'org.apache.poof.serde.Baff' WITH SERDEPROPERTIES ('k1'='v1')"
+    val query3 =
+      s"""
+        |$baseQuery DELIMITED FIELDS TERMINATED BY 'x' ESCAPED BY 'y'
+        |COLLECTION ITEMS TERMINATED BY 'a'
+        |MAP KEYS TERMINATED BY 'b'
+        |LINES TERMINATED BY '\n'
+        |NULL DEFINED AS 'c'
+      """.stripMargin
+    val (desc1, _) = extractTableDesc(query1)
+    val (desc2, _) = extractTableDesc(query2)
+    val (desc3, _) = extractTableDesc(query3)
+    assert(desc1.storage.serde == Some("org.apache.poof.serde.Baff"))
+    assert(desc1.storage.serdeProperties.isEmpty)
+    assert(desc2.storage.serde == Some("org.apache.poof.serde.Baff"))
+    assert(desc2.storage.serdeProperties == Map("k1" -> "v1"))
+    assert(desc3.storage.serdeProperties == Map(
+      "field.delim" -> "x",
+      "escape.delim" -> "y",
+      "serialization.format" -> "x",
+      "line.delim" -> "\n",
+      "colelction.delim" -> "a", // yes, it's a typo from Hive :)
+      "mapkey.delim" -> "b"))
+  }
+
+  test("create table - file format") {
+    val baseQuery = "CREATE TABLE my_table (id int, name string) STORED AS"
+    val query1 = s"$baseQuery INPUTFORMAT 'winput' OUTPUTFORMAT 'wowput'"
+    val query2 = s"$baseQuery ORC"
+    val (desc1, _) = extractTableDesc(query1)
+    val (desc2, _) = extractTableDesc(query2)
+    assert(desc1.storage.inputFormat == Some("winput"))
+    assert(desc1.storage.outputFormat == Some("wowput"))
+    assert(desc1.storage.serde.isEmpty)
+    assert(desc2.storage.inputFormat == Some("org.apache.hadoop.hive.ql.io.orc.OrcInputFormat"))
+    assert(desc2.storage.outputFormat == Some("org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat"))
+    assert(desc2.storage.serde == Some("org.apache.hadoop.hive.ql.io.orc.OrcSerde"))
+  }
+
+  test("create table - storage handler") {
+    val baseQuery = "CREATE TABLE my_table (id int, name string) STORED BY"
+    val query1 = s"$baseQuery 'org.papachi.StorageHandler'"
+    val query2 = s"$baseQuery 'org.mamachi.StorageHandler' WITH SERDEPROPERTIES ('k1'='v1')"
+    val e1 = intercept[ParseException] { parser.parsePlan(query1) }
+    val e2 = intercept[ParseException] { parser.parsePlan(query2) }
+    assert(e1.getMessage.contains("Operation not allowed"))
+    assert(e2.getMessage.contains("Operation not allowed"))
+  }
+
+  test("create table - location") {
+    val query = "CREATE TABLE my_table (id int, name string) LOCATION '/path/to/mars'"
+    val (desc, _) = extractTableDesc(query)
+    assert(desc.storage.locationUri == Some("/path/to/mars"))
+  }
+
+  test("create table - properties") {
+    val query = "CREATE TABLE my_table (id int, name string) TBLPROPERTIES ('k1'='v1', 'k2'='v2')"
+    val (desc, _) = extractTableDesc(query)
+    assert(desc.properties == Map("k1" -> "v1", "k2" -> "v2"))
+  }
+
+  test("create table - everything!") {
+    val query =
+      """
+        |CREATE EXTERNAL TABLE IF NOT EXISTS dbx.my_table (id int, name string)
+        |COMMENT 'no comment'
+        |PARTITIONED BY (month int)
+        |ROW FORMAT SERDE 'org.apache.poof.serde.Baff' WITH SERDEPROPERTIES ('k1'='v1')
+        |STORED AS INPUTFORMAT 'winput' OUTPUTFORMAT 'wowput'
+        |LOCATION '/path/to/mercury'
+        |TBLPROPERTIES ('k1'='v1', 'k2'='v2')
+      """.stripMargin
+    val (desc, allowExisting) = extractTableDesc(query)
+    assert(allowExisting)
+    assert(desc.identifier.database == Some("dbx"))
+    assert(desc.identifier.table == "my_table")
+    assert(desc.tableType == CatalogTableType.EXTERNAL_TABLE)
+    assert(desc.schema == Seq(
+      CatalogColumn("id", "int"),
+      CatalogColumn("name", "string"),
+      CatalogColumn("month", "int")))
+    assert(desc.partitionColumnNames == Seq("month"))
+    assert(desc.sortColumnNames.isEmpty)
+    assert(desc.bucketColumnNames.isEmpty)
+    assert(desc.numBuckets == 0)
+    assert(desc.viewText.isEmpty)
+    assert(desc.viewOriginalText.isEmpty)
+    assert(desc.storage.locationUri == Some("/path/to/mercury"))
+    assert(desc.storage.inputFormat == Some("winput"))
+    assert(desc.storage.outputFormat == Some("wowput"))
+    assert(desc.storage.serde == Some("org.apache.poof.serde.Baff"))
+    assert(desc.storage.serdeProperties == Map("k1" -> "v1"))
+    assert(desc.properties == Map("k1" -> "v1", "k2" -> "v2"))
+    assert(desc.comment == Some("no comment"))
+  }
+
   test("create view -- basic") {
     val v1 = "CREATE VIEW view1 AS SELECT * FROM tab1"
     val (desc, exists) = extractTableDesc(v1)

From 250f402372e9826865749f3b81cd96a7cdaff657 Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Tue, 12 Apr 2016 10:58:42 -0700
Subject: [PATCH 08/18] Not OK

---
 .../org/apache/spark/sql/hive/thriftserver/CliSuite.scala | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/CliSuite.scala b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/CliSuite.scala
index bfc3d195ff2ab..eb49eabcb1ba9 100644
--- a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/CliSuite.scala
+++ b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/CliSuite.scala
@@ -162,7 +162,7 @@ class CliSuite extends SparkFunSuite with BeforeAndAfterAll with Logging {
 
     runCliWithin(3.minute)(
       "CREATE TABLE hive_test(key INT, val STRING);"
-        -> "OK",
+        -> "",
       "SHOW TABLES;"
         -> "hive_test",
       s"LOAD DATA LOCAL INPATH '$dataFilePath' OVERWRITE INTO TABLE hive_test;"
@@ -187,7 +187,7 @@ class CliSuite extends SparkFunSuite with BeforeAndAfterAll with Logging {
       "USE hive_test_db;"
         -> "",
       "CREATE TABLE hive_test(key INT, val STRING);"
-        -> "OK",
+        -> "",
       "SHOW TABLES;"
         -> "hive_test"
     )
@@ -210,9 +210,9 @@ class CliSuite extends SparkFunSuite with BeforeAndAfterAll with Logging {
       """CREATE TABLE t1(key string, val string)
         |ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe';
       """.stripMargin
-        -> "OK",
+        -> "",
       "CREATE TABLE sourceTable (key INT, val STRING);"
-        -> "OK",
+        -> "",
       s"LOAD DATA LOCAL INPATH '$dataFilePath' OVERWRITE INTO TABLE sourceTable;"
         -> "OK",
       "INSERT INTO TABLE t1 SELECT key, val FROM sourceTable;"

From efecac9b01b3ff8be296234392f4a6c922fa2d25 Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Tue, 12 Apr 2016 15:22:54 -0700
Subject: [PATCH 09/18] Fix part of InsertIntoHiveTableSuite

We weren't using the right default serde in Hive. Note that this
still fails a test with "Reference 'ds' is ambiguous ...", but
this error is common across many tests so it will be addressed
in a future commit.
---
 .../apache/spark/sql/hive/execution/HiveSqlParser.scala    | 7 ++++---
 .../apache/spark/sql/hive/InsertIntoHiveTableSuite.scala   | 2 +-
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
index 98a3ba02125f0..538f771597ee2 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
@@ -201,10 +201,11 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
       CatalogStorageFormat(
         locationUri = None,
         inputFormat = defaultHiveSerde.flatMap(_.inputFormat)
-          .orElse(Option("org.apache.hadoop.mapred.TextInputFormat")),
+          .orElse(Some("org.apache.hadoop.mapred.TextInputFormat")),
         outputFormat = defaultHiveSerde.flatMap(_.outputFormat)
-          .orElse(Option("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat")),
-        serde = defaultHiveSerde.flatMap(_.serde),
+          .orElse(Some("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat")),
+        serde = defaultHiveSerde.flatMap(_.serde)
+          .orElse(Some("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe")),
         serdeProperties = Map())
     }
     val fileStorage = Option(ctx.createFileFormat).map(visitCreateFileFormat)
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertIntoHiveTableSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertIntoHiveTableSuite.scala
index 40e9c9362cf5e..4db95636e7610 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertIntoHiveTableSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertIntoHiveTableSuite.scala
@@ -81,7 +81,7 @@ class InsertIntoHiveTableSuite extends QueryTest with TestHiveSingleton with Bef
   test("Double create fails when allowExisting = false") {
     sql("CREATE TABLE doubleCreateAndInsertTest (key int, value string)")
 
-    intercept[QueryExecutionException] {
+    intercept[AnalysisException] {
       sql("CREATE TABLE doubleCreateAndInsertTest (key int, value string)")
     }
   }

From 50a2054ec7a7276d45c1ab5adabd4550e00c7811 Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Tue, 12 Apr 2016 15:39:15 -0700
Subject: [PATCH 10/18] Fix ambiguous reference bug

In HiveMetastoreCatalog we already combined the schema and the
partition keys to compensate for the fact that Hive separates it.
Now this logic is pushed to the edges where Spark talks to Hive.
---
 .../org/apache/spark/sql/hive/HiveMetastoreCatalog.scala     | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
index be1db95ad43b5..87a794bc5d77b 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
@@ -1018,7 +1018,10 @@ private[hive] case class MetastoreRelation(
   val partitionKeys = table.partitionColumns.map(_.toAttribute)
 
   /** Non-partitionKey attributes */
-  val attributes = table.schema.map(_.toAttribute)
+  // TODO: just make this hold the schema itself, not just non-partition columns
+  val attributes = table.schema
+    .filter { c => !table.partitionColumnNames.contains(c.name) }
+    .map(_.toAttribute)
 
   val output = attributes ++ partitionKeys
 

From 8dc554a38c9989fc43b119645bfe5c8ceb7b6cdb Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Tue, 12 Apr 2016 16:42:29 -0700
Subject: [PATCH 11/18] Fix ParquetMetastoreSuite

Previously we always converted the data type string to lower case.
However, for struct fields this also converts the struct field
names to lower case. This is not what tests (or perhaps user code)
expects.
---
 .../sql/hive/execution/HiveSqlParser.scala     | 18 ++++++++++--------
 1 file changed, 10 insertions(+), 8 deletions(-)

diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
index 538f771597ee2..6926be1b23394 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
@@ -36,7 +36,7 @@ import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.execution.SparkSqlAstBuilder
 import org.apache.spark.sql.execution.command.CreateTable
 import org.apache.spark.sql.hive.{CreateTableAsSelect => CTAS, CreateViewAsSelect => CreateView}
-import org.apache.spark.sql.hive.{HiveGenericUDTF, HiveSerDe}
+import org.apache.spark.sql.hive.{HiveGenericUDTF, HiveMetastoreTypes, HiveSerDe}
 import org.apache.spark.sql.hive.HiveShim.HiveFunctionWrapper
 
 /**
@@ -185,8 +185,8 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
       CatalogTableType.MANAGED_TABLE
     }
     val comment = Option(ctx.STRING).map(string)
-    val partitionCols = Option(ctx.partitionColumns).toSeq.flatMap(visitCatalogColumns(_))
-    val cols = Option(ctx.columns).toSeq.flatMap(visitCatalogColumns(_, _.toLowerCase))
+    val partitionCols = Option(ctx.partitionColumns).toSeq.flatMap(visitCatalogColumns)
+    val cols = Option(ctx.columns).toSeq.flatMap(visitCatalogColumns)
     val properties = Option(ctx.tablePropertyList).map(visitTablePropertyList).getOrElse(Map.empty)
     val selectQuery = Option(ctx.query).map(plan)
 
@@ -479,13 +479,15 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
   /**
    * Create a sequence of [[CatalogColumn]]s from a column list
    */
-  private def visitCatalogColumns(
-      ctx: ColTypeListContext,
-      formatter: String => String = identity): Seq[CatalogColumn] = withOrigin(ctx) {
+  private def visitCatalogColumns(ctx: ColTypeListContext): Seq[CatalogColumn] = withOrigin(ctx) {
     ctx.colType.asScala.map { col =>
       CatalogColumn(
-        formatter(col.identifier.getText),
-        col.dataType.getText.toLowerCase, // TODO validate this?
+        col.identifier.getText,
+        // Note: for types like "STRUCT<myFirstName: STRING, myLastName: STRING>" we can't
+        // just convert the whole type string to lower case, otherwise the struct field names
+        // will no longer be case sensitive. Instead, we rely on our parser to get the proper
+        // case before passing it to Hive.
+        HiveMetastoreTypes.toDataType(col.dataType.getText).simpleString,
         nullable = true,
         Option(col.STRING).map(string))
     }

From a4f67f2a53ecb63decd348ce57b22519e3cd78c0 Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Tue, 12 Apr 2016 16:53:01 -0700
Subject: [PATCH 12/18] Fix SQLQuerySuite

---
 .../org/apache/spark/sql/hive/execution/SQLQuerySuite.scala     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
index 7eaf19dfe9fd2..5ce16be4dc059 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
@@ -360,7 +360,7 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
       var message = intercept[AnalysisException] {
         sql("CREATE TABLE ctas1 AS SELECT key k, value FROM src ORDER BY k, value")
       }.getMessage
-      assert(message.contains("ctas1 already exists"))
+      assert(message.contains("already exists"))
       checkRelation("ctas1", true)
       sql("DROP TABLE ctas1")
 

From 045820cf8a5aaf74304aea763d804ddfe98d2806 Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Tue, 12 Apr 2016 18:01:48 -0700
Subject: [PATCH 13/18] Fix HiveCompatibilitySuite (ignored some tests)

---
 .../execution/HiveCompatibilitySuite.scala    | 143 +++++++++---------
 1 file changed, 74 insertions(+), 69 deletions(-)

diff --git a/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala b/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala
index f0eeda09dba5a..42dbb80b7c65c 100644
--- a/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala
+++ b/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala
@@ -366,10 +366,77 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "sort_merge_join_desc_6",
     "sort_merge_join_desc_7",
 
+    // These tests try to create a table with bucketed columns, which we don't support
+    "auto_join32",
+    "auto_join_filters",
+    "auto_smb_mapjoin_14",
+    "ct_case_insensitive",
+    "explain_rearrange",
+    "groupby_sort_10",
+    "groupby_sort_2",
+    "groupby_sort_3",
+    "groupby_sort_4",
+    "groupby_sort_5",
+    "groupby_sort_7",
+    "groupby_sort_8",
+    "groupby_sort_9",
+    "groupby_sort_test_1",
+    "inputddl4",
+    "join_filters",
+    "join_nulls",
+    "join_nullsafe",
+    "load_dyn_part2",
+    "orc_empty_files",
+    "reduce_deduplicate",
+    "smb_mapjoin9",
+    "smb_mapjoin_1",
+    "smb_mapjoin_10",
+    "smb_mapjoin_13",
+    "smb_mapjoin_14",
+    "smb_mapjoin_15",
+    "smb_mapjoin_16",
+    "smb_mapjoin_17",
+    "smb_mapjoin_2",
+    "smb_mapjoin_21",
+    "smb_mapjoin_25",
+    "smb_mapjoin_3",
+    "smb_mapjoin_4",
+    "smb_mapjoin_5",
+    "smb_mapjoin_6",
+    "smb_mapjoin_7",
+    "smb_mapjoin_8",
+    "sort_merge_join_desc_1",
+    "sort_merge_join_desc_2",
+    "sort_merge_join_desc_3",
+    "sort_merge_join_desc_4",
+
+    // These tests try to create a table with skewed columns, which we don't support
+    "create_skewed_table1",
+    "skewjoinopt13",
+    "skewjoinopt18",
+    "skewjoinopt9",
+
+    // IWASHERE
     // Index commands are not supported
     "drop_index",
     "drop_index_removes_partition_dirs",
     "alter_index",
+    "auto_sortmerge_join_1",
+    "auto_sortmerge_join_10",
+    "auto_sortmerge_join_11",
+    "auto_sortmerge_join_12",
+    "auto_sortmerge_join_13",
+    "auto_sortmerge_join_14",
+    "auto_sortmerge_join_15",
+    "auto_sortmerge_join_16",
+    "auto_sortmerge_join_2",
+    "auto_sortmerge_join_3",
+    "auto_sortmerge_join_4",
+    "auto_sortmerge_join_5",
+    "auto_sortmerge_join_6",
+    "auto_sortmerge_join_7",
+    "auto_sortmerge_join_8",
+    "auto_sortmerge_join_9",
 
     // Macro commands are not supported
     "macro",
@@ -435,33 +502,14 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "auto_join3",
     "auto_join30",
     "auto_join31",
-    "auto_join32",
     "auto_join4",
     "auto_join5",
     "auto_join6",
     "auto_join7",
     "auto_join8",
     "auto_join9",
-    "auto_join_filters",
     "auto_join_nulls",
     "auto_join_reordering_values",
-    "auto_smb_mapjoin_14",
-    "auto_sortmerge_join_1",
-    "auto_sortmerge_join_10",
-    "auto_sortmerge_join_11",
-    "auto_sortmerge_join_12",
-    "auto_sortmerge_join_13",
-    "auto_sortmerge_join_14",
-    "auto_sortmerge_join_15",
-    "auto_sortmerge_join_16",
-    "auto_sortmerge_join_2",
-    "auto_sortmerge_join_3",
-    "auto_sortmerge_join_4",
-    "auto_sortmerge_join_5",
-    "auto_sortmerge_join_6",
-    "auto_sortmerge_join_7",
-    "auto_sortmerge_join_8",
-    "auto_sortmerge_join_9",
     "binary_constant",
     "binarysortable_1",
     "cast1",
@@ -490,15 +538,13 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "count",
     "cp_mj_rc",
     "create_insert_outputformat",
-    "create_like_tbl_props",
+    //"create_like_tbl_props", // TODO: results don't match
     "create_nested_type",
-    "create_skewed_table1",
     "create_struct_table",
     "create_view_translate",
     "cross_join",
     "cross_product_check_1",
     "cross_product_check_2",
-    "ct_case_insensitive",
     "database_drop",
     "database_location",
     "database_properties",
@@ -509,7 +555,7 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "date_comparison",
     "date_join1",
     "date_serde",
-    "decimal_1",
+    //"decimal_1", // TODO: cannot parse column decimal(5)
     "decimal_4",
     "decimal_join",
     "default_partition_name",
@@ -534,7 +580,6 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "escape_distributeby1",
     "escape_orderby1",
     "escape_sortby1",
-    "explain_rearrange",
     "fileformat_mix",
     "fileformat_sequencefile",
     "fileformat_text",
@@ -589,16 +634,7 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "groupby_neg_float",
     "groupby_ppd",
     "groupby_ppr",
-    "groupby_sort_10",
-    "groupby_sort_2",
-    "groupby_sort_3",
-    "groupby_sort_4",
-    "groupby_sort_5",
     "groupby_sort_6",
-    "groupby_sort_7",
-    "groupby_sort_8",
-    "groupby_sort_9",
-    "groupby_sort_test_1",
     "having",
     "implicit_cast1",
     "index_serde",
@@ -653,7 +689,6 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "inputddl1",
     "inputddl2",
     "inputddl3",
-    "inputddl4",
     "inputddl6",
     "inputddl7",
     "inputddl8",
@@ -709,11 +744,8 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "join_array",
     "join_casesensitive",
     "join_empty",
-    "join_filters",
     "join_hive_626",
     "join_map_ppr",
-    "join_nulls",
-    "join_nullsafe",
     "join_rc",
     "join_reorder2",
     "join_reorder3",
@@ -737,7 +769,6 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "load_dyn_part13",
     "load_dyn_part14",
     "load_dyn_part14_win",
-    "load_dyn_part2",
     "load_dyn_part3",
     "load_dyn_part4",
     "load_dyn_part5",
@@ -790,14 +821,13 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "nullscript",
     "optional_outer",
     "orc_dictionary_threshold",
-    "orc_empty_files",
     "order",
     "order2",
     "outer_join_ppr",
     "parallel",
     "parenthesis_star_by",
     "part_inherit_tbl_props",
-    "part_inherit_tbl_props_empty",
+    //"part_inherit_tbl_props_empty", // TODO: results don't match
     "part_inherit_tbl_props_with_star",
     "partcols1",
     "partition_date",
@@ -846,7 +876,6 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "rcfile_null_value",
     "rcfile_toleratecorruptions",
     "rcfile_union",
-    "reduce_deduplicate",
     "reduce_deduplicate_exclude_gby",
     "reduce_deduplicate_exclude_join",
     "reduce_deduplicate_extended",
@@ -867,34 +896,10 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "show_functions",
     "show_partitions",
     "show_tblproperties",
-    "skewjoinopt13",
-    "skewjoinopt18",
-    "skewjoinopt9",
-    "smb_mapjoin9",
-    "smb_mapjoin_1",
-    "smb_mapjoin_10",
-    "smb_mapjoin_13",
-    "smb_mapjoin_14",
-    "smb_mapjoin_15",
-    "smb_mapjoin_16",
-    "smb_mapjoin_17",
-    "smb_mapjoin_2",
-    "smb_mapjoin_21",
-    "smb_mapjoin_25",
-    "smb_mapjoin_3",
-    "smb_mapjoin_4",
-    "smb_mapjoin_5",
-    "smb_mapjoin_6",
-    "smb_mapjoin_7",
-    "smb_mapjoin_8",
     "sort",
-    "sort_merge_join_desc_1",
-    "sort_merge_join_desc_2",
-    "sort_merge_join_desc_3",
-    "sort_merge_join_desc_4",
     "stats0",
     "stats_aggregator_error_1",
-    "stats_empty_partition",
+    //"stats_empty_partition", // TODO: results don't match
     "stats_publisher_error_1",
     "subq2",
     "tablename_with_select",
@@ -1051,7 +1056,7 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "udf_xpath_long",
     "udf_xpath_short",
     "udf_xpath_string",
-    "unicode_notation",
+    //"unicode_notation", // TODO: results don't match
     "union10",
     "union11",
     "union13",
@@ -1085,8 +1090,8 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "union_date",
     "union_lateralview",
     "union_ppr",
-    "union_remove_11",
-    "union_remove_3",
+    //"union_remove_11", // TODO: results don't match
+    //"union_remove_3", // TODO: results don't match
     "union_remove_6",
     "union_script",
     "varchar_2",

From 8e273fdc4f95d08cb6d09f4641472861587a3a01 Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Tue, 12 Apr 2016 18:05:02 -0700
Subject: [PATCH 14/18] Fix HiveDDLCommandSuite

---
 .../org/apache/spark/sql/hive/execution/HiveSqlParser.scala | 2 +-
 .../org/apache/spark/sql/hive/HiveDDLCommandSuite.scala     | 6 +++---
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
index 6926be1b23394..0f53b1abd4bfd 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
@@ -482,7 +482,7 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
   private def visitCatalogColumns(ctx: ColTypeListContext): Seq[CatalogColumn] = withOrigin(ctx) {
     ctx.colType.asScala.map { col =>
       CatalogColumn(
-        col.identifier.getText,
+        col.identifier.getText.toLowerCase,
         // Note: for types like "STRUCT<myFirstName: STRING, myLastName: STRING>" we can't
         // just convert the whole type string to lower case, otherwise the struct field names
         // will no longer be case sensitive. Instead, we rely on our parser to get the proper
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
index efd59d332316e..b16076da73ccb 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
@@ -160,7 +160,7 @@ class HiveDDLCommandSuite extends PlanTest {
     assert(desc.storage.inputFormat == Some("org.apache.hadoop.mapred.TextInputFormat"))
     assert(desc.storage.outputFormat ==
       Some("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat"))
-    assert(desc.storage.serde.isEmpty)
+    assert(desc.storage.serde == Some("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe"))
     assert(desc.properties == Map())
   }
 
@@ -340,7 +340,7 @@ class HiveDDLCommandSuite extends PlanTest {
       Some("org.apache.hadoop.mapred.TextInputFormat"))
     assert(desc.storage.outputFormat ==
       Some("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat"))
-    assert(desc.storage.serde.isEmpty)
+    assert(desc.storage.serde == Some("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe"))
     assert(desc.storage.serdeProperties.isEmpty)
     assert(desc.properties.isEmpty)
     assert(desc.comment.isEmpty)
@@ -446,7 +446,7 @@ class HiveDDLCommandSuite extends PlanTest {
     val (desc2, _) = extractTableDesc(query2)
     assert(desc1.storage.inputFormat == Some("winput"))
     assert(desc1.storage.outputFormat == Some("wowput"))
-    assert(desc1.storage.serde.isEmpty)
+    assert(desc1.storage.serde == Some("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe"))
     assert(desc2.storage.inputFormat == Some("org.apache.hadoop.hive.ql.io.orc.OrcInputFormat"))
     assert(desc2.storage.outputFormat == Some("org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat"))
     assert(desc2.storage.serde == Some("org.apache.hadoop.hive.ql.io.orc.OrcSerde"))

From 59edce332f87b07bdfb07e2e385431b2b123e1b0 Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Tue, 12 Apr 2016 23:12:10 -0700
Subject: [PATCH 15/18] Fix SQLQuerySuite CTAS

---
 .../org/apache/spark/sql/hive/client/HiveClientImpl.scala   | 3 ++-
 .../org/apache/spark/sql/hive/execution/HiveSqlParser.scala | 5 +++--
 .../org/apache/spark/sql/hive/HiveDDLCommandSuite.scala     | 6 +++---
 3 files changed, 8 insertions(+), 6 deletions(-)

diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index 760cd842cc722..975444192c9ea 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -710,7 +710,8 @@ private[hive] class HiveClientImpl(
     table.storage.locationUri.foreach { loc => shim.setDataLocation(hiveTable, loc) }
     table.storage.inputFormat.map(toInputFormat).foreach(hiveTable.setInputFormatClass)
     table.storage.outputFormat.map(toOutputFormat).foreach(hiveTable.setOutputFormatClass)
-    table.storage.serde.foreach(hiveTable.setSerializationLib)
+    hiveTable.setSerializationLib(
+      table.storage.serde.getOrElse("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe"))
     table.storage.serdeProperties.foreach { case (k, v) => hiveTable.setSerdeParam(k, v) }
     table.properties.foreach { case (k, v) => hiveTable.setProperty(k, v) }
     table.comment.foreach { c => hiveTable.setProperty("comment", c) }
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
index 0f53b1abd4bfd..82eb3f21d5330 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
@@ -204,8 +204,9 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
           .orElse(Some("org.apache.hadoop.mapred.TextInputFormat")),
         outputFormat = defaultHiveSerde.flatMap(_.outputFormat)
           .orElse(Some("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat")),
-        serde = defaultHiveSerde.flatMap(_.serde)
-          .orElse(Some("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe")),
+        // Note: Keep this unspecified because we use the presence of the serde to decide
+        // whether to convert a table created by CTAS to a datasource table.
+        serde = None,
         serdeProperties = Map())
     }
     val fileStorage = Option(ctx.createFileFormat).map(visitCreateFileFormat)
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
index b16076da73ccb..efd59d332316e 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
@@ -160,7 +160,7 @@ class HiveDDLCommandSuite extends PlanTest {
     assert(desc.storage.inputFormat == Some("org.apache.hadoop.mapred.TextInputFormat"))
     assert(desc.storage.outputFormat ==
       Some("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat"))
-    assert(desc.storage.serde == Some("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe"))
+    assert(desc.storage.serde.isEmpty)
     assert(desc.properties == Map())
   }
 
@@ -340,7 +340,7 @@ class HiveDDLCommandSuite extends PlanTest {
       Some("org.apache.hadoop.mapred.TextInputFormat"))
     assert(desc.storage.outputFormat ==
       Some("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat"))
-    assert(desc.storage.serde == Some("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe"))
+    assert(desc.storage.serde.isEmpty)
     assert(desc.storage.serdeProperties.isEmpty)
     assert(desc.properties.isEmpty)
     assert(desc.comment.isEmpty)
@@ -446,7 +446,7 @@ class HiveDDLCommandSuite extends PlanTest {
     val (desc2, _) = extractTableDesc(query2)
     assert(desc1.storage.inputFormat == Some("winput"))
     assert(desc1.storage.outputFormat == Some("wowput"))
-    assert(desc1.storage.serde == Some("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe"))
+    assert(desc1.storage.serde.isEmpty)
     assert(desc2.storage.inputFormat == Some("org.apache.hadoop.hive.ql.io.orc.OrcInputFormat"))
     assert(desc2.storage.outputFormat == Some("org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat"))
     assert(desc2.storage.serde == Some("org.apache.hadoop.hive.ql.io.orc.OrcSerde"))

From 7b1a1e381c97cbcb59fa2c36e15523273e6f7c28 Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Tue, 12 Apr 2016 23:26:54 -0700
Subject: [PATCH 16/18] Fix all but 1 ignored test in HiveCompatibilitySuite

There were a few differences in DESCRIBE TABLE:
- output format should be HiveIgnoreKeyTextOutputFormat
- num buckets should be -1
- last access time should be -1
- EXTERNAL should not be set to false for managed table

After making these changes out result now matches Hive's.
---
 .../spark/sql/catalyst/catalog/interface.scala      |  4 ++--
 .../sql/hive/execution/HiveCompatibilitySuite.scala | 13 ++++++-------
 .../spark/sql/hive/HiveMetastoreCatalog.scala       |  2 +-
 .../spark/sql/hive/client/HiveClientImpl.scala      |  9 +++------
 .../spark/sql/hive/execution/HiveSqlParser.scala    |  2 +-
 .../apache/spark/sql/hive/HiveDDLCommandSuite.scala |  8 ++++----
 6 files changed, 17 insertions(+), 21 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
index 4601502e50e84..ad989a97e4afa 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
@@ -223,9 +223,9 @@ case class CatalogTable(
     partitionColumnNames: Seq[String] = Seq.empty,
     sortColumnNames: Seq[String] = Seq.empty,
     bucketColumnNames: Seq[String] = Seq.empty,
-    numBuckets: Int = 0,
+    numBuckets: Int = -1,
     createTime: Long = System.currentTimeMillis,
-    lastAccessTime: Long = System.currentTimeMillis,
+    lastAccessTime: Long = -1,
     properties: Map[String, String] = Map.empty,
     viewOriginalText: Option[String] = None,
     viewText: Option[String] = None,
diff --git a/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala b/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala
index 42dbb80b7c65c..448fd8a4b0845 100644
--- a/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala
+++ b/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala
@@ -416,7 +416,6 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "skewjoinopt18",
     "skewjoinopt9",
 
-    // IWASHERE
     // Index commands are not supported
     "drop_index",
     "drop_index_removes_partition_dirs",
@@ -538,7 +537,7 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "count",
     "cp_mj_rc",
     "create_insert_outputformat",
-    //"create_like_tbl_props", // TODO: results don't match
+    "create_like_tbl_props",
     "create_nested_type",
     "create_struct_table",
     "create_view_translate",
@@ -827,7 +826,7 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "parallel",
     "parenthesis_star_by",
     "part_inherit_tbl_props",
-    //"part_inherit_tbl_props_empty", // TODO: results don't match
+    "part_inherit_tbl_props_empty",
     "part_inherit_tbl_props_with_star",
     "partcols1",
     "partition_date",
@@ -899,7 +898,7 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "sort",
     "stats0",
     "stats_aggregator_error_1",
-    //"stats_empty_partition", // TODO: results don't match
+    "stats_empty_partition",
     "stats_publisher_error_1",
     "subq2",
     "tablename_with_select",
@@ -1056,7 +1055,7 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "udf_xpath_long",
     "udf_xpath_short",
     "udf_xpath_string",
-    //"unicode_notation", // TODO: results don't match
+    "unicode_notation",
     "union10",
     "union11",
     "union13",
@@ -1090,8 +1089,8 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "union_date",
     "union_lateralview",
     "union_ppr",
-    //"union_remove_11", // TODO: results don't match
-    //"union_remove_3", // TODO: results don't match
+    "union_remove_11",
+    "union_remove_3",
     "union_remove_6",
     "union_script",
     "varchar_2",
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
index 87a794bc5d77b..ccc8345d7375d 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
@@ -91,7 +91,7 @@ private[hive] object HiveSerDe {
       "textfile" ->
         HiveSerDe(
           inputFormat = Option("org.apache.hadoop.mapred.TextInputFormat"),
-          outputFormat = Option("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat")),
+          outputFormat = Option("org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat")),
 
       "avro" ->
         HiveSerDe(
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index 975444192c9ea..d71071d43c77e 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -680,17 +680,14 @@ private[hive] class HiveClientImpl(
 
   private def toHiveTable(table: CatalogTable): HiveTable = {
     val hiveTable = new HiveTable(table.database, table.identifier.table)
-    // For EXTERNAL_TABLE/MANAGED_TABLE, we also need to set EXTERNAL field in
-    // the table properties accodringly. Otherwise, if EXTERNAL_TABLE is the table type
-    // but EXTERNAL field is not set, Hive metastore will change the type to
-    // MANAGED_TABLE (see
-    // metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java#L1095-L1105)
+    // For EXTERNAL_TABLE, we also need to set EXTERNAL field in the table properties.
+    // Otherwise, Hive metastore will change the table to a MANAGED_TABLE.
+    // (metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java#L1095-L1105)
     hiveTable.setTableType(table.tableType match {
       case CatalogTableType.EXTERNAL_TABLE =>
         hiveTable.setProperty("EXTERNAL", "TRUE")
         HiveTableType.EXTERNAL_TABLE
       case CatalogTableType.MANAGED_TABLE =>
-        hiveTable.setProperty("EXTERNAL", "FALSE")
         HiveTableType.MANAGED_TABLE
       case CatalogTableType.INDEX_TABLE => HiveTableType.INDEX_TABLE
       case CatalogTableType.VIRTUAL_VIEW => HiveTableType.VIRTUAL_VIEW
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
index 82eb3f21d5330..a2df68897036e 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
@@ -203,7 +203,7 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
         inputFormat = defaultHiveSerde.flatMap(_.inputFormat)
           .orElse(Some("org.apache.hadoop.mapred.TextInputFormat")),
         outputFormat = defaultHiveSerde.flatMap(_.outputFormat)
-          .orElse(Some("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat")),
+          .orElse(Some("org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat")),
         // Note: Keep this unspecified because we use the presence of the serde to decide
         // whether to convert a table created by CTAS to a datasource table.
         serde = None,
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
index efd59d332316e..68d3ea6ed93db 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveDDLCommandSuite.scala
@@ -159,7 +159,7 @@ class HiveDDLCommandSuite extends PlanTest {
     assert(desc.storage.serdeProperties == Map())
     assert(desc.storage.inputFormat == Some("org.apache.hadoop.mapred.TextInputFormat"))
     assert(desc.storage.outputFormat ==
-      Some("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat"))
+      Some("org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat"))
     assert(desc.storage.serde.isEmpty)
     assert(desc.properties == Map())
   }
@@ -332,14 +332,14 @@ class HiveDDLCommandSuite extends PlanTest {
     assert(desc.partitionColumnNames.isEmpty)
     assert(desc.sortColumnNames.isEmpty)
     assert(desc.bucketColumnNames.isEmpty)
-    assert(desc.numBuckets == 0)
+    assert(desc.numBuckets == -1)
     assert(desc.viewText.isEmpty)
     assert(desc.viewOriginalText.isEmpty)
     assert(desc.storage.locationUri.isEmpty)
     assert(desc.storage.inputFormat ==
       Some("org.apache.hadoop.mapred.TextInputFormat"))
     assert(desc.storage.outputFormat ==
-      Some("org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat"))
+      Some("org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat"))
     assert(desc.storage.serde.isEmpty)
     assert(desc.storage.serdeProperties.isEmpty)
     assert(desc.properties.isEmpty)
@@ -497,7 +497,7 @@ class HiveDDLCommandSuite extends PlanTest {
     assert(desc.partitionColumnNames == Seq("month"))
     assert(desc.sortColumnNames.isEmpty)
     assert(desc.bucketColumnNames.isEmpty)
-    assert(desc.numBuckets == 0)
+    assert(desc.numBuckets == -1)
     assert(desc.viewText.isEmpty)
     assert(desc.viewOriginalText.isEmpty)
     assert(desc.storage.locationUri == Some("/path/to/mercury"))

From a60e66a71dec96973043ec62e2b6d4213c5add2c Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Tue, 12 Apr 2016 23:31:31 -0700
Subject: [PATCH 17/18] Fix last ignored test in HiveCompatibilitySuite

CatalystSqlParser knows how to parse decimal(5)!
---
 .../spark/sql/hive/execution/HiveCompatibilitySuite.scala       | 2 +-
 .../org/apache/spark/sql/hive/execution/HiveSqlParser.scala     | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala b/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala
index 448fd8a4b0845..a45d180464602 100644
--- a/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala
+++ b/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala
@@ -554,7 +554,7 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "date_comparison",
     "date_join1",
     "date_serde",
-    //"decimal_1", // TODO: cannot parse column decimal(5)
+    "decimal_1",
     "decimal_4",
     "decimal_join",
     "default_partition_name",
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
index a2df68897036e..b14db7fe71619 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveSqlParser.scala
@@ -488,7 +488,7 @@ class HiveSqlAstBuilder extends SparkSqlAstBuilder {
         // just convert the whole type string to lower case, otherwise the struct field names
         // will no longer be case sensitive. Instead, we rely on our parser to get the proper
         // case before passing it to Hive.
-        HiveMetastoreTypes.toDataType(col.dataType.getText).simpleString,
+        CatalystSqlParser.parseDataType(col.dataType.getText).simpleString,
         nullable = true,
         Option(col.STRING).map(string))
     }

From 55957bdcd1dea06da750c5cf7cfa587da5250574 Mon Sep 17 00:00:00 2001
From: Yin Huai <yhuai@databricks.com>
Date: Wed, 13 Apr 2016 09:08:36 -0700
Subject: [PATCH 18/18] Preserve an existing behavior.

---
 .../spark/sql/hive/client/HiveClientImpl.scala      | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index d71071d43c77e..2a1fff92b570a 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -696,7 +696,18 @@ private[hive] class HiveClientImpl(
     val (partCols, schema) = table.schema.map(toHiveColumn).partition { c =>
       table.partitionColumnNames.contains(c.getName)
     }
-    hiveTable.setFields(schema.asJava)
+    if (table.schema.isEmpty) {
+      // This is a hack to preserve existing behavior. Before Spark 2.0, we do not
+      // set a default serde here (this was done in Hive), and so if the user provides
+      // an empty schema Hive would automatically populate the schema with a single
+      // field "col". However, after SPARK-14388, we set the default serde to
+      // LazySimpleSerde so this implicit behavior no longer happens. Therefore,
+      // we need to do it in Spark ourselves.
+      hiveTable.setFields(
+        Seq(new FieldSchema("col", "array<string>", "from deserializer")).asJava)
+    } else {
+      hiveTable.setFields(schema.asJava)
+    }
     hiveTable.setPartCols(partCols.asJava)
     // TODO: set sort columns here too
     hiveTable.setBucketCols(table.bucketColumnNames.asJava)