[SPARK-16471] [SQL] Remove Hive-specific CreateHiveTableAsSelectLogicalPlan [WIP]

[gatorsmile]

What changes were proposed in this pull request?

CreateHiveTableAsSelectLogicalPlan is a Hive-specific logical node. This is not a good design. We need to consolidate it with CreateTableUsingAsSelect to build a unified logical node CreateTableAsSelect.

The first step is to make more general the signature of CreateTableUsingAsSelect by using CatalogTable as the input of Table metadata. The logical node is renamed to CreateTableAsSelect. The new interface will be like

caseclassCreateTableAsSelect(
tableDesc: CatalogTable,
provider: String,
mode: SaveMode,
child: LogicalPlan) extends logical.UnaryNode

The second step is to convert CreateHiveTableAsSelectLogicalPlan into CreateTableAsSelect.

This PR is based on the compare of the two interfaces. The details are described below.

Currently, the SQL interface is the only only entrance to CreateHiveTableAsSelectLogicalPlan. Below describes the correspondence between the SQL interface and CreateHiveTableAsSelectLogicalPlan

caseclassCreateHiveTableAsSelectLogicalPlan(
tableDesc: CatalogTable,
child: LogicalPlan,
allowExisting: Boolean)
extendsUnaryNodewithCommand

SQL:When conf.convertCTAS ==false|| either [ROWFORMAT row_format] or [STOREDAS file_format] is specified
CREATE [EXTERNAL] [TEMPORARY] TABLE [IFNOTEXISTS] [db_name.]table_name
[(col1[:] data_type [COMMENT col_comment], ...)]
[COMMENT table_comment]
[PARTITIONEDBY (col2[:] data_type [COMMENT col_comment], ...)]
[ROWFORMAT row_format]
[STOREDAS file_format]
[LOCATION path]
[TBLPROPERTIES (property_name=property_value, ...)]
[AS select_statement];
-->
[TEMPORARY] is not allowed.
allowExisting: Boolean= [IFNOTEXISTS]
child: LogicalPlan= select_statement
tableDesc: CatalogTable=CatalogTable(
identifier = [db_name.]table_name,
tableType = [EXTERNAL],
storage = [ROWFORMAT row_format +
[STOREDAS file_format] +
[LOCATION path],
schema =Seq.empty,
partitionColumnNames =Seq.empty,
properties = [TBLPROPERTIES (property_name=property_value, ...)],
comment = [COMMENT table_comment])

CreateTableUsingAsSelect has three entrances. Below is the the correspondence:

caseclassCreateTableUsingAsSelect(
tableIdent: TableIdentifier,
provider: String,
partitionColumns: Array[String],
bucketSpec: Option[BucketSpec],
mode: SaveMode,
options: Map[String, String],
child: LogicalPlan) extends logical.UnaryNode

SQLInterfaceI:When conf.convertCTAS ==true&& [ROWFORMAT row_format] and [STOREDAS file_format] are not specified
CREATE [EXTERNAL] [TEMPORARY] TABLE [IFNOTEXISTS] [db_name.]table_name
[(col1[:] data_type [COMMENT col_comment], ...)]
[COMMENT table_comment]
[PARTITIONEDBY (col2[:] data_type [COMMENT col_comment], ...)]
[ROWFORMAT row_format]
[STOREDAS file_format]
[LOCATION path]
[TBLPROPERTIES (property_name=property_value, ...)]
[AS select_statement];
-->tableIdent: TableIdentifier= [db_name.]table_name,
provider: String= conf.defaultDataSourceName,
partitionColumns: Array[String] =Seq.empty,
bucketSpec: Option[BucketSpec] =None,
mode: SaveMode= [IFNOTEXISTS],
options: Map[String, String] = [LOCATION path],
child: LogicalPlan= [AS select_statement]

SQLInterfaceII:CREATE [EXTERNAL] [TEMPORARY] TABLE [IFNOTEXISTS] [db_name.]table_name
[(col1[:] data_type [COMMENT col_comment], ...)]
USING qualifiedName
[OPTIONS tablePropertyList)]
[PARTITIONEDBY (col2[:] data_type [COMMENT col_comment], ...)]
[CLUSTEREDBY (col3, ...) (SORTEDBY orderedIdentifierList)?INTOINTEGER_VALUEBUCKETS]
[AS select_statement];
-->
[EXTERNAL] is not allowed.
[TEMPORARY] is not allowed.
tableIdent: TableIdentifier= [db_name.]table_name,
provider: String=USING qualifiedName,
partitionColumns: Array[String] = [PARTITIONEDBY (col2[:] data_type [COMMENT col_comment], ...)],
bucketSpec: Option[BucketSpec] = [CLUSTEREDBY (col3, ...) (SORTEDBY orderedIdentifierList)?INTOINTEGER_VALUEBUCKETS],
mode: SaveMode= [IFNOTEXISTS],
options: Map[String, String] = [OPTIONS tablePropertyList)],
child: LogicalPlan= [AS select_statement]

DataFrameWriterInterface:tableIdent: TableIdentifier= tableIdent (from saveAsTable API),
provider: String= source (from format API),
partitionColumns: Array[String] = partitioningColumns (from partitionBy API),
bucketSpec: Option[BucketSpec] = getBucketSpec function (from bucketBy API and sortBy API),
mode: SaveMode= mode (from mode API),
options: Map[String, String] = extraOptions (from option and options API),
child: LogicalPlan= df.logicalPlan (from DataFrameWriter)

How was this patch tested?

The existing test cases cover the code refactoring

[SparkQA]

Test build #62047 has finished for PR 14123 at commit 5bef1e8.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

should we wait for #14071?

[SparkQA]

Test build #62068 has finished for PR 14123 at commit 082040f.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[gatorsmile]

@cloud-fan Yeah! Will be in [WIP] until #14071 is merged.

[gatorsmile]

This is part of #14482. Close it now