Improvements

Author: andreaschat-db

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

@@ -30,7 +30,7 @@ import org.apache.spark.sql.catalyst.analysis.TimeTravelSpec
 import org.apache.spark.sql.catalyst.expressions.Literal
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
-import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogV2Util, Identifier, SessionConfigSupport, StagedTable, StagingTableCatalog, SupportsCatalogOptions, SupportsRead, Table, TableProvider}
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogV2Util, DataSourceCatalogResolver, Identifier, SessionConfigSupport, StagedTable, StagingTableCatalog, SupportsCatalogOptions, SupportsRead, Table, TableProvider}
 import org.apache.spark.sql.connector.catalog.TableCapability.BATCH_READ
 import org.apache.spark.sql.errors.QueryExecutionErrors
 import org.apache.spark.sql.execution.SQLExecution
@@ -129,6 +129,30 @@ private[sql] object DataSourceV2Utils extends Logging {
     (catalogName, ident)
   }
 
+  /**
+   * Resolver bound to a session [[SQLConf]] that maps a multipart SQL identifier
+   * (e.g. `pathformat.\`/path/to/t\``) to a `(catalogName, identifier)` pair when the head
+   * names a registered [[SupportsCatalogOptions]] data source. Returns `None` for non-SCO
+   * sources or unknown format heads, letting the caller fall back to standard catalog
+   * resolution.
+   */
+  def supportsCatalogOptionsResolver(conf: SQLConf): DataSourceCatalogResolver =
+    (nameParts: Seq[String]) =>
+      try {
+        DataSource.lookupDataSourceV2(nameParts.head, conf).flatMap {
+          case sco: SupportsCatalogOptions =>
+            val optionsWithPath = getOptionsWithPaths(
+              CaseInsensitiveMap(Map.empty), nameParts.tail.mkString("."))
+            val dsOptions = buildDsOptions(sco, conf, optionsWithPath)
+            Some(extractCatalogAndIdentifier(sco, dsOptions))
+          case _ => None
+        }
+      } catch {
+        // The format name is not a registered data source. Fall through and let the caller
+        // treat it as a catalog/namespace name.
+        case _: ClassNotFoundException => None
+      }
+
   def loadV2Source(
       sparkSession: SparkSession,
       provider: TableProvider,

sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala

@@ -28,9 +28,8 @@ import org.apache.spark.sql.catalyst.optimizer.Optimizer
 import org.apache.spark.sql.catalyst.parser.ParserInterface
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
 import org.apache.spark.sql.classic.{SparkSession, Strategy, StreamingCheckpointManager, StreamingQueryManager, UDFRegistration}
-import org.apache.spark.sql.connector.catalog.{CatalogManager, DataSourceCatalogResolver, SupportsCatalogOptions}
+import org.apache.spark.sql.connector.catalog.{CatalogManager, DataSourceCatalogResolver}
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.execution.{ColumnarRule, CommandExecutionMode, QueryExecution, SparkOptimizer, SparkPlanner, SparkSqlParser}
 import org.apache.spark.sql.execution.adaptive.AdaptiveRulesHolder
@@ -162,21 +161,7 @@ abstract class BaseSessionStateBuilder(
   protected lazy val v2SessionCatalog = new V2SessionCatalog(catalog)
 
   protected lazy val dataSourceCatalogResolver: DataSourceCatalogResolver =
-    (nameParts: Seq[String]) =>
-      try {
-        DataSource.lookupDataSourceV2(nameParts.head, conf).flatMap {
-          case sco: SupportsCatalogOptions =>
-            val optionsWithPath = DataSourceV2Utils.getOptionsWithPaths(
-              CaseInsensitiveMap(Map.empty), nameParts.tail.mkString("."))
-            val dsOptions = DataSourceV2Utils.buildDsOptions(sco, conf, optionsWithPath)
-            Some(DataSourceV2Utils.extractCatalogAndIdentifier(sco, dsOptions))
-          case _ => None
-        }
-      } catch {
-        // The format name is not a registered data source. Fall through and let the caller
-        // treat it as a catalog/namespace name.
-        case _: ClassNotFoundException => None
-      }
+    DataSourceV2Utils.supportsCatalogOptionsResolver(conf)
 
   protected lazy val catalogManager = {
     val cm = new CatalogManager(v2SessionCatalog, catalog, dataSourceCatalogResolver)

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

@@ -121,14 +121,13 @@ class PathBasedTableSuite extends QueryTest with SharedSparkSession {
     }
     // The format head is not a registered data source and not a catalog. Resolution
     // falls through to a normal table-not-found error.
-    assert(e.getCondition == "TABLE_OR_VIEW_NOT_FOUND" ||
-      e.getMessage.toLowerCase.contains("not found"))
+    assert(e.getCondition == "TABLE_OR_VIEW_NOT_FOUND")
   }
 
   test("VERSION AS OF on path-based table") {
     val base = "pathformat.`/p`"
     // InMemoryTableCatalog implements time travel by appending the version string to the
-    // identifier name (see loadTable(ident, version) — it looks up name + version).
+    // identifier name (see loadTable(ident, version) - it looks up name + version).
     val versioned = "pathformat.`/pv1`"
     sql(s"CREATE TABLE $base (id INT)")
     sql(s"CREATE TABLE $versioned (id INT)")