andreaschat-db commented on code in PR #56039:
URL: https://github.com/apache/spark/pull/56039#discussion_r3324969166
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala:
##########
@@ -99,6 +99,60 @@ private[sql] object DataSourceV2Utils extends Logging {
}
}
+ /**
+ * Builds the [[CaseInsensitiveStringMap]] passed to a v2 [[TableProvider]]:
session configs
+ * extracted from the provider, merged with the caller-supplied
options-with-path map.
+ */
+ def buildDsOptions(
+ provider: TableProvider,
+ conf: SQLConf,
+ optionsWithPath: CaseInsensitiveMap[String]): CaseInsensitiveStringMap =
{
+ val sessionOptions = extractSessionConfigs(provider, conf)
+ val finalOptions = sessionOptions.filter { case (k, _) =>
!optionsWithPath.contains(k) } ++
+ optionsWithPath.originalMap
+ new CaseInsensitiveStringMap(finalOptions.asJava)
+ }
+
+ /**
+ * Extracts the catalog name and connector-canonical identifier from a
+ * [[SupportsCatalogOptions]] provider. Shared by all SCO entry points
(DataFrame reader, SQL
+ * multipart-name resolution) so they observe identical null-handling
semantics:
+ * `extractCatalog` returning null falls back to the session catalog,
matching
+ * [[CatalogV2Util.getTableProviderCatalog]].
+ */
+ def extractCatalogAndIdentifier(
+ provider: SupportsCatalogOptions,
+ dsOptions: CaseInsensitiveStringMap): (String, Identifier) = {
+ val ident = provider.extractIdentifier(dsOptions)
+ val catalogName = Option(provider.extractCatalog(dsOptions))
+ .getOrElse(CatalogManager.SESSION_CATALOG_NAME)
+ (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]) =>
Review Comment:
> Should we attempt to do this only if we have 2 name parts?
We are talking about SQL on file scenarios, right?
Yes these are SQL on file scenarios. The 2 name part sounds like a
limitation of v1? IIUC in v2 dataframes we do not have such restriction. I
guess the connector can decided how to handle it in
`extractCatalog`/`extractIdentifier`?
> Do we care about spark.sql.runSQLonFiles? See ResolveSQLOnFile
I added the sentinel.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/LookupCatalog.scala:
##########
@@ -135,7 +135,18 @@ private[sql] trait LookupCatalog extends Logging {
Some((catalog, ident))
} catch {
case _: CatalogNotFoundException =>
- Some((currentCatalog, nameParts.asIdentifier))
+ // No catalog matched. As a fallback, try path-based data sources:
+ // formats implementing SupportsCatalogOptions (e.g.
`pathformat.`/path/to/t``)
+ // route to the catalog the connector designates. If no SCO format
claims the
+ // identifier head, fall through to currentCatalog and let later
analysis raise
+ // table-not-found. This matches the v1 file-format precedence
(catalog first,
+ // path-based as fallback).
+
Option(catalogManager.catalogAndIdentForDataSource(nameParts)).flatten match {
Review Comment:
Yes thank you. I fixed the resolution order.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
