Add specific node CreateHiveTableAsSelect and remove type parameter of CreateTableAsSelect.

Author: viirya

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala

@@ -137,12 +137,13 @@ case class InsertIntoTable(
   }
 }
 
-case class CreateTableAsSelect[T](
-    databaseName: Option[String],
-    tableName: String,
-    child: LogicalPlan,
-    allowExisting: Boolean,
-    desc: T) extends UnaryNode {
+trait CreateTableAsSelect extends UnaryNode {
+  self: Product =>
+  def databaseName: Option[String]
+  def tableName: String
+  def child: LogicalPlan
+  def allowExisting: Boolean
+
   override def output = Seq.empty[Attribute]
   override lazy val resolved = databaseName != None && childrenResolved
 }

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

@@ -141,7 +141,7 @@ class DataFrame private[sql](
     // happen right away to let these side effects take place eagerly.
     case _: Command |
          _: InsertIntoTable |
-         _: CreateTableAsSelect[_] |
+         _: CreateTableAsSelect |
          _: CreateTableUsingAsSelect |
          _: WriteToFile =>
       LogicalRDD(queryExecution.analyzed.output, queryExecution.toRdd)(sqlContext)

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

@@ -24,6 +24,7 @@ import com.google.common.cache.{CacheBuilder, CacheLoader, LoadingCache}
 import org.apache.hadoop.hive.metastore.api.{FieldSchema, Partition => TPartition, Table => TTable}
 import org.apache.hadoop.hive.metastore.{TableType, Warehouse}
 import org.apache.hadoop.hive.ql.metadata._
+import org.apache.hadoop.hive.ql.lib.Node
 import org.apache.hadoop.hive.ql.plan.CreateTableDesc
 import org.apache.hadoop.hive.serde.serdeConstants
 import org.apache.hadoop.hive.serde2.`lazy`.LazySimpleSerDe
@@ -527,7 +528,7 @@ private[hive] class HiveMetastoreCatalog(hive: HiveContext) extends Catalog with
 
       // TODO extra is in type of ASTNode which means the logical plan is not resolved
       // Need to think about how to implement the CreateTableAsSelect.resolved
-      case CreateTableAsSelect(db, tableName, child, allowExisting, extra: ASTNode) =>
+      case CreateHiveTableAsSelect(db, tableName, child, allowExisting, extra) =>
         val (dbName, tblName) = processDatabaseAndTableName(db, tableName)
         val databaseName = dbName.getOrElse(hive.sessionState.getCurrentDatabase)
 
@@ -546,7 +547,7 @@ private[hive] class HiveMetastoreCatalog(hive: HiveContext) extends Catalog with
             }
           }
 
-          sa.analyze(extra, new Context(hive.hiveconf))
+          sa.analyze(extra.asInstanceOf[ASTNode], new Context(hive.hiveconf))
           Some(sa.getQB().getTableDesc)
         }
 
@@ -663,6 +664,14 @@ private[hive] case class InsertIntoHiveTable(
   }
 }
 
+private[hive] case class CreateHiveTableAsSelect(
+    databaseName: Option[String],
+    tableName: String,
+    child: LogicalPlan,
+    allowExisting: Boolean,
+    desc: Node) extends CreateTableAsSelect {
+}
+
 private[hive] case class MetastoreRelation
     (databaseName: String, tableName: String, alias: Option[String])
     (val table: TTable, val partitions: Seq[TPartition])

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

@@ -557,7 +557,7 @@ https://cwiki.apache.org/confluence/display/Hive/Enhanced+Aggregation%2C+Cube%2C
             "TOK_TABLEPROPERTIES"),
           children)
       val (db, tableName) = extractDbNameTableName(tableNameParts)
-      CreateTableAsSelect(db, tableName, nodeToPlan(query), allowExisting != None, node)
+      CreateHiveTableAsSelect(db, tableName, nodeToPlan(query), allowExisting != None, node)
 
     // If its not a "CREATE TABLE AS" like above then just pass it back to hive as a native command.
     case Token("TOK_CREATETABLE", _) => NativePlaceholder