[SPARK-34710][SQL] Add tableType column for SHOW TABLES to distinguish view and tables

docs/sql-migration-guide.md

@@ -45,7 +45,7 @@ license: |
 
   - In Spark 3.2, the auto-generated `Cast` (such as those added by type coercion rules) will be stripped when generating column alias names. E.g., `sql("SELECT floor(1)").columns` will be `FLOOR(1)` instead of `FLOOR(CAST(1 AS DOUBLE))`.
 
-  - In Spark 3.2, the output schema of `SHOW TABLES` becomes `namespace: string, tableName: string, isTemporary: boolean`. In Spark 3.1 or earlier, the `namespace` field was named `database` for the builtin catalog, and there is no `isTemporary` field for v2 catalogs. To restore the old schema with the builtin catalog, you can set `spark.sql.legacy.keepCommandOutputSchema` to `true`.
+  - In Spark 3.2, the output schema of `SHOW TABLES` becomes `namespace: string, tableName: string, isTemporary: boolean, tableType: string`. In Spark 3.1 or earlier, the `namespace` field was named `database` for the builtin catalog, and there is no `isTemporary` field for v2 catalogs and `tableType` for both v1 and v2 catalogs. To restore the old schema with the builtin catalog, you can set `spark.sql.legacy.keepCommandOutputSchema` to `true`.
 
   - In Spark 3.2, the output schema of `SHOW TABLE EXTENDED` becomes `namespace: string, tableName: string, isTemporary: boolean, information: string`. In Spark 3.1 or earlier, the `namespace` field was named `database` for the builtin catalog, and no change for the v2 catalogs. To restore the old schema with the builtin catalog, you can set `spark.sql.legacy.keepCommandOutputSchema` to `true`.
 

python/pyspark/sql/context.py

@@ -478,7 +478,7 @@ def tables(self, dbName=None):
         >>> sqlContext.registerDataFrameAsTable(df, "table1")
         >>> df2 = sqlContext.tables()
         >>> df2.filter("tableName = 'table1'").first()
-        Row(namespace='', tableName='table1', isTemporary=True)
+        Row(namespace='', tableName='table1', isTemporary=True, tableType='VIEW')
         """
         if dbName is None:
             return DataFrame(self._ssql_ctx.tables(), self)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala

@@ -742,6 +742,13 @@ object CatalogTableType {
   val VIEW = new CatalogTableType("VIEW")
 
   val tableTypes = Seq(EXTERNAL, MANAGED, VIEW)
+
+  def classicTableTypeString(tableType: CatalogTableType): String = tableType match {
+    case EXTERNAL | MANAGED => "TABLE"
+    case VIEW => "VIEW"
+    case t =>
+      throw new IllegalArgumentException(s"Unknown table type is found: $t")
+  }
 }
 
 

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

@@ -515,7 +515,13 @@ object ShowTables {
   def getOutputAttrs: Seq[Attribute] = Seq(
     AttributeReference("namespace", StringType, nullable = false)(),
     AttributeReference("tableName", StringType, nullable = false)(),
-    AttributeReference("isTemporary", BooleanType, nullable = false)())
+    AttributeReference("isTemporary", BooleanType, nullable = false)(),
+    AttributeReference("tableType", StringType, nullable = false)())
+
+  def getLegacyOutputAttrs: Seq[Attribute] = {
+    val output = getOutputAttrs
+    output.head.withName("database") +: output.slice(1, output.length - 1)
+  }
 }
 
 /**
@@ -534,7 +540,13 @@ object ShowTableExtended {
     AttributeReference("namespace", StringType, nullable = false)(),
     AttributeReference("tableName", StringType, nullable = false)(),
     AttributeReference("isTemporary", BooleanType, nullable = false)(),
-    AttributeReference("information", StringType, nullable = false)())
+    AttributeReference("information", StringType, nullable = false)(),
+    AttributeReference("tableType", StringType, nullable = false)())
+
+  def getLegacyOutputAttrs: Seq[Attribute] = {
+    val output = getOutputAttrs
+    output.head.withName("database") +: output.slice(1, output.length - 1)
+  }
 }
 
 /**

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -3095,7 +3095,7 @@ object SQLConf {
     buildConf("spark.sql.legacy.keepCommandOutputSchema")
       .internal()
       .doc("When true, Spark will keep the output schema of commands such as SHOW DATABASES " +
-        "unchanged, for v1 catalog and/or table.")
+        "as same as Spark 3.0 and earlier, for v1 catalog and/or table.")
       .version("3.0.2")
       .booleanConf
       .createWithDefault(false)

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala

@@ -353,8 +353,8 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
 
     case ShowTables(DatabaseInSessionCatalog(db), pattern, output) =>
       val newOutput = if (conf.getConf(SQLConf.LEGACY_KEEP_COMMAND_OUTPUT_SCHEMA)) {
-        assert(output.length == 3)
-        output.head.withName("database") +: output.tail
+        assert(output.length == 4)
+        ShowTables.getLegacyOutputAttrs
       } else {
         output
       }
@@ -366,8 +366,8 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
         partitionSpec @ (None | Some(UnresolvedPartitionSpec(_, _))),
         output) =>
       val newOutput = if (conf.getConf(SQLConf.LEGACY_KEEP_COMMAND_OUTPUT_SCHEMA)) {
-        assert(output.length == 4)
-        output.head.withName("database") +: output.tail
+        assert(output.length == 5)
+        ShowTableExtended.getLegacyOutputAttrs
       } else {
         output
       }

sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala

@@ -843,11 +843,22 @@ case class ShowTablesCommand(
         val database = tableIdent.database.getOrElse("")
         val tableName = tableIdent.table
         val isTemp = catalog.isTempView(tableIdent)
+        val catalogTable = catalog.getTempViewOrPermanentTableMetadata(tableIdent)
+        val tableType = classicTableTypeString(catalogTable.tableType)
         if (isExtended) {
-          val information = catalog.getTempViewOrPermanentTableMetadata(tableIdent).simpleString
-          Row(database, tableName, isTemp, s"$information\n")
+          val information = catalogTable.simpleString
+          if (output.size == 5) {
+            Row(database, tableName, isTemp, s"$information\n", tableType)
+          } else {
+            Row(database, tableName, isTemp, s"$information\n")
+          }
         } else {
-          Row(database, tableName, isTemp)
+          if (output.size == 4) {
+            Row(database, tableName, isTemp, tableType)
+          } else {
+            Row(database, tableName, isTemp)
+          }
+
         }
       }
     } else {
@@ -870,7 +881,12 @@ case class ShowTablesCommand(
       val tableName = tableIdent.table
       val isTemp = catalog.isTempView(tableIdent)
       val information = partition.simpleString
-      Seq(Row(database, tableName, isTemp, s"$information\n"))
+      if (output.size == 5) {
+        val tableType = classicTableTypeString(table.tableType)
+        Seq(Row(database, tableName, isTemp, s"$information\n", tableType))
+      } else {
+        Seq(Row(database, tableName, isTemp, s"$information\n"))
+      }
     }
   }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowTablesExec.scala

@@ -40,7 +40,7 @@ case class ShowTablesExec(
     val tables = catalog.listTables(namespace.toArray)
     tables.map { table =>
       if (pattern.map(StringUtils.filterPattern(Seq(table.name()), _).nonEmpty).getOrElse(true)) {
-        rows += toCatalystRow(table.namespace().quoted, table.name(), false)
+        rows += toCatalystRow(table.namespace().quoted, table.name(), false, "TABLE")
       }
     }
 

sql/core/src/test/resources/log4j.properties

@@ -22,7 +22,7 @@ log4j.rootLogger=INFO, CA, FA
 log4j.appender.CA=org.apache.log4j.ConsoleAppender
 log4j.appender.CA.layout=org.apache.log4j.PatternLayout
 log4j.appender.CA.layout.ConversionPattern=%d{HH:mm:ss.SSS} %p %c: %m%n
-log4j.appender.CA.Threshold = WARN
+log4j.appender.CA.Threshold = FATAL
 log4j.appender.CA.follow = true
 
 

sql/core/src/test/resources/sql-tests/inputs/show-tables-legacy.sql

@@ -0,0 +1,2 @@
+--SET spark.sql.legacy.keepCommandOutputSchema=true
+--IMPORT show-tables.sql