[SPARK-14348][SQL] Support native execution of SHOW TBLPROPERTIES command

## What changes were proposed in this pull request?

This PR adds Native execution of SHOW TBLPROPERTIES command.

Command Syntax:
``` SQL
SHOW TBLPROPERTIES table_name[(property_key_literal)]
```
## How was this patch tested?

Tests added in HiveComandSuiie and DDLCommandSuite

Author: Dilip Biswal <dbiswal@us.ibm.com>

Closes #12133 from dilipbiswal/dkb_show_tblproperties.

7 files changed, 219 insertions, 30 deletions

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 6cf47b5c30..27b01e0bed 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
@@ -116,6 +116,8 @@ statement
     | SHOW TABLES ((FROM | IN) db=identifier)?
         (LIKE? pattern=STRING)?                                        #showTables
     | SHOW DATABASES (LIKE pattern=STRING)?                            #showDatabases
+    | SHOW TBLPROPERTIES table=tableIdentifier
+        ('(' key=tablePropertyKey ')')?                                #showTblProperties
     | SHOW FUNCTIONS (LIKE? (qualifiedName | pattern=STRING))?         #showFunctions
     | (DESC | DESCRIBE) FUNCTION EXTENDED? qualifiedName               #describeFunction
     | (DESC | DESCRIBE) option=(EXTENDED | FORMATTED)?
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
index 569b99e414..3b8ce6373d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
@@ -163,6 +163,7 @@ class SessionCatalog(
   /**
    * Retrieve the metadata of an existing metastore table.
    * If no database is specified, assume the table is in the current database.
+   * If the specified table is not found in the database then an [[AnalysisException]] is thrown.
    */
   def getTable(name: TableIdentifier): CatalogTable = {
     val db = name.database.getOrElse(currentDb)
@@ -272,6 +273,16 @@ class SessionCatalog(
   }
 
   /**
+   * Return whether a table with the specified name is a temporary table.
+   *
+   * Note: The temporary table cache is checked only when database is not
+   * explicitly specified.
+   */
+  def isTemporaryTable(name: TableIdentifier): Boolean = {
+    !name.database.isDefined && tempTables.contains(formatTableName(name.table))
+  }
+
+  /**
    * List all tables in the specified database, including temporary tables.
    */
   def listTables(db: String): Seq[TableIdentifier] = listTables(db, "*")
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 ff3ab7746c..fb106d1aef 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
@@ -93,6 +93,22 @@ class SparkSqlAstBuilder extends AstBuilder {
   }
 
   /**
+   * A command for users to list the properties for a table. If propertyKey is specified, the value
+   * for the propertyKey is returned. If propertyKey is not specified, all the keys and their
+   * corresponding values are returned.
+   * The syntax of using this command in SQL is:
+   * {{{
+   *   SHOW TBLPROPERTIES table_name[('propertyKey')];
+   * }}}
+   */
+  override def visitShowTblProperties(
+      ctx: ShowTblPropertiesContext): LogicalPlan = withOrigin(ctx) {
+    ShowTablePropertiesCommand(
+      visitTableIdentifier(ctx.tableIdentifier),
+      Option(ctx.key).map(visitTablePropertyKey))
+  }
+
+  /**
    * Create a [[RefreshTable]] logical plan.
    */
   override def visitRefreshTable(ctx: RefreshTableContext): LogicalPlan = withOrigin(ctx) {
@@ -220,19 +236,26 @@ class SparkSqlAstBuilder extends AstBuilder {
   override def visitTablePropertyList(
       ctx: TablePropertyListContext): Map[String, String] = withOrigin(ctx) {
     ctx.tableProperty.asScala.map { property =>
-      // A key can either be a String or a collection of dot separated elements. We need to treat
-      // these differently.
-      val key = if (property.key.STRING != null) {
-        string(property.key.STRING)
-      } else {
-        property.key.getText
-      }
+      val key = visitTablePropertyKey(property.key)
       val value = Option(property.value).map(string).orNull
       key -> value
     }.toMap
   }
 
   /**
+   * A table property key can either be String or a collection of dot separated elements. This
+   * function extracts the property key based on whether its a string literal or a table property
+   * identifier.
+   */
+  override def visitTablePropertyKey(key: TablePropertyKeyContext): String = {
+    if (key.STRING != null) {
+      string(key.STRING)
+    } else {
+      key.getText
+    }
+  }
+
+  /**
    * Create a [[CreateDatabase]] command.
    *
    * For example:
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/commands.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/commands.scala
index 4eb8d7ff0d..a4be3bc333 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/commands.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/commands.scala
@@ -21,7 +21,7 @@ import java.util.NoSuchElementException
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{Dataset, Row, SQLContext}
+import org.apache.spark.sql.{AnalysisException, Dataset, Row, SQLContext}
 import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow, TableIdentifier}
 import org.apache.spark.sql.catalyst.errors.TreeNodeException
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
@@ -381,6 +381,48 @@ case class ShowDatabasesCommand(databasePattern: Option[String]) extends Runnabl
 }
 
 /**
+ * A command for users to list the properties for a table If propertyKey is specified, the value
+ * for the propertyKey is returned. If propertyKey is not specified, all the keys and their
+ * corresponding values are returned.
+ * The syntax of using this command in SQL is:
+ * {{{
+ *   SHOW TBLPROPERTIES table_name[('propertyKey')];
+ * }}}
+ */
+case class ShowTablePropertiesCommand(
+    table: TableIdentifier,
+    propertyKey: Option[String]) extends RunnableCommand {
+
+  override val output: Seq[Attribute] = {
+    val schema = AttributeReference("value", StringType, nullable = false)() :: Nil
+    propertyKey match {
+      case None => AttributeReference("key", StringType, nullable = false)() :: schema
+      case _ => schema
+    }
+  }
+
+  override def run(sqlContext: SQLContext): Seq[Row] = {
+    val catalog = sqlContext.sessionState.catalog
+
+    if (catalog.isTemporaryTable(table)) {
+      Seq.empty[Row]
+    } else {
+      val catalogTable = sqlContext.sessionState.catalog.getTable(table)
+
+      propertyKey match {
+        case Some(p) =>
+          val propValue = catalogTable
+            .properties
+            .getOrElse(p, s"Table ${catalogTable.qualifiedName} does not have property: $p")
+          Seq(Row(propValue))
+        case None =>
+          catalogTable.properties.map(p => Row(p._1, p._2)).toSeq
+      }
+    }
+  }
+}
+
+/**
  * A command for users to list all of the registered functions.
  * The syntax of using this command in SQL is:
  * {{{
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 458f36e832..8b2a5979e2 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
@@ -773,4 +773,12 @@ class DDLCommandSuite extends PlanTest {
     comparePlans(parsed2, expected2)
   }
 
+  test("show tblproperties") {
+    val parsed1 = parser.parsePlan("SHOW TBLPROPERTIES tab1")
+    val expected1 = ShowTablePropertiesCommand(TableIdentifier("tab1", None), None)
+    val parsed2 = parser.parsePlan("SHOW TBLPROPERTIES tab1('propKey1')")
+    val expected2 = ShowTablePropertiesCommand(TableIdentifier("tab1", None), Some("propKey1"))
+    comparePlans(parsed1, expected1)
+    comparePlans(parsed2, expected2)
+  }
 }
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala
new file mode 100644
index 0000000000..4c3f450522
--- /dev/null
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala
@@ -0,0 +1,125 @@
+/*
+ * 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.hive.execution
+
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.hive.test.TestHiveSingleton
+import org.apache.spark.sql.test.SQLTestUtils
+
+class HiveCommandSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
+   protected override def beforeAll(): Unit = {
+    super.beforeAll()
+    sql(
+      """
+        |CREATE EXTERNAL TABLE parquet_tab1 (c1 INT, c2 STRING)
+        |USING org.apache.spark.sql.parquet.DefaultSource
+      """.stripMargin)
+
+     sql(
+      """
+        |CREATE EXTERNAL TABLE parquet_tab2 (c1 INT, c2 STRING)
+        |STORED AS PARQUET
+        |TBLPROPERTIES('prop1Key'="prop1Val", '`prop2Key`'="prop2Val")
+      """.stripMargin)
+  }
+
+  override protected def afterAll(): Unit = {
+    try {
+      sql("DROP TABLE IF EXISTS parquet_tab1")
+      sql("DROP TABLE IF EXISTS parquet_tab2")
+    } finally {
+      super.afterAll()
+    }
+  }
+
+  test("show tables") {
+    withTable("show1a", "show2b") {
+      sql("CREATE TABLE show1a(c1 int)")
+      sql("CREATE TABLE show2b(c2 int)")
+      checkAnswer(
+        sql("SHOW TABLES IN default 'show1*'"),
+        Row("show1a", false) :: Nil)
+      checkAnswer(
+        sql("SHOW TABLES IN default 'show1*|show2*'"),
+        Row("show1a", false) ::
+          Row("show2b", false) :: Nil)
+      checkAnswer(
+        sql("SHOW TABLES 'show1*|show2*'"),
+        Row("show1a", false) ::
+          Row("show2b", false) :: Nil)
+      assert(
+        sql("SHOW TABLES").count() >= 2)
+      assert(
+        sql("SHOW TABLES IN default").count() >= 2)
+    }
+  }
+
+  test("show tblproperties of data source tables - basic") {
+    checkAnswer(
+      sql("SHOW TBLPROPERTIES parquet_tab1")
+        .filter(s"key = 'spark.sql.sources.provider'"),
+      Row("spark.sql.sources.provider", "org.apache.spark.sql.parquet.DefaultSource") :: Nil
+    )
+
+    checkAnswer(
+      sql("SHOW TBLPROPERTIES parquet_tab1(spark.sql.sources.provider)"),
+      Row("org.apache.spark.sql.parquet.DefaultSource") :: Nil
+    )
+
+    checkAnswer(
+      sql("SHOW TBLPROPERTIES parquet_tab1")
+        .filter(s"key = 'spark.sql.sources.schema.numParts'"),
+      Row("spark.sql.sources.schema.numParts", "1") :: Nil
+    )
+
+    checkAnswer(
+      sql("SHOW TBLPROPERTIES parquet_tab1('spark.sql.sources.schema.numParts')"),
+      Row("1"))
+  }
+
+  test("show tblproperties for datasource table - errors") {
+    val message1 = intercept[AnalysisException] {
+      sql("SHOW TBLPROPERTIES badtable")
+    }.getMessage
+    assert(message1.contains("Table badtable not found in database default"))
+
+    // When key is not found, a row containing the error is returned.
+    checkAnswer(
+      sql("SHOW TBLPROPERTIES parquet_tab1('invalid.prop.key')"),
+      Row("Table default.parquet_tab1 does not have property: invalid.prop.key") :: Nil
+    )
+  }
+
+  test("show tblproperties for hive table") {
+    checkAnswer(sql("SHOW TBLPROPERTIES parquet_tab2('prop1Key')"), Row("prop1Val"))
+    checkAnswer(sql("SHOW TBLPROPERTIES parquet_tab2('`prop2Key`')"), Row("prop2Val"))
+  }
+
+  test("show tblproperties for spark temporary table - empty row") {
+    withTempTable("parquet_temp") {
+      sql(
+        """
+          |CREATE TEMPORARY TABLE parquet_temp (c1 INT, c2 STRING)
+          |USING org.apache.spark.sql.parquet.DefaultSource
+        """.stripMargin)
+
+      // An empty sequence of row is returned for session temporary table.
+      checkAnswer(sql("SHOW TBLPROPERTIES parquet_temp"), Nil)
+    }
+  }
+}
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 c203518fdd..6199253d34 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
@@ -1811,26 +1811,4 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
       }
     }
   }
-
-  test("show tables") {
-    withTable("show1a", "show2b") {
-      sql("CREATE TABLE show1a(c1 int)")
-      sql("CREATE TABLE show2b(c2 int)")
-      checkAnswer(
-        sql("SHOW TABLES IN default 'show1*'"),
-        Row("show1a", false) :: Nil)
-      checkAnswer(
-        sql("SHOW TABLES IN default 'show1*|show2*'"),
-        Row("show1a", false) ::
-          Row("show2b", false) :: Nil)
-      checkAnswer(
-        sql("SHOW TABLES 'show1*|show2*'"),
-        Row("show1a", false) ::
-          Row("show2b", false) :: Nil)
-      assert(
-        sql("SHOW TABLES").count() >= 2)
-      assert(
-        sql("SHOW TABLES IN default").count() >= 2)
-    }
-  }
 }