-
Notifications
You must be signed in to change notification settings - Fork 29.3k
[SPARK-14346][SQL] Native SHOW CREATE TABLE for Hive tables/views #13079
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 3 commits
b3cc55f
d116866
2020d77
f7cbc3f
74fd5d3
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -79,6 +79,12 @@ case class CatalogTablePartition( | |
| * | ||
| * Note that Hive's metastore also tracks skewed columns. We should consider adding that in the | ||
| * future once we have a better understanding of how we want to handle skewed columns. | ||
| * | ||
| * Field `fullyMapped` is used to indicate whether all table metadata entries retrieved from the | ||
| * concrete underlying external catalog (e.g. Hive metastore) are mapped to fields of this | ||
| * [[CatalogTable]]. For example, if the underlying Hive table has skewed columns, this information | ||
| * can't be mapped to [[CatalogTable]] since Spark SQL doesn't handle skewed columns for now. In | ||
| * this case `fullyMapped` is set to false. | ||
| */ | ||
| case class CatalogTable( | ||
| identifier: TableIdentifier, | ||
|
|
@@ -95,7 +101,8 @@ case class CatalogTable( | |
| properties: Map[String, String] = Map.empty, | ||
| viewOriginalText: Option[String] = None, | ||
| viewText: Option[String] = None, | ||
| comment: Option[String] = None) { | ||
| comment: Option[String] = None, | ||
| fullyMapped: Boolean = true) { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think something like
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. +1
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Thanks, renamed to |
||
|
|
||
| // Verify that the provided columns are part of the schema | ||
| private val colNames = schema.map(_.name).toSet | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -626,40 +626,142 @@ case class ShowCreateTableCommand(table: TableIdentifier) extends RunnableComman | |
| val stmt = if (DDLUtils.isDatasourceTable(tableMetadata)) { | ||
| showCreateDataSourceTable(tableMetadata) | ||
| } else { | ||
| throw new UnsupportedOperationException( | ||
| "SHOW CREATE TABLE only supports Spark SQL data source tables.") | ||
| showCreateHiveTable(tableMetadata) | ||
| } | ||
|
|
||
| Seq(Row(stmt)) | ||
| } | ||
|
|
||
| private def showCreateHiveTable(metadata: CatalogTable): String = { | ||
| def reportUnsupportedError(): Unit = { | ||
| throw new UnsupportedOperationException( | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Elsewhere we normally throw
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. also, can you add tests for this case?
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. But this isn't an analysis failure, is it?
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Test case added.
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Let's be consistent with other places and change this to AnalysisException. |
||
| s"Failed to execute SHOW CREATE TABLE against table ${metadata.identifier.quotedString}, " + | ||
| "because it contains table structure(s) (e.g. skewed columns) that Spark SQL doesn't " + | ||
| "support yet." | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think we need to explicitly say that the table was created by Hive with keywords that we do not support.
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Do you mean we should mention the exact keywords, or just saying that the table is created by Hive?
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Replaced |
||
| ) | ||
| } | ||
|
|
||
| if (!metadata.fullyMapped) { | ||
| reportUnsupportedError() | ||
| } | ||
|
|
||
| val builder = StringBuilder.newBuilder | ||
|
|
||
| val tableTypeString = metadata.tableType match { | ||
| case EXTERNAL => " EXTERNAL TABLE" | ||
| case VIEW => " VIEW" | ||
| case MANAGED => " TABLE" | ||
| case INDEX => reportUnsupportedError() | ||
| } | ||
|
|
||
| builder ++= s"CREATE$tableTypeString ${table.quotedString}" | ||
|
|
||
| if (metadata.tableType == VIEW) { | ||
| if (metadata.schema.nonEmpty) { | ||
| builder ++= metadata.schema.map(_.name).mkString("(", ", ", ")") | ||
| } | ||
| builder ++= metadata.viewOriginalText.mkString(" AS\n", "", "\n") | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Looks like we need to use
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Good point, thanks. |
||
| } else { | ||
| showHiveTableDataColumns(metadata, builder) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Should there be table comment clause generated here, after the data column list?
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Thanks. |
||
| showHiveTableNonDataColumns(metadata, builder) | ||
| showHiveTableStorageInfo(metadata, builder) | ||
| showHiveTableProperties(metadata, builder) | ||
| } | ||
|
|
||
| builder.toString() | ||
| } | ||
|
|
||
| private def showHiveTableDataColumns(metadata: CatalogTable, builder: StringBuilder): Unit = { | ||
| val columns = metadata.schema.filterNot { column => | ||
| metadata.partitionColumnNames.contains(column.name) | ||
| }.map(columnToDDLFragment) | ||
|
|
||
| if (columns.nonEmpty) { | ||
| builder ++= columns.mkString("(", ", ", ")\n") | ||
| } | ||
| } | ||
|
|
||
| private def columnToDDLFragment(column: CatalogColumn): String = { | ||
| val comment = column.comment.map(escapeSingleQuotedString).map(" COMMENT '" + _ + "'") | ||
| s"${quoteIdentifier(column.name)} ${column.dataType}${comment.getOrElse("")}" | ||
| } | ||
|
|
||
| private def showHiveTableNonDataColumns(metadata: CatalogTable, builder: StringBuilder): Unit = { | ||
| if (metadata.partitionColumns.nonEmpty) { | ||
| val partCols = metadata.partitionColumns.map(columnToDDLFragment) | ||
| builder ++= partCols.mkString("PARTITIONED BY (", ", ", ")\n") | ||
| } | ||
|
|
||
| if (metadata.bucketColumnNames.nonEmpty) { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. we don't support this I don't think. You can't actually create a table with bucketed columns.
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. if you remove this then you probably don't need this helper method.
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yea. Let's throw an exception since we do not allow users to create bucketed Hive tables. |
||
| builder ++= metadata.bucketColumnNames.mkString("CLUSTERED BY (", ", ", ")\n") | ||
|
|
||
| if (metadata.sortColumnNames.nonEmpty) { | ||
| builder ++= metadata.bucketColumnNames.mkString("SORTED BY (", ", ", ")\n") | ||
| } | ||
|
|
||
| builder ++= s"INTO ${metadata.numBuckets} BUCKETS\n" | ||
| } | ||
| } | ||
|
|
||
| private def showHiveTableStorageInfo(metadata: CatalogTable, builder: StringBuilder): Unit = { | ||
| val storage = metadata.storage | ||
|
|
||
| storage.serde.foreach { serde => | ||
| builder ++= s"ROW FORMAT SERDE '$serde'\n" | ||
|
|
||
| val serdeProps = metadata.storage.serdeProperties.map { | ||
| case (key, value) => | ||
| s"'${escapeSingleQuotedString(key)}' = '${escapeSingleQuotedString(value)}'" | ||
| } | ||
|
|
||
| builder ++= serdeProps.mkString("WITH SERDEPROPERTIES (", ",\n ", "\n)\n") | ||
| } | ||
|
|
||
| if (storage.inputFormat.isDefined || storage.outputFormat.isDefined) { | ||
| builder ++= "STORED AS\n" | ||
|
|
||
| storage.inputFormat.foreach { format => | ||
| builder ++= s" INPUTFORMAT '${escapeSingleQuotedString(format)}'\n" | ||
| } | ||
|
|
||
| storage.outputFormat.foreach { format => | ||
| builder ++= s" OUTPUTFORMAT '${escapeSingleQuotedString(format)}'\n" | ||
| } | ||
| } | ||
|
|
||
| if (metadata.tableType == EXTERNAL) { | ||
| storage.locationUri.foreach { uri => | ||
| builder ++= s"LOCATION '$uri'\n" | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private def showHiveTableProperties(metadata: CatalogTable, builder: StringBuilder): Unit = { | ||
| if (metadata.properties.nonEmpty) { | ||
| val props = metadata.properties.map { case (key, value) => | ||
| s"'${escapeSingleQuotedString(key)}' = '${escapeSingleQuotedString(value)}'" | ||
| } | ||
|
|
||
| builder ++= props.mkString("TBLPROPERTIES (", ",\n ", ")\n") | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Will there be duplicate property key in this list? For example, if the table was created as EXTERNAL table, the generated DDL will be
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Thanks, filtered out this property. |
||
| } | ||
| } | ||
|
|
||
| private def showCreateDataSourceTable(metadata: CatalogTable): String = { | ||
| val builder = StringBuilder.newBuilder | ||
|
|
||
| builder ++= s"CREATE TABLE ${table.quotedString} " | ||
| showDataSourceTableDataCols(metadata, builder) | ||
| showDataSourceTableDataColumns(metadata, builder) | ||
| showDataSourceTableOptions(metadata, builder) | ||
| showDataSourceTableNonDataColumns(metadata, builder) | ||
|
|
||
| builder.toString() | ||
| } | ||
|
|
||
| private def showDataSourceTableDataCols(metadata: CatalogTable, builder: StringBuilder): Unit = { | ||
| val props = metadata.properties | ||
| val schemaParts = for { | ||
| numParts <- props.get("spark.sql.sources.schema.numParts").toSeq | ||
| index <- 0 until numParts.toInt | ||
| } yield props.getOrElse( | ||
| s"spark.sql.sources.schema.part.$index", | ||
| throw new AnalysisException( | ||
| s"Corrupted schema in catalog: $numParts parts expected, but part $index is missing." | ||
| ) | ||
| ) | ||
|
|
||
| if (schemaParts.nonEmpty) { | ||
| val fields = DataType.fromJson(schemaParts.mkString).asInstanceOf[StructType].fields | ||
| val colTypeList = fields.map(f => s"${quoteIdentifier(f.name)} ${f.dataType.sql}") | ||
| builder ++= colTypeList.mkString("(", ", ", ")") | ||
| private def showDataSourceTableDataColumns( | ||
| metadata: CatalogTable, builder: StringBuilder): Unit = { | ||
| DDLUtils.getSchemaFromTableProperties(metadata).foreach { schema => | ||
| val columns = schema.fields.map(f => s"${quoteIdentifier(f.name)} ${f.dataType.sql}") | ||
| builder ++= columns.mkString("(", ", ", ")") | ||
| } | ||
|
|
||
| builder ++= "\n" | ||
|
|
@@ -688,40 +790,21 @@ case class ShowCreateTableCommand(table: TableIdentifier) extends RunnableComman | |
|
|
||
| private def showDataSourceTableNonDataColumns( | ||
| metadata: CatalogTable, builder: StringBuilder): Unit = { | ||
| val props = metadata.properties | ||
|
|
||
| def getColumnNamesByType(colType: String, typeName: String): Seq[String] = { | ||
| (for { | ||
| numCols <- props.get(s"spark.sql.sources.schema.num${colType.capitalize}Cols").toSeq | ||
| index <- 0 until numCols.toInt | ||
| } yield props.getOrElse( | ||
| s"spark.sql.sources.schema.${colType}Col.$index", | ||
| throw new AnalysisException( | ||
| s"Corrupted $typeName in catalog: $numCols parts expected, but part $index is missing." | ||
| ) | ||
| )).map(quoteIdentifier) | ||
| } | ||
|
|
||
| val partCols = getColumnNamesByType("part", "partitioning columns") | ||
| val partCols = DDLUtils.getPartitionColumnsFromTableProperties(metadata) | ||
| if (partCols.nonEmpty) { | ||
| builder ++= s"PARTITIONED BY ${partCols.mkString("(", ", ", ")")}\n" | ||
| } | ||
|
|
||
| val bucketCols = getColumnNamesByType("bucket", "bucketing columns") | ||
| if (bucketCols.nonEmpty) { | ||
| builder ++= s"CLUSTERED BY ${bucketCols.mkString("(", ", ", ")")}\n" | ||
|
|
||
| val sortCols = getColumnNamesByType("sort", "sorting columns") | ||
| if (sortCols.nonEmpty) { | ||
| builder ++= s"SORTED BY ${sortCols.mkString("(", ", ", ")")}\n" | ||
| } | ||
| DDLUtils.getBucketSpecFromTableProperties(metadata).foreach { spec => | ||
| if (spec.bucketColumnNames.nonEmpty) { | ||
| builder ++= s"CLUSTERED BY ${spec.bucketColumnNames.mkString("(", ", ", ")")}\n" | ||
|
|
||
| val numBuckets = props.getOrElse( | ||
| "spark.sql.sources.schema.numBuckets", | ||
| throw new AnalysisException("Corrupted bucket spec in catalog: missing bucket number") | ||
| ) | ||
| if (spec.sortColumnNames.nonEmpty) { | ||
| builder ++= s"SORTED BY ${spec.sortColumnNames.mkString("(", ", ", ")")}\n" | ||
| } | ||
|
|
||
| builder ++= s"INTO $numBuckets BUCKETS\n" | ||
| builder ++= s"INTO ${spec.numBuckets} BUCKETS\n" | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -338,6 +338,14 @@ private[hive] class HiveClientImpl( | |
| // partition columns are part of the schema | ||
| val partCols = h.getPartCols.asScala.map(fromHiveColumn) | ||
| val schema = h.getCols.asScala.map(fromHiveColumn) ++ partCols | ||
|
|
||
| // Skew spec, storage handler, and bucketing info can't be mapped to CatalogTable (yet) | ||
| val fullyMapped = ( | ||
| h.getSkewedColNames.isEmpty | ||
| && h.getStorageHandler == null | ||
| && h.getBucketCols.isEmpty | ||
| ) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nit: can you use |
||
|
|
||
| CatalogTable( | ||
| identifier = TableIdentifier(h.getTableName, Option(h.getDbName)), | ||
| tableType = h.getTableType match { | ||
|
|
@@ -364,7 +372,8 @@ private[hive] class HiveClientImpl( | |
| ), | ||
| properties = h.getParameters.asScala.toMap, | ||
| viewOriginalText = Option(h.getViewOriginalText), | ||
| viewText = Option(h.getViewExpandedText)) | ||
| viewText = Option(h.getViewExpandedText), | ||
| fullyMapped = fullyMapped) | ||
| } | ||
| } | ||
|
|
||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
you can use
@paramhere