Skip to content

[SPARK-20067] [SQL] Unify and Clean Up Desc Commands Using Catalog Interface #17394

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.

Sign up for GitHub

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

Closed
wants to merge 15 commits into from
Closed

[SPARK-20067] [SQL] Unify and Clean Up Desc Commands Using Catalog Interface #17394

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
2ebeac8
fix.
gatorsmile Mar 23, 2017
8720919
improve
gatorsmile Mar 23, 2017
d999147
revert
gatorsmile Mar 23, 2017
3859097
address comments.
gatorsmile Mar 24, 2017
1d72079
fixed the test case
gatorsmile Mar 25, 2017
68bb05c
fix.
gatorsmile Mar 26, 2017
ac3f351
migrate the test case
gatorsmile Mar 26, 2017
bef1134
remove :
gatorsmile Mar 28, 2017
29817ea
test case fix.
gatorsmile Mar 28, 2017
6c56041
address comments.
gatorsmile Mar 28, 2017
36b501e
fix.
gatorsmile Mar 29, 2017
e116018
address comments.
gatorsmile Apr 1, 2017
a6db8a3
address comments.
gatorsmile Apr 1, 2017
43668be
fix the test cases
gatorsmile Apr 2, 2017
862a4d7
address comments.
gatorsmile Apr 4, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
136 changes: 87 additions & 49 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@ package org.apache.spark.sql.catalyst.catalog
import java.net.URI
import java.util.Date

import scala.collection.mutable

import com.google.common.base.Objects

import org.apache.spark.sql.AnalysisException
Expand Down Expand Up @@ -57,20 +59,25 @@ case class CatalogStorageFormat(
properties: Map[String, String]) {

override def toString: String = {
val serdePropsToString = CatalogUtils.maskCredentials(properties) match {
case props if props.isEmpty => ""
case props => "Properties: " + props.map(p => p._1 + "=" + p._2).mkString("[", ", ", "]")
}
val output =
Seq(locationUri.map("Location: " + _).getOrElse(""),
inputFormat.map("InputFormat: " + _).getOrElse(""),
outputFormat.map("OutputFormat: " + _).getOrElse(""),
if (compressed) "Compressed" else "",
serde.map("Serde: " + _).getOrElse(""),
serdePropsToString)
output.filter(_.nonEmpty).mkString("Storage(", ", ", ")")
toLinkedHashMap.map { case ((key, value)) =>
if (value.isEmpty) key else s"$key: $value"
}.mkString("Storage(", ", ", ")")
}

def toLinkedHashMap: mutable.LinkedHashMap[String, String] = {
val map = new mutable.LinkedHashMap[String, String]()
locationUri.foreach(l => map.put("Location", l.toString))
serde.foreach(map.put("Serde Library", _))
inputFormat.foreach(map.put("InputFormat", _))
outputFormat.foreach(map.put("OutputFormat", _))
if (compressed) map.put("Compressed", "")
CatalogUtils.maskCredentials(properties) match {
case props if props.isEmpty => // No-op
case props =>
map.put("Properties", props.map(p => p._1 + "=" + p._2).mkString("[", ", ", "]"))
}
map
}
}

object CatalogStorageFormat {
Expand All @@ -91,15 +98,28 @@ case class CatalogTablePartition(
storage: CatalogStorageFormat,
parameters: Map[String, String] = Map.empty) {

override def toString: String = {
def toLinkedHashMap: mutable.LinkedHashMap[String, String] = {
val map = new mutable.LinkedHashMap[String, String]()
val specString = spec.map { case (k, v) => s"$k=$v" }.mkString(", ")
val output =
Seq(
s"Partition Values: [$specString]",
s"$storage",
s"Partition Parameters:{${parameters.map(p => p._1 + "=" + p._2).mkString(", ")}}")
map.put("Partition Values", s"[$specString]")
map ++= storage.toLinkedHashMap
if (parameters.nonEmpty) {
map.put("Partition Parameters", s"{${parameters.map(p => p._1 + "=" + p._2).mkString(", ")}}")
}
map
}

output.filter(_.nonEmpty).mkString("CatalogPartition(\n\t", "\n\t", ")")
override def toString: String = {
toLinkedHashMap.map { case ((key, value)) =>
if (value.isEmpty) key else s"$key: $value"
}.mkString("CatalogPartition(\n\t", "\n\t", ")")
}

/** Readable string representation for the CatalogTablePartition. */
def simpleString: String = {
toLinkedHashMap.map { case ((key, value)) =>
if (value.isEmpty) key else s"$key: $value"
}.mkString("", "\n", "")
}

/** Return the partition location, assuming it is specified. */
Expand Down Expand Up @@ -154,6 +174,14 @@ case class BucketSpec(
}
s"$numBuckets buckets, $bucketString$sortString"
}

def toLinkedHashMap: mutable.LinkedHashMap[String, String] = {
mutable.LinkedHashMap[String, String](
"Num Buckets" -> numBuckets.toString,
"Bucket Columns" -> bucketColumnNames.map(quoteIdentifier).mkString("[", ", ", "]"),
"Sort Columns" -> sortColumnNames.map(quoteIdentifier).mkString("[", ", ", "]")
)
}
}

/**
Expand Down Expand Up @@ -261,40 +289,50 @@ case class CatalogTable(
locationUri, inputFormat, outputFormat, serde, compressed, properties))
}

override def toString: String = {

def toLinkedHashMap: mutable.LinkedHashMap[String, String] = {
val map = new mutable.LinkedHashMap[String, String]()
val tableProperties = properties.map(p => p._1 + "=" + p._2).mkString("[", ", ", "]")
val partitionColumns = partitionColumnNames.map(quoteIdentifier).mkString("[", ", ", "]")
val bucketStrings = bucketSpec match {
case Some(BucketSpec(numBuckets, bucketColumnNames, sortColumnNames)) =>
val bucketColumnsString = bucketColumnNames.map(quoteIdentifier).mkString("[", ", ", "]")
val sortColumnsString = sortColumnNames.map(quoteIdentifier).mkString("[", ", ", "]")
Seq(
s"Num Buckets: $numBuckets",
if (bucketColumnNames.nonEmpty) s"Bucket Columns: $bucketColumnsString" else "",
if (sortColumnNames.nonEmpty) s"Sort Columns: $sortColumnsString" else ""
)

case _ => Nil

identifier.database.foreach(map.put("Database", _))
map.put("Table", identifier.table)
if (owner.nonEmpty) map.put("Owner", owner)
map.put("Created", new Date(createTime).toString)
map.put("Last Access", new Date(lastAccessTime).toString)
map.put("Type", tableType.name)
provider.foreach(map.put("Provider", _))
bucketSpec.foreach(map ++= _.toLinkedHashMap)
comment.foreach(map.put("Comment", _))
if (tableType == CatalogTableType.VIEW) {
viewText.foreach(map.put("View Text", _))
viewDefaultDatabase.foreach(map.put("View Default Database", _))
if (viewQueryColumnNames.nonEmpty) {
map.put("View Query Output Columns", viewQueryColumnNames.mkString("[", ", ", "]"))
}
}

val output =
Seq(s"Table: ${identifier.quotedString}",
if (owner.nonEmpty) s"Owner: $owner" else "",
s"Created: ${new Date(createTime).toString}",
s"Last Access: ${new Date(lastAccessTime).toString}",
s"Type: ${tableType.name}",
if (schema.nonEmpty) s"Schema: ${schema.mkString("[", ", ", "]")}" else "",
if (provider.isDefined) s"Provider: ${provider.get}" else "",
if (partitionColumnNames.nonEmpty) s"Partition Columns: $partitionColumns" else ""
) ++ bucketStrings ++ Seq(
viewText.map("View: " + _).getOrElse(""),
comment.map("Comment: " + _).getOrElse(""),
if (properties.nonEmpty) s"Properties: $tableProperties" else "",
if (stats.isDefined) s"Statistics: ${stats.get.simpleString}" else "",
s"$storage",
if (tracksPartitionsInCatalog) "Partition Provider: Catalog" else "")

output.filter(_.nonEmpty).mkString("CatalogTable(\n\t", "\n\t", ")")
if (properties.nonEmpty) map.put("Properties", tableProperties)
stats.foreach(s => map.put("Statistics", s.simpleString))
map ++= storage.toLinkedHashMap
if (tracksPartitionsInCatalog) map.put("Partition Provider", "Catalog")
if (partitionColumnNames.nonEmpty) map.put("Partition Columns", partitionColumns)
if (schema.nonEmpty) map.put("Schema", schema.treeString)

map
}

override def toString: String = {
toLinkedHashMap.map { case ((key, value)) =>
if (value.isEmpty) key else s"$key: $value"
}.mkString("CatalogTable(\n", "\n", ")")
}

/** Readable string representation for the CatalogTable. */
def simpleString: String = {
toLinkedHashMap.map { case ((key, value)) =>
if (value.isEmpty) key else s"$key: $value"
}.mkString("", "\n", "")
}
}

Expand Down
3 changes: 1 addition & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -322,8 +322,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
DescribeTableCommand(
visitTableIdentifier(ctx.tableIdentifier),
partitionSpec,
ctx.EXTENDED != null,
ctx.FORMATTED != null)
ctx.EXTENDED != null || ctx.FORMATTED != null)
}
}

Expand Down
124 changes: 27 additions & 97 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -500,8 +500,7 @@ case class TruncateTableCommand(
case class DescribeTableCommand(
table: TableIdentifier,
partitionSpec: TablePartitionSpec,
isExtended: Boolean,
Copy link
Contributor

@cloud-fan cloud-fan Mar 31, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

seems DESC EXTENDED is better, normal DESC tbl is also formated

Copy link
Member Author

@gatorsmile gatorsmile Mar 31, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure, let me make the change.

isFormatted: Boolean)
isExtended: Boolean)
extends RunnableCommand {

override val output: Seq[Attribute] = Seq(
Expand Down Expand Up @@ -536,14 +535,12 @@ case class DescribeTableCommand(

describePartitionInfo(metadata, result)

if (partitionSpec.isEmpty) {
if (isExtended) {
describeExtendedTableInfo(metadata, result)
} else if (isFormatted) {
describeFormattedTableInfo(metadata, result)
}
} else {
if (partitionSpec.nonEmpty) {
// Outputs the partition-specific info for the DDL command:
// "DESCRIBE [EXTENDED|FORMATTED] table_name PARTITION (partitionVal*)"
describeDetailedPartitionInfo(sparkSession, catalog, metadata, result)
} else if (isExtended) {
describeFormattedTableInfo(metadata, result)
}
}

Expand All @@ -553,76 +550,20 @@ case class DescribeTableCommand(
private def describePartitionInfo(table: CatalogTable, buffer: ArrayBuffer[Row]): Unit = {
if (table.partitionColumnNames.nonEmpty) {
append(buffer, "# Partition Information", "", "")
append(buffer, s"# ${output.head.name}", output(1).name, output(2).name)
describeSchema(table.partitionSchema, buffer)
}
}

private def describeExtendedTableInfo(table: CatalogTable, buffer: ArrayBuffer[Row]): Unit = {
append(buffer, "", "", "")
append(buffer, "# Detailed Table Information", table.toString, "")
}

private def describeFormattedTableInfo(table: CatalogTable, buffer: ArrayBuffer[Row]): Unit = {
// The following information has been already shown in the previous outputs
val excludedTableInfo = Seq(
"Partition Columns",
"Schema"
)
append(buffer, "", "", "")
append(buffer, "# Detailed Table Information", "", "")
append(buffer, "Database:", table.database, "")
append(buffer, "Owner:", table.owner, "")
append(buffer, "Created:", new Date(table.createTime).toString, "")
append(buffer, "Last Access:", new Date(table.lastAccessTime).toString, "")
append(buffer, "Location:", table.storage.locationUri.map(CatalogUtils.URIToString(_))
.getOrElse(""), "")
append(buffer, "Table Type:", table.tableType.name, "")
append(buffer, "Comment:", table.comment.getOrElse(""), "")
table.stats.foreach(s => append(buffer, "Statistics:", s.simpleString, ""))

append(buffer, "Table Parameters:", "", "")
table.properties.foreach { case (key, value) =>
append(buffer, s" $key", value, "")
}

describeStorageInfo(table, buffer)

if (table.tableType == CatalogTableType.VIEW) describeViewInfo(table, buffer)

if (DDLUtils.isDatasourceTable(table) && table.tracksPartitionsInCatalog) {
append(buffer, "Partition Provider:", "Catalog", "")
}
}

private def describeStorageInfo(metadata: CatalogTable, buffer: ArrayBuffer[Row]): Unit = {
append(buffer, "", "", "")
append(buffer, "# Storage Information", "", "")
metadata.storage.serde.foreach(serdeLib => append(buffer, "SerDe Library:", serdeLib, ""))
metadata.storage.inputFormat.foreach(format => append(buffer, "InputFormat:", format, ""))
metadata.storage.outputFormat.foreach(format => append(buffer, "OutputFormat:", format, ""))
append(buffer, "Compressed:", if (metadata.storage.compressed) "Yes" else "No", "")
describeBucketingInfo(metadata, buffer)

append(buffer, "Storage Desc Parameters:", "", "")
val maskedProperties = CatalogUtils.maskCredentials(metadata.storage.properties)
maskedProperties.foreach { case (key, value) =>
append(buffer, s" $key", value, "")
}
}

private def describeViewInfo(metadata: CatalogTable, buffer: ArrayBuffer[Row]): Unit = {
append(buffer, "", "", "")
append(buffer, "# View Information", "", "")
append(buffer, "View Text:", metadata.viewText.getOrElse(""), "")
append(buffer, "View Default Database:", metadata.viewDefaultDatabase.getOrElse(""), "")
append(buffer, "View Query Output Columns:",
metadata.viewQueryColumnNames.mkString("[", ", ", "]"), "")
}

private def describeBucketingInfo(metadata: CatalogTable, buffer: ArrayBuffer[Row]): Unit = {
metadata.bucketSpec match {
case Some(BucketSpec(numBuckets, bucketColumnNames, sortColumnNames)) =>
append(buffer, "Num Buckets:", numBuckets.toString, "")
append(buffer, "Bucket Columns:", bucketColumnNames.mkString("[", ", ", "]"), "")
append(buffer, "Sort Columns:", sortColumnNames.mkString("[", ", ", "]"), "")

case _ =>
table.toLinkedHashMap.filterKeys(!excludedTableInfo.contains(_)).foreach {
s => append(buffer, s._1, s._2, "")
}
}

Expand All @@ -637,21 +578,7 @@ case class DescribeTableCommand(
}
DDLUtils.verifyPartitionProviderIsHive(spark, metadata, "DESC PARTITION")
val partition = catalog.getPartition(table, partitionSpec)
if (isExtended) {
describeExtendedDetailedPartitionInfo(table, metadata, partition, result)
} else if (isFormatted) {
describeFormattedDetailedPartitionInfo(table, metadata, partition, result)
describeStorageInfo(metadata, result)
}
}

private def describeExtendedDetailedPartitionInfo(
tableIdentifier: TableIdentifier,
table: CatalogTable,
partition: CatalogTablePartition,
buffer: ArrayBuffer[Row]): Unit = {
append(buffer, "", "", "")
append(buffer, "Detailed Partition Information " + partition.toString, "", "")
if (isExtended) describeFormattedDetailedPartitionInfo(table, metadata, partition, result)
Copy link
Contributor

@cloud-fan cloud-fan Apr 1, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

not related to this PR, but it looks weird that DESC tbl and DESC tbl PARTITION (xxx) has the same result.

Copy link
Member Author

@gatorsmile gatorsmile Apr 1, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This function describeDetailedPartitionInfo will only be called for the DDL command

  DESCRIBE [EXTENDED|FORMATTED] table_name PARTITION (partitionVal*)

}

private def describeFormattedDetailedPartitionInfo(
Expand All @@ -661,18 +588,21 @@ case class DescribeTableCommand(
buffer: ArrayBuffer[Row]): Unit = {
append(buffer, "", "", "")
append(buffer, "# Detailed Partition Information", "", "")
append(buffer, "Partition Value:", s"[${partition.spec.values.mkString(", ")}]", "")
append(buffer, "Database:", table.database, "")
append(buffer, "Table:", tableIdentifier.table, "")
append(buffer, "Location:", partition.storage.locationUri.map(CatalogUtils.URIToString(_))
.getOrElse(""), "")
append(buffer, "Partition Parameters:", "", "")
partition.parameters.foreach { case (key, value) =>
append(buffer, s" $key", value, "")
append(buffer, "Database", table.database, "")
append(buffer, "Table", tableIdentifier.table, "")
partition.toLinkedHashMap.foreach(s => append(buffer, s._1, s._2, ""))
append(buffer, "", "", "")
append(buffer, "# Storage Information", "", "")
table.bucketSpec match {
case Some(spec) =>
spec.toLinkedHashMap.foreach(s => append(buffer, s._1, s._2, ""))
case _ =>
}
table.storage.toLinkedHashMap.foreach(s => append(buffer, s._1, s._2, ""))
}

private def describeSchema(schema: StructType, buffer: ArrayBuffer[Row]): Unit = {
append(buffer, s"# ${output.head.name}", output(1).name, output(2).name)
schema.foreach { column =>
append(buffer, column.name, column.dataType.simpleString, column.getComment().orNull)
}
Expand Down Expand Up @@ -728,7 +658,7 @@ case class ShowTablesCommand(
val tableName = tableIdent.table
val isTemp = catalog.isTemporaryTable(tableIdent)
if (isExtended) {
val information = catalog.getTempViewOrPermanentTableMetadata(tableIdent).toString
val information = catalog.getTempViewOrPermanentTableMetadata(tableIdent).simpleString
Row(database, tableName, isTemp, s"$information\n")
} else {
Row(database, tableName, isTemp)
Expand All @@ -745,7 +675,7 @@ case class ShowTablesCommand(
val database = table.database.getOrElse("")
val tableName = table.table
val isTemp = catalog.isTemporaryTable(table)
val information = partition.toString
val information = partition.simpleString
Seq(Row(database, tableName, isTemp, s"$information\n"))
}
}
Expand Down
Loading