[SPARK-17717][SQL] Add Exist/find methods to Catalog [FOLLOW-UP]

## What changes were proposed in this pull request?
We added find and exists methods for Databases, Tables and Functions to the user facing Catalog in PR https://github.com/apache/spark/pull/15301. However, it was brought up that the semantics of the  `find` methods are more in line a `get` method (get an object or else fail). So we rename these in this PR.

## How was this patch tested?
Existing tests.

Author: Herman van Hovell <hvanhovell@databricks.com>

Closes #15308 from hvanhovell/SPARK-17717-2.

4 files changed, 71 insertions, 88 deletions

diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala
index 2ffe0ac9bc..7362041428 100644
--- a/project/MimaExcludes.scala
+++ b/project/MimaExcludes.scala
@@ -48,14 +48,12 @@ object MimaExcludes {
       // [SPARK-16240] ML persistence backward compatibility for LDA
       ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.ml.clustering.LDA$"),
       // [SPARK-17717] Add Find and Exists method to Catalog.
-      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.findDatabase"),
-      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.findTable"),
-      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.findFunction"),
-      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.findColumn"),
+      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.getDatabase"),
+      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.getTable"),
+      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.getFunction"),
       ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.databaseExists"),
       ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.tableExists"),
-      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.functionExists"),
-      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.columnExists")
+      ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.functionExists")
     )
   }
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala
index b439022d22..7f2762c7da 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala
@@ -102,50 +102,51 @@ abstract class Catalog {
   def listColumns(dbName: String, tableName: String): Dataset[Column]
 
   /**
-   * Find the database with the specified name. This throws an AnalysisException when the database
+   * Get the database with the specified name. This throws an AnalysisException when the database
    * cannot be found.
    *
    * @since 2.1.0
    */
   @throws[AnalysisException]("database does not exist")
-  def findDatabase(dbName: String): Database
+  def getDatabase(dbName: String): Database
 
   /**
-   * Find the table with the specified name. This table can be a temporary table or a table in the
-   * current database. This throws an AnalysisException when the table cannot be found.
+   * Get the table or view with the specified name. This table can be a temporary view or a
+   * table/view in the current database. This throws an AnalysisException when no Table
+   * can be found.
    *
    * @since 2.1.0
    */
   @throws[AnalysisException]("table does not exist")
-  def findTable(tableName: String): Table
+  def getTable(tableName: String): Table
 
   /**
-   * Find the table with the specified name in the specified database. This throws an
-   * AnalysisException when the table cannot be found.
+   * Get the table or view with the specified name in the specified database. This throws an
+   * AnalysisException when no Table can be found.
    *
    * @since 2.1.0
    */
   @throws[AnalysisException]("database or table does not exist")
-  def findTable(dbName: String, tableName: String): Table
+  def getTable(dbName: String, tableName: String): Table
 
   /**
-   * Find the function with the specified name. This function can be a temporary function or a
+   * Get the function with the specified name. This function can be a temporary function or a
    * function in the current database. This throws an AnalysisException when the function cannot
    * be found.
    *
    * @since 2.1.0
    */
   @throws[AnalysisException]("function does not exist")
-  def findFunction(functionName: String): Function
+  def getFunction(functionName: String): Function
 
   /**
-   * Find the function with the specified name. This throws an AnalysisException when the function
+   * Get the function with the specified name. This throws an AnalysisException when the function
    * cannot be found.
    *
    * @since 2.1.0
    */
   @throws[AnalysisException]("database or function does not exist")
-  def findFunction(dbName: String, functionName: String): Function
+  def getFunction(dbName: String, functionName: String): Function
 
   /**
    * Check if the database with the specified name exists.
@@ -155,15 +156,15 @@ abstract class Catalog {
   def databaseExists(dbName: String): Boolean
 
   /**
-   * Check if the table with the specified name exists. This can either be a temporary table or a
-   * table in the current database.
+   * Check if the table or view with the specified name exists. This can either be a temporary
+   * view or a table/view in the current database.
    *
    * @since 2.1.0
    */
   def tableExists(tableName: String): Boolean
 
   /**
-   * Check if the table with the specified name exists in the specified database.
+   * Check if the table or view with the specified name exists in the specified database.
    *
    * @since 2.1.0
    */
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
index a1087edd03..e412e1b4b3 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
@@ -68,13 +68,12 @@ class CatalogImpl(sparkSession: SparkSession) extends Catalog {
    * Returns a list of databases available across all sessions.
    */
   override def listDatabases(): Dataset[Database] = {
-    val databases = sessionCatalog.listDatabases().map { dbName =>
-      makeDatabase(sessionCatalog.getDatabaseMetadata(dbName))
-    }
+    val databases = sessionCatalog.listDatabases().map(makeDatabase)
     CatalogImpl.makeDataset(databases, sparkSession)
   }
 
-  private def makeDatabase(metadata: CatalogDatabase): Database = {
+  private def makeDatabase(dbName: String): Database = {
+    val metadata = sessionCatalog.getDatabaseMetadata(dbName)
     new Database(
       name = metadata.name,
       description = metadata.description,
@@ -96,20 +95,19 @@ class CatalogImpl(sparkSession: SparkSession) extends Catalog {
   @throws[AnalysisException]("database does not exist")
   override def listTables(dbName: String): Dataset[Table] = {
     requireDatabaseExists(dbName)
-    val tables = sessionCatalog.listTables(dbName).map { tableIdent =>
-      makeTable(tableIdent, tableIdent.database.isEmpty)
-    }
+    val tables = sessionCatalog.listTables(dbName).map(makeTable)
     CatalogImpl.makeDataset(tables, sparkSession)
   }
 
-  private def makeTable(tableIdent: TableIdentifier, isTemp: Boolean): Table = {
-    val metadata = if (isTemp) None else Some(sessionCatalog.getTableMetadata(tableIdent))
+  private def makeTable(tableIdent: TableIdentifier): Table = {
+    val metadata = sessionCatalog.getTempViewOrPermanentTableMetadata(tableIdent)
+    val database = metadata.identifier.database
     new Table(
-      name = tableIdent.identifier,
-      database = metadata.flatMap(_.identifier.database).orNull,
-      description = metadata.flatMap(_.comment).orNull,
-      tableType = metadata.map(_.tableType.name).getOrElse("TEMPORARY"),
-      isTemporary = isTemp)
+      name = tableIdent.table,
+      database = database.orNull,
+      description = metadata.comment.orNull,
+      tableType = if (database.isEmpty) "TEMPORARY" else metadata.tableType.name,
+      isTemporary = database.isEmpty)
   }
 
   /**
@@ -178,59 +176,45 @@ class CatalogImpl(sparkSession: SparkSession) extends Catalog {
   }
 
   /**
-   * Find the database with the specified name. This throws an [[AnalysisException]] when no
+   * Get the database with the specified name. This throws an [[AnalysisException]] when no
    * [[Database]] can be found.
    */
-  override def findDatabase(dbName: String): Database = {
-    if (sessionCatalog.databaseExists(dbName)) {
-      makeDatabase(sessionCatalog.getDatabaseMetadata(dbName))
-    } else {
-      throw new AnalysisException(s"The specified database $dbName does not exist.")
-    }
+  override def getDatabase(dbName: String): Database = {
+    makeDatabase(dbName)
   }
 
   /**
-   * Find the table with the specified name. This table can be a temporary table or a table in the
-   * current database. This throws an [[AnalysisException]] when no [[Table]] can be found.
+   * Get the table or view with the specified name. This table can be a temporary view or a
+   * table/view in the current database. This throws an [[AnalysisException]] when no [[Table]]
+   * can be found.
    */
-  override def findTable(tableName: String): Table = {
-    findTable(null, tableName)
+  override def getTable(tableName: String): Table = {
+    getTable(null, tableName)
   }
 
   /**
-   * Find the table with the specified name in the specified database. This throws an
+   * Get the table or view with the specified name in the specified database. This throws an
    * [[AnalysisException]] when no [[Table]] can be found.
    */
-  override def findTable(dbName: String, tableName: String): Table = {
-    val tableIdent = TableIdentifier(tableName, Option(dbName))
-    val isTemporary = sessionCatalog.isTemporaryTable(tableIdent)
-    if (isTemporary || sessionCatalog.tableExists(tableIdent)) {
-      makeTable(tableIdent, isTemporary)
-    } else {
-      throw new AnalysisException(s"The specified table $tableIdent does not exist.")
-    }
+  override def getTable(dbName: String, tableName: String): Table = {
+    makeTable(TableIdentifier(tableName, Option(dbName)))
   }
 
   /**
-   * Find the function with the specified name. This function can be a temporary function or a
+   * Get the function with the specified name. This function can be a temporary function or a
    * function in the current database. This throws an [[AnalysisException]] when no [[Function]]
    * can be found.
    */
-  override def findFunction(functionName: String): Function = {
-    findFunction(null, functionName)
+  override def getFunction(functionName: String): Function = {
+    getFunction(null, functionName)
   }
 
   /**
-   * Find the function with the specified name. This returns [[None]] when no [[Function]] can be
+   * Get the function with the specified name. This returns [[None]] when no [[Function]] can be
    * found.
    */
-  override def findFunction(dbName: String, functionName: String): Function = {
-    val functionIdent = FunctionIdentifier(functionName, Option(dbName))
-    if (sessionCatalog.functionExists(functionIdent)) {
-      makeFunction(functionIdent)
-    } else {
-      throw new AnalysisException(s"The specified function $functionIdent does not exist.")
-    }
+  override def getFunction(dbName: String, functionName: String): Function = {
+    makeFunction(FunctionIdentifier(functionName, Option(dbName)))
   }
 
   /**
@@ -241,15 +225,15 @@ class CatalogImpl(sparkSession: SparkSession) extends Catalog {
   }
 
   /**
-   * Check if the table with the specified name exists. This can either be a temporary table or a
-   * table in the current database.
+   * Check if the table or view with the specified name exists. This can either be a temporary
+   * view or a table/view in the current database.
    */
   override def tableExists(tableName: String): Boolean = {
     tableExists(null, tableName)
   }
 
   /**
-   * Check if the table with the specified name exists in the specified database.
+   * Check if the table or view with the specified name exists in the specified database.
    */
   override def tableExists(dbName: String, tableName: String): Boolean = {
     val tableIdent = TableIdentifier(tableName, Option(dbName))
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/internal/CatalogSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/internal/CatalogSuite.scala
index 783bf77f86..214bc736bd 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/internal/CatalogSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/internal/CatalogSuite.scala
@@ -340,61 +340,61 @@ class CatalogSuite
     }
   }
 
-  test("find database") {
-    intercept[AnalysisException](spark.catalog.findDatabase("db10"))
+  test("get database") {
+    intercept[AnalysisException](spark.catalog.getDatabase("db10"))
     withTempDatabase { db =>
-      assert(spark.catalog.findDatabase(db).name === db)
+      assert(spark.catalog.getDatabase(db).name === db)
     }
   }
 
-  test("find table") {
+  test("get table") {
     withTempDatabase { db =>
       withTable(s"tbl_x", s"$db.tbl_y") {
         // Try to find non existing tables.
-        intercept[AnalysisException](spark.catalog.findTable("tbl_x"))
-        intercept[AnalysisException](spark.catalog.findTable("tbl_y"))
-        intercept[AnalysisException](spark.catalog.findTable(db, "tbl_y"))
+        intercept[AnalysisException](spark.catalog.getTable("tbl_x"))
+        intercept[AnalysisException](spark.catalog.getTable("tbl_y"))
+        intercept[AnalysisException](spark.catalog.getTable(db, "tbl_y"))
 
         // Create objects.
         createTempTable("tbl_x")
         createTable("tbl_y", Some(db))
 
         // Find a temporary table
-        assert(spark.catalog.findTable("tbl_x").name === "tbl_x")
+        assert(spark.catalog.getTable("tbl_x").name === "tbl_x")
 
         // Find a qualified table
-        assert(spark.catalog.findTable(db, "tbl_y").name === "tbl_y")
+        assert(spark.catalog.getTable(db, "tbl_y").name === "tbl_y")
 
         // Find an unqualified table using the current database
-        intercept[AnalysisException](spark.catalog.findTable("tbl_y"))
+        intercept[AnalysisException](spark.catalog.getTable("tbl_y"))
         spark.catalog.setCurrentDatabase(db)
-        assert(spark.catalog.findTable("tbl_y").name === "tbl_y")
+        assert(spark.catalog.getTable("tbl_y").name === "tbl_y")
       }
     }
   }
 
-  test("find function") {
+  test("get function") {
     withTempDatabase { db =>
       withUserDefinedFunction("fn1" -> true, s"$db.fn2" -> false) {
         // Try to find non existing functions.
-        intercept[AnalysisException](spark.catalog.findFunction("fn1"))
-        intercept[AnalysisException](spark.catalog.findFunction("fn2"))
-        intercept[AnalysisException](spark.catalog.findFunction(db, "fn2"))
+        intercept[AnalysisException](spark.catalog.getFunction("fn1"))
+        intercept[AnalysisException](spark.catalog.getFunction("fn2"))
+        intercept[AnalysisException](spark.catalog.getFunction(db, "fn2"))
 
         // Create objects.
         createTempFunction("fn1")
         createFunction("fn2", Some(db))
 
         // Find a temporary function
-        assert(spark.catalog.findFunction("fn1").name === "fn1")
+        assert(spark.catalog.getFunction("fn1").name === "fn1")
 
         // Find a qualified function
-        assert(spark.catalog.findFunction(db, "fn2").name === "fn2")
+        assert(spark.catalog.getFunction(db, "fn2").name === "fn2")
 
         // Find an unqualified function using the current database
-        intercept[AnalysisException](spark.catalog.findFunction("fn2"))
+        intercept[AnalysisException](spark.catalog.getFunction("fn2"))
         spark.catalog.setCurrentDatabase(db)
-        assert(spark.catalog.findFunction("fn2").name === "fn2")
+        assert(spark.catalog.getFunction("fn2").name === "fn2")
       }
     }
   }