[SPARK-53558][SQL] Show fully qualified table name including the catalog name in the exception message when the table is not found

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java

@@ -182,7 +182,7 @@ default Table loadTable(
    * @throws NoSuchTableException If the table doesn't exist or is a view
    */
   default Table loadTable(Identifier ident, String version) throws NoSuchTableException {
-    throw QueryCompilationErrors.noSuchTableError(ident);
+    throw QueryCompilationErrors.noSuchTableError(name(), ident);
   }
 
   /**
@@ -197,7 +197,7 @@ default Table loadTable(Identifier ident, String version) throws NoSuchTableExce
    * @throws NoSuchTableException If the table doesn't exist or is a view
    */
   default Table loadTable(Identifier ident, long timestamp) throws NoSuchTableException {
-    throw QueryCompilationErrors.noSuchTableError(ident);
+    throw QueryCompilationErrors.noSuchTableError(name(), ident);
   }
 
   /**

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

@@ -268,7 +268,8 @@ class SessionCatalog(
 
   private def requireTableExists(name: TableIdentifier): Unit = {
     if (!tableExists(name)) {
-      throw new NoSuchTableException(db = name.database.get, table = name.table)
+      throw new NoSuchTableException(
+        Seq(name.catalog.get, name.database.get, name.table))
     }
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -1511,8 +1511,8 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
         "continuousSources" -> continuousSources.mkString(", ")))
   }
 
-  def noSuchTableError(ident: Identifier): NoSuchTableException = {
-    new NoSuchTableException(ident.asMultipartIdentifier)
+  def noSuchTableError(catalogName: String, ident: Identifier): NoSuchTableException = {
+    new NoSuchTableException(catalogName +: ident.asMultipartIdentifier)
   }
 
   def noSuchTableError(nameParts: Seq[String]): Throwable = {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala

@@ -850,14 +850,16 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
           catalog.getTempViewOrPermanentTableMetadata(TableIdentifier("view1"))
         },
         condition = "TABLE_OR_VIEW_NOT_FOUND",
-        parameters = Map("relationName" -> "`default`.`view1`")
+        parameters = Map("relationName" ->
+          s"`${CatalogManager.SESSION_CATALOG_NAME}`.`default`.`view1`")
       )
       checkError(
         exception = intercept[NoSuchTableException] {
           catalog.getTempViewOrPermanentTableMetadata(TableIdentifier("view1", Some("default")))
         },
         condition = "TABLE_OR_VIEW_NOT_FOUND",
-        parameters = Map("relationName" -> "`default`.`view1`")
+        parameters = Map("relationName" ->
+          s"`${CatalogManager.SESSION_CATALOG_NAME}`.`default`.`view1`")
       )
 
       createTempView(catalog, "view1", tempTable, overrideIfExists = false)
@@ -871,7 +873,8 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
           catalog.getTempViewOrPermanentTableMetadata(TableIdentifier("view1", Some("default")))
         },
         condition = "TABLE_OR_VIEW_NOT_FOUND",
-        parameters = Map("relationName" -> "`default`.`view1`")
+        parameters = Map("relationName" ->
+          s"`${CatalogManager.SESSION_CATALOG_NAME}`.`default`.`view1`")
       )
     }
   }

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

@@ -126,7 +126,7 @@ class V2SessionCatalog(catalog: SessionCatalog)
       }
     } catch {
       case _: NoSuchNamespaceException =>
-        throw QueryCompilationErrors.noSuchTableError(ident)
+        throw QueryCompilationErrors.noSuchTableError(name(), ident)
     }
   }
 
@@ -284,7 +284,7 @@ class V2SessionCatalog(catalog: SessionCatalog)
       catalog.getTableMetadata(ident.asTableIdentifier)
     } catch {
       case _: NoSuchTableException =>
-        throw QueryCompilationErrors.noSuchTableError(ident)
+        throw QueryCompilationErrors.noSuchTableError(name(), ident)
     }
 
     val properties = CatalogV2Util.applyPropertiesChanges(catalogTable.properties, changes)
@@ -313,7 +313,7 @@ class V2SessionCatalog(catalog: SessionCatalog)
       }
     } catch {
       case _: NoSuchTableException =>
-        throw QueryCompilationErrors.noSuchTableError(ident)
+        throw QueryCompilationErrors.noSuchTableError(name(), ident)
     }
 
     null // Return null to save the `loadTable` call for ALTER TABLE.

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

@@ -142,7 +142,7 @@ class JDBCTableCatalog extends TableCatalog
   override def loadTable(ident: Identifier): Table = {
     JdbcUtils.withConnection(options) { conn =>
       if (!tableExists(ident, conn)) {
-        throw QueryCompilationErrors.noSuchTableError(ident)
+        throw QueryCompilationErrors.noSuchTableError(name(), ident)
       }
 
       val optionsWithTableName = new JDBCOptions(

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSessionCatalogSuite.scala

@@ -139,7 +139,7 @@ class InMemoryTableSessionCatalog extends TestV2SessionCatalogBase[InMemoryTable
 
         newTable
       case _ =>
-        throw QueryCompilationErrors.noSuchTableError(ident)
+        throw QueryCompilationErrors.noSuchTableError(name(), ident)
     }
   }
 }

sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala

@@ -221,7 +221,8 @@ class SupportsCatalogOptionsSuite extends QueryTest with SharedSparkSession with
     val e = intercept[NoSuchTableException] {
       spark.read.format(format).option("name", "non_existent_table").load()
     }
-    checkErrorTableNotFound(e, "`default`.`non_existent_table`")
+    val currentCatalogName = spark.catalog.currentCatalog().name()
+    checkErrorTableNotFound(e, s"`$currentCatalogName`.`default`.`non_existent_table`")
   }
 
   test("DataFrameWriter creates v2Relation with identifiers") {

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalogSuite.scala

@@ -31,7 +31,7 @@ import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.analysis.{NamespaceAlreadyExistsException, NoSuchNamespaceException, NoSuchTableException, TableAlreadyExistsException}
 import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
 import org.apache.spark.sql.catalyst.util.quoteIdentifier
-import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column, Identifier, NamespaceChange, SupportsNamespaces, TableCatalog, TableChange, TableSummary, V1Table}
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogV2Util, Column, Identifier, NamespaceChange, SupportsNamespaces, TableCatalog, TableChange, TableSummary, V1Table}
 import org.apache.spark.sql.connector.expressions.Transform
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.sql.types.{DoubleType, IntegerType, LongType, StringType, StructType, TimestampType}
@@ -48,8 +48,9 @@ abstract class V2SessionCatalogBaseSuite extends SharedSparkSession with BeforeA
   val testNs: Array[String] = Array("db")
   val defaultNs: Array[String] = Array("default")
   val testIdent: Identifier = Identifier.of(testNs, "test_table")
-  val testIdentQuoted: String = (testIdent.namespace :+ testIdent.name)
-    .map(part => quoteIdentifier(part)).mkString(".")
+  val catalogName: String = CatalogManager.SESSION_CATALOG_NAME
+  val testIdentQuoted: String = (catalogName +: testIdent.namespace :+ testIdent.name)
+    .map(quoteIdentifier).mkString(".")
 
   def newCatalog(): V2SessionCatalog = {
     val newCatalog = new V2SessionCatalog(spark.sessionState.catalog)