[CARMEL-6531] Fail to instantiate the custom v2 session catalog: DelegatingCatalog (#1221)

* [CARMEL-6531] Fail to instantiate the custom v2 session catalog: DelegatingCatalog

Cherry-pick from:
https://github.corp.ebay.com/hadoop/spark-longwing3/pull/194
https://github.corp.ebay.com/hadoop/spark-longwing3/pull/203

* Fix ut

Author: fenzhu

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.connector.catalog;
 
 import com.google.common.collect.Maps;
+import org.apache.spark.SparkException;
 import org.apache.spark.sql.catalyst.analysis.NamespaceAlreadyExistsException;
 import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
 import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
@@ -46,6 +47,8 @@
  *  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.connector.catalog.DelegatingCatalog
  *  --conf spark.sql.catalog.spark_catalog.delegating.provider.delta=DeltaCatalog
  *  --conf spark.sql.catalog.DeltaCatalog=org.apache.spark.sql.delta.catalog.DeltaCatalog
+ * and it is default to ignore catalog loading exception, which can be disabled by :
+ *  --conf spark.sql.catalog.spark_catalog.catch_catalog_load_exception=false
  */
 public class DelegatingCatalog implements CatalogExtension, StagingTableCatalog {
 
@@ -56,6 +59,8 @@ public class DelegatingCatalog implements CatalogExtension, StagingTableCatalog
   private Map<String, String> catalogs = Maps.newHashMap();
   private CatalogManager catalogManager = null;
   private SupportMetadataTable metadataTableCatalog = null;
+  private boolean catchCatalogLoadException;
+  public static final String CATCH_CATALOG_LOAD_EXCEPTION = "catch_catalog_load_exception";
   private static final String ICEBERG_PROVIDER = "iceberg";
   private static final String HIVE_PROVIDER = "hive";
 
@@ -103,6 +108,7 @@ public final void initialize(String name, CaseInsensitiveStringMap options) {
         catalogs.put(provider, catalogName);
       }
     }
+    catchCatalogLoadException = options.getBoolean(CATCH_CATALOG_LOAD_EXCEPTION, true);
   }
 
   @Override
@@ -119,7 +125,7 @@ public void setCatalogManager(CatalogManager manager) {
     catalogManager = manager;
     CatalogPlugin sessionCatalog = asTableCatalog();
     for (String catalogName : catalogs.values()) {
-      CatalogPlugin catalog = catalogManager.catalog(catalogName);
+      CatalogPlugin catalog = tryLoadCatalog(catalogName);
       if (catalog instanceof CatalogExtension) {
         ((CatalogExtension) catalog).setDelegateCatalog(sessionCatalog);
       }
@@ -138,14 +144,30 @@ private CatalogExtension getV2Catalog(String provider) {
     if (catalogName == null) {
       return null;
     }
-    CatalogPlugin catalog = catalogManager.catalog(catalogName);
-    if (catalog instanceof CatalogExtension) {
+    CatalogPlugin catalog = tryLoadCatalog(catalogName);
+    if (catalog == null) {
+      return null;
+    } else if (catalog instanceof CatalogExtension) {
       return (CatalogExtension) catalog;
     } else {
       throw new RuntimeException(catalog.name() + " is not a CatalogExtension");
     }
   }
 
+  private CatalogPlugin tryLoadCatalog(String name) {
+    CatalogPlugin catalog = null;
+    try {
+      catalog = catalogManager.catalog(name);
+    } catch (Exception e) {
+      if ((e instanceof SparkException) && catchCatalogLoadException) {
+        logger.warn("Exception caught when init loading catalog " + name);
+      } else {
+        throw e;
+      }
+    }
+    return catalog;
+  }
+
   private StagingTableCatalog getStagingTableCatalog(String provider) {
     CatalogExtension v2Catalog = getV2Catalog(provider);
     if (v2Catalog == null) {

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

@@ -202,25 +202,40 @@ private [connector] trait DelegatingCatalogTest[T <: Table]
 
   protected val catalogClassName: String = classOf[DelegatingCatalog].getName
 
-  private val FAKEV2_PROVIDER_KEY: String = V2_SESSION_CATALOG_IMPLEMENTATION.key +
+  private val FAKE_V2_PROVIDER_KEY: String = V2_SESSION_CATALOG_IMPLEMENTATION.key +
     ".delegating.provider." + v2Format
-  private val Delegated_Catalog_Name: String = classOf[StagingInMemoryTableDelegatedCatalog]
+  private val DELEGATED_CATALOG_NAME: String = classOf[StagingInMemoryTableDelegatedCatalog]
     .getSimpleName
-  private val Delegated_Catalog: String = s"spark.sql.catalog.$Delegated_Catalog_Name"
-  private val Delegated_Catalog_Impl: String = classOf[StagingInMemoryTableDelegatedCatalog].getName
+  private val DELEGATED_CATALOG: String = s"spark.sql.catalog.$DELEGATED_CATALOG_NAME"
+  private val DELEGATED_CATALOG_IMPL: String = classOf[StagingInMemoryTableDelegatedCatalog].getName
+
+  private val NONEXISTENT_FAKE_V2_PROVIDER_KEY: String = V2_SESSION_CATALOG_IMPLEMENTATION.key +
+    ".delegating.provider." + v2Format + "_not_exist"
+  private val NONEXISTENT_DELEGATED_CATALOG_NAME: String
+  = classOf[StagingInMemoryTableDelegatedCatalog].getSimpleName + "NotExist"
+  private val NONEXISTENT_DELEGATED_CATALOG: String
+  = s"spark.sql.catalog.$NONEXISTENT_DELEGATED_CATALOG_NAME"
+  private val NONEXISTENT_DELEGATED_CATALOG_IMPL: String
+  = classOf[StagingInMemoryTableDelegatedCatalog].getName + "NotExist"
 
   before {
     spark.conf.set(V2_SESSION_CATALOG_IMPLEMENTATION.key, catalogClassName)
-    spark.conf.set(FAKEV2_PROVIDER_KEY, Delegated_Catalog_Name)
-    spark.conf.set(Delegated_Catalog, Delegated_Catalog_Impl)
+    spark.conf.set(FAKE_V2_PROVIDER_KEY, DELEGATED_CATALOG_NAME)
+    spark.conf.set(DELEGATED_CATALOG, DELEGATED_CATALOG_IMPL)
+    // add catalog configuration that fails to load
+    spark.conf.set(NONEXISTENT_FAKE_V2_PROVIDER_KEY, NONEXISTENT_DELEGATED_CATALOG_NAME)
+    spark.conf.set(NONEXISTENT_DELEGATED_CATALOG, NONEXISTENT_DELEGATED_CATALOG_IMPL)
   }
 
   override def afterEach(): Unit = {
     super.afterEach()
-    catalog(Delegated_Catalog_Name).asInstanceOf[StagingInMemoryTableDelegatedCatalog].clearTables()
+    catalog(DELEGATED_CATALOG_NAME).asInstanceOf[StagingInMemoryTableDelegatedCatalog].clearTables()
     spark.conf.unset(V2_SESSION_CATALOG_IMPLEMENTATION.key)
-    spark.conf.unset(FAKEV2_PROVIDER_KEY)
-    spark.conf.unset(Delegated_Catalog)
+    spark.conf.unset(FAKE_V2_PROVIDER_KEY)
+    spark.conf.unset(DELEGATED_CATALOG)
+    // unset catalog configuration that fails to load
+    spark.conf.unset(NONEXISTENT_FAKE_V2_PROVIDER_KEY)
+    spark.conf.unset(NONEXISTENT_DELEGATED_CATALOG)
   }
 
   protected def verifyTable(tableName: String, expected: DataFrame): Unit
@@ -311,3 +326,56 @@ private [connector] trait DelegatingCatalogTest[T <: Table]
     verifyTable(t1, Seq(("c", "d")).toDF("id", "data"))
   }
 }
+
+class DataSourceV2DataFrameDelegatingCatalogFailLoadingSuite
+  extends QueryTest
+    with DelegatingCatalogFailLoadingTest[InMemoryTable] {
+
+  test("showTables should fail with SparkException that can't load delegated catalog") {
+    val expectedErrorMsg
+    = "Cannot find catalog plugin class for catalog '" + NONEXISTENT_DELEGATED_CATALOG_NAME + "':"
+    val msg = intercept[org.apache.spark.SparkException] {
+        sql("show tables")
+      Catalogs.load(NONEXISTENT_DELEGATED_CATALOG_NAME, spark.sessionState.conf)
+    }.getMessage
+    assert(msg.contains(expectedErrorMsg))
+  }
+}
+
+private [connector] trait DelegatingCatalogFailLoadingTest[T <: Table]
+  extends QueryTest
+    with SharedSparkSession
+    with BeforeAndAfter {
+
+  protected val v2Format: String = classOf[FakeV2Provider].getName
+
+  protected val catalogClassName: String = classOf[DelegatingCatalog].getName
+
+  private val NONEXISTENT_FAKE_V2_PROVIDER_KEY: String = V2_SESSION_CATALOG_IMPLEMENTATION.key +
+    ".delegating.provider." + v2Format + "_not_exist"
+  protected val NONEXISTENT_DELEGATED_CATALOG_NAME: String
+  = classOf[StagingInMemoryTableDelegatedCatalog].getSimpleName + "NotExist"
+  private val NONEXISTENT_DELEGATED_CATALOG: String
+  = s"spark.sql.catalog.$NONEXISTENT_DELEGATED_CATALOG_NAME"
+  private val NONEXISTENT_DELEGATED_CATALOG_IMPL: String
+  = classOf[StagingInMemoryTableDelegatedCatalog].getName + "NotExist"
+
+  private val CATCH_CATALOG_LOAD_EXCEPTION_KEY: String = V2_SESSION_CATALOG_IMPLEMENTATION.key +
+    '.' + DelegatingCatalog.CATCH_CATALOG_LOAD_EXCEPTION
+
+  before {
+    spark.conf.set(V2_SESSION_CATALOG_IMPLEMENTATION.key, catalogClassName)
+    spark.conf.set(CATCH_CATALOG_LOAD_EXCEPTION_KEY, false)
+    // add catalog configuration that fails to load
+    spark.conf.set(NONEXISTENT_FAKE_V2_PROVIDER_KEY, NONEXISTENT_DELEGATED_CATALOG_NAME)
+    spark.conf.set(NONEXISTENT_DELEGATED_CATALOG, NONEXISTENT_DELEGATED_CATALOG_IMPL)
+  }
+
+  override def afterEach(): Unit = {
+    super.afterEach()
+    spark.conf.unset(V2_SESSION_CATALOG_IMPLEMENTATION.key)
+    // unset catalog configuration that fails to load
+    spark.conf.unset(NONEXISTENT_FAKE_V2_PROVIDER_KEY)
+    spark.conf.unset(NONEXISTENT_DELEGATED_CATALOG)
+  }
+}