[SPARK-3299][SQL] add to SQLContext API to show tables

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Catalog.scala

@@ -60,6 +60,9 @@ trait Catalog {
   protected def getDBTable(tableIdent: Seq[String]) : (Option[String], String) = {
     (tableIdent.lift(tableIdent.size - 2), tableIdent.last)
   }
+
+  def showTables(databaseName: Option[String]): Array[String]
+
 }
 
 class SimpleCatalog(val caseSensitive: Boolean) extends Catalog {
@@ -81,6 +84,10 @@ class SimpleCatalog(val caseSensitive: Boolean) extends Catalog {
     tables.clear()
   }
 
+  override def showTables(databaseName: Option[String]): Array[String] = {
+    tables.keys.toArray
+  }
+
   override def tableExists(tableIdentifier: Seq[String]): Boolean = {
     val tableIdent = processTableIdentifier(tableIdentifier)
     tables.get(getDbTableName(tableIdent)) match {
@@ -152,6 +159,14 @@ trait OverrideCatalog extends Catalog {
   override def unregisterAllTables(): Unit = {
     overrides.clear()
   }
+
+  override def showTables(databaseName: Option[String]): Array[String] = {
+    val dbName = if(!caseSensitive) databaseName.map(_.toLowerCase) else databaseName
+    dbName match {
+      case Some(dbName) => overrides.keys.filter(_._1.exists(_ == dbName)).map(t => t._2).toArray
+      case None => overrides.keys.map(t => t._2).toArray
+    }
+  }
 }
 
 /**
@@ -181,4 +196,8 @@ object EmptyCatalog extends Catalog {
   }
 
   override def unregisterAllTables(): Unit = {}
+
+  override def showTables(databaseName: Option[String]): Array[String] = {
+    throw new UnsupportedOperationException
+  }
 }

sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala

@@ -313,6 +313,13 @@ class SQLContext(@transient val sparkContext: SparkContext)
     catalog.unregisterTable(Seq(tableName))
   }
 
+  /**
+   * Returns the tables for this SQLContext
+   */
+  def showTables(): Array[String] = {
+    catalog.showTables(None)
+  }
+
   /**
    * Executes a SQL query using Spark, returning the result as a SchemaRDD.  The dialect that is
    * used for SQL parsing can be configured with 'spark.sql.dialect'.

sql/core/src/test/scala/org/apache/spark/sql/ShowTablesSuite.scala

@@ -0,0 +1,103 @@
+/*
+ * 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
+
+import org.apache.spark.sql.catalyst.analysis.{EmptyCatalog, OverrideCatalog, SimpleCatalog}
+import org.apache.spark.sql.test.TestSQLContext
+import org.scalatest.{BeforeAndAfter, FunSuite, Matchers}
+
+class ShowTablesSuite extends FunSuite with Matchers with BeforeAndAfter {
+
+  val simpleCatalog = new SimpleCatalog(true)
+
+  val expectedTablesOne = List("org.apache.spark.sql.ListTablesSuite.foo",
+    "org.apache.spark.sql.ListTablesSuite.bar",
+    "org.apache.spark.sql.ListTablesSuite.baz")
+
+  val expectedTablesTwo = List("org.apache.spark.sql.ListTablesSuite.Larry",
+    "org.apache.spark.sql.ListTablesSuite.Moe",
+    "org.apache.spark.sql.ListTablesSuite.Curly")
+
+  val overrideCatalog = new SimpleCatalog(true) with OverrideCatalog
+
+  val overrideCaseInsensitiveCatalog = new SimpleCatalog(false) with OverrideCatalog
+
+
+  before {
+    expectedTablesOne.map(t => simpleCatalog.registerTable(Seq(t), null))
+    expectedTablesOne.map(t => overrideCatalog.registerTable(Seq("A", t), null))
+    expectedTablesTwo.map(t => overrideCatalog.registerTable(Seq("B", t), null))
+    expectedTablesOne.map(t => overrideCaseInsensitiveCatalog.registerTable(Seq("x", t), null))
+    expectedTablesTwo.map(t => overrideCaseInsensitiveCatalog.registerTable(Seq("y", t), null))
+
+    expectedTablesOne.map(t =>
+      TestSQLContext.registerRDDAsTable(new SchemaRDD(TestSQLContext,null),t))
+  }
+
+  after {
+    simpleCatalog.unregisterAllTables()
+    overrideCatalog.unregisterAllTables()
+  }
+
+  test("SPARK-3299 showTables (foo, bar, baz) from SimpleCatalog") {
+    val returnedTables = simpleCatalog.showTables(None)
+    returnedTables should contain theSameElementsAs expectedTablesOne
+  }
+
+  test("SPARK-3299 showTables works correctly when no databases defined for tables"){
+      val newTables = List("tableX", "tableY", "tableZ")
+      newTables.map(t => overrideCatalog.registerTable(Seq(t), null))
+
+      val returnedTables = overrideCatalog.showTables(Some("A"))
+      returnedTables should contain theSameElementsAs expectedTablesOne
+  }
+
+  test("SPARK-3299 showTables correct with database names from OverrideCatalog") {
+    var returnedTables = overrideCatalog.showTables(Some("A"))
+    returnedTables should contain theSameElementsAs expectedTablesOne
+
+    returnedTables = overrideCatalog.showTables(Some("B"))
+    returnedTables should contain theSameElementsAs expectedTablesTwo
+  }
+
+  test("SPARK-3299 showTables with database names and case insensitivity from OverrideCatalog") {
+    var returnedTables = overrideCaseInsensitiveCatalog.showTables(Some("X"))
+    returnedTables should contain theSameElementsAs expectedTablesOne.map(_.toLowerCase)
+
+    returnedTables = overrideCaseInsensitiveCatalog.showTables(Some("Y"))
+    returnedTables should contain theSameElementsAs expectedTablesTwo.map(_.toLowerCase)
+  }
+
+  test("SPARK-3299 showTables contains all tables no database specified from OverrideCatalog") {
+    val returnedTables = overrideCatalog.showTables(None)
+    returnedTables should contain theSameElementsAs expectedTablesOne ++ expectedTablesTwo
+  }
+
+  test("SPARK-3329 showTables contains all tables for SQLContext") {
+    val returnedTables = TestSQLContext.showTables()
+    returnedTables should contain allOf("org.apache.spark.sql.ListTablesSuite.foo",
+      "org.apache.spark.sql.ListTablesSuite.bar",
+      "org.apache.spark.sql.ListTablesSuite.baz")
+  }
+
+  test("SPARK-3299 EmptyCatalog throws UnsupportedOperationException when showTables is called") {
+    intercept[UnsupportedOperationException] {
+      EmptyCatalog.showTables(None)
+    }
+  }
+}

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala

@@ -103,6 +103,19 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
     catalog.invalidateTable("default", tableName)
   }
 
+  /**
+   *
+   * @return All tables for given catalog
+   */
+  override def showTables(): Array[String] = catalog.showTables(None)
+
+  /**
+   *
+   * @param databaseName the name of database to retrieve tables for
+   * @return the list of tables for given database
+   */
+  def showTables(databaseName: String): Array[String] = catalog.showTables(Some(databaseName))
+
   /**
    * Analyzes the given table in the current database to generate statistics, which will be
    * used in query optimizations.

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala

@@ -456,6 +456,19 @@ private[hive] class HiveMetastoreCatalog(hive: HiveContext) extends Catalog with
   override def unregisterTable(tableIdentifier: Seq[String]): Unit = ???
 
   override def unregisterAllTables() = {}
+
+  // Necessary helper method so showTables can return Array[String] vs List
+  private def convertToArray(tables: java.util.List[String]): Array[String] = {
+    tables.toArray(new Array[String](tables.length))
+  }
+
+  override def showTables(databaseName: Option[String]): Array[String] = {
+    val dbName = if ( !caseSensitive ) databaseName.map(_.toLowerCase) else databaseName
+    dbName match {
+      case Some(dbName) => convertToArray(client.getAllTables(dbName))
+      case None => convertToArray(client.getAllTables)
+    }
+  }
 }
 
 /**