diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala index b816016a3ec84..081792bd8ec3a 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala @@ -1885,9 +1885,9 @@ case class RepairTable( * * Extends [[AnalysisOnlyCommand]] so [[Analyzer.HandleSpecialCommand]] captures * `referredTempFunctions` from [[AnalysisContext]]; this list is needed by - * [[CheckViewReferences]] and by the v2 execs when the target is a non-session catalog. - * Session-catalog targets are still rewritten to [[AlterViewAsCommand]] by - * `ResolveSessionCatalog` and the captured value is dropped there (the v1 command re-captures). + * [[CheckViewReferences]] and by the v2 execs when the target is a ViewCatalog. Session-catalog + * targets without ViewCatalog are rewritten to [[AlterViewAsCommand]] by `ResolveSessionCatalog` + * and the captured value is dropped there (the v1 command re-captures). */ case class AlterViewAs( child: LogicalPlan, diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/metricview/logical/metricViewNodes.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/metricview/logical/metricViewNodes.scala index 76163794db33d..763bc1e5c244f 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/metricview/logical/metricViewNodes.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/metricview/logical/metricViewNodes.scala @@ -50,10 +50,11 @@ case class MeasureInputColumn( /** * Logical plan for `CREATE VIEW ... WITH METRICS`. This is the v1/v2-agnostic representation * the parser returns; downstream analysis decides which runnable form it becomes: - * - For the session catalog: [[org.apache.spark.sql.execution.command.CreateMetricViewCommand]] - * via an analyzer rule that fires once the identifier is resolved. - * - For non-session v2 [[org.apache.spark.sql.connector.catalog.ViewCatalog]]s: a - * `CreateV2MetricViewExec` produced by `DataSourceV2Strategy`. + * - For session catalogs without + * [[org.apache.spark.sql.connector.catalog.ViewCatalog]]: + * [[org.apache.spark.sql.execution.command.CreateMetricViewCommand]]. + * - For [[org.apache.spark.sql.connector.catalog.ViewCatalog]]s, including custom session + * catalogs: a `CreateV2MetricViewExec` produced by `DataSourceV2Strategy`. * * Splitting this from the runnable command lets the parser return a single logical shape * regardless of target catalog (instead of pre-committing to a runnable command at parse diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala index e6fc6d8d862ce..9908b7534f5cd 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala @@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.rules.Rule import org.apache.spark.sql.catalyst.util.{quoteIfNeeded, toPrettySQL, CharVarcharUtils, ResolveDefaultColumns => DefaultCols} import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._ -import org.apache.spark.sql.connector.catalog.{CatalogExtension, CatalogManager, CatalogPlugin, CatalogV2Util, LookupCatalog, SupportsNamespaces, V1Table, ViewCatalog} +import org.apache.spark.sql.connector.catalog.{CatalogExtension, CatalogManager, CatalogPlugin, CatalogV2Util, Identifier, LookupCatalog, RelationCatalog, SupportsNamespaces, V1Table, V1View, ViewCatalog} import org.apache.spark.sql.connector.expressions.Transform import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors} import org.apache.spark.sql.execution.command._ @@ -233,9 +233,9 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager) output) => DescribeTableCommand(resolvedChild, ident, spec, isExtended, output) - // `DESCRIBE TABLE PARTITION (...)` against a non-session v2 view: the v1 rewrite - // above is gated on `ResolvedV1TableOrViewIdentifier` (session-only), so non-session v2 - // views fall through. Reject early with the same `FORBIDDEN_OPERATION` v1 raises at + // `DESCRIBE TABLE PARTITION (...)` against a v2 view: the v1 rewrite above is gated + // on `ResolvedV1TableOrViewIdentifier`, so ViewCatalog-backed views fall through. Reject + // early with the same `FORBIDDEN_OPERATION` v1 raises at // runtime in `DescribeTableCommand.describeDetailedPartitionInfo`. Without this rewrite, // CheckAnalysis surfaces a generic "Found the unresolved operator" INTERNAL_ERROR // because `UnresolvedPartitionSpec` is never resolved on the v2 view path. @@ -249,7 +249,7 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager) // typically resolves the column to an `Attribute` here. We also accept the legacy // `UnresolvedAttribute` form (e.g. the parser referenced a non-existent column whose // resolution was skipped) so the rewrite stays robust across analyzer ordering changes. - // The unwrap logic is shared with the non-session v2 view path in `DataSourceV2Strategy`. + // The unwrap logic is shared with the v2 view path in `DataSourceV2Strategy`. val nameParts = DescribeColumn.extractColumnNameParts(column) DescribeColumnCommand(ident, nameParts, isExtended, output) @@ -368,7 +368,7 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager) // ViewCatalog catalogs fall through to `DataSourceV2Strategy`, which routes DROP VIEW to // `ViewCatalog.dropView` (this also covers METRIC_VIEW since metric views are persisted - // through the same ViewCatalog interface). Other non-session catalogs get + // through the same ViewCatalog interface). Other catalogs get // `MISSING_CATALOG_ABILITY.VIEWS`, matching the error raised from `CheckViewReferences` for // CREATE/ALTER VIEW and from the analyzer gate on UnresolvedView. case DropView(r @ ResolvedIdentifier(catalog, ident), ifExists) @@ -595,11 +595,11 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager) viewType = PersistedView, viewSchemaMode = viewSchemaMode) - // CREATE VIEW ... WITH METRICS on the session catalog -> V1 runnable command. Non-session - // v2 catalogs leave [[CreateMetricView]] in place for `DataSourceV2Strategy` to dispatch - // to `CreateV2MetricViewExec`. + // Session catalogs without ViewCatalog use the V1 runnable command. ViewCatalog + // implementations, including custom session catalogs, leave [[CreateMetricView]] in place + // for `DataSourceV2Strategy` to dispatch to `CreateV2MetricViewExec`. case cm @ CreateMetricView(ResolvedIdentifier(catalog, _), _, _, _, _, _, _) - if isSessionCatalog(catalog) => + if useV1ViewCommands(catalog) => CreateMetricViewCommand( cm.child, cm.userSpecifiedColumns, @@ -610,8 +610,8 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager) cm.replace) // ViewCatalog catalogs are handled by the v2 strategy (enumerates via listViews); we skip - // the match here so the plan flows through unchanged. Only non-session, non-ViewCatalog - // catalogs hit the MISSING_CATALOG_ABILITY.VIEWS rejection. + // the match here so the plan flows through unchanged. Session catalogs without ViewCatalog + // use the V1 command; other catalogs without ViewCatalog are rejected. case ShowViews(ns: ResolvedNamespace, pattern, output) if !ns.catalog.isInstanceOf[ViewCatalog] => ns match { @@ -844,15 +844,29 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager) ) } + private def useV1ViewCommands(catalog: CatalogPlugin): Boolean = { + isSessionCatalog(catalog) && !catalog.isInstanceOf[ViewCatalog] + } + + private def isDelegatedV1View(catalog: CatalogPlugin, ident: Identifier): Boolean = { + catalog match { + case relationCatalog: RelationCatalog if isSessionCatalog(catalog) => + try { + relationCatalog.loadRelation(ident).isInstanceOf[V1View] + } catch { + case _: NoSuchNamespaceException | _: NoSuchTableException => false + } + case _ => false + } + } + object ResolvedViewIdentifier { - // Only matches session-catalog persistent views. Non-session-catalog persistent views - // (produced for `DelegatingTable`) fall through and are picked up by dedicated v2 strategy - // cases in `DataSourceV2Strategy` -- AlterViewAs, SET/UNSET TBLPROPERTIES, ALTER VIEW ... - // WITH SCHEMA, RENAME TO, SHOW CREATE TABLE, SHOW TBLPROPERTIES, SHOW COLUMNS, DESCRIBE - // [COLUMN] all dispatch to v2 view execs that consume `ResolvedPersistentView.info` - // directly. + // Only matches persistent views loaded from the V1 session catalog. A custom session + // ViewCatalog can still delegate table lookup to the built-in catalog, so route based on + // the resolved V1View payload rather than the catalog's capabilities. Native ViewCatalog + // views fall through to dedicated v2 strategy cases in `DataSourceV2Strategy`. def unapply(resolved: LogicalPlan): Option[TableIdentifier] = resolved match { - case ResolvedPersistentView(catalog, ident, _) if isSessionCatalog(catalog) => + case ResolvedPersistentView(catalog, ident, _: V1View) if isSessionCatalog(catalog) => Some(ident.asTableIdentifier.copy(catalog = Some(catalog.name))) case ResolvedTempView(ident, _) => @@ -900,11 +914,15 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager) } private object CreateViewInSessionCatalog - extends ResolvedIdentifierInSessionCatalog("CREATE", "VIEW") + extends ResolvedIdentifierInSessionCatalog( + "CREATE", "VIEW", (catalog, _) => useV1ViewCommands(catalog)) private object DropViewInSessionCatalog - extends ResolvedIdentifierInSessionCatalog("DROP", "VIEW") + extends ResolvedIdentifierInSessionCatalog( + "DROP", "VIEW", + (catalog, ident) => useV1ViewCommands(catalog) || isDelegatedV1View(catalog, ident)) private object CreateFunctionInSessionCatalog - extends ResolvedIdentifierInSessionCatalog("CREATE", "FUNCTION") + extends ResolvedIdentifierInSessionCatalog( + "CREATE", "FUNCTION", (catalog, _) => isSessionCatalog(catalog)) /** * Extractor for resolved identifiers in the session catalog. @@ -912,10 +930,14 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager) * * @param statement the SQL statement (e.g. "CREATE", "DROP") for error messages * @param objectType the object type (e.g. "FUNCTION", "VIEW") for error messages + * @param identifierPredicate whether the catalog and identifier should use the V1 command */ - class ResolvedIdentifierInSessionCatalog(statement: String, objectType: String) { + class ResolvedIdentifierInSessionCatalog( + statement: String, + objectType: String, + identifierPredicate: (CatalogPlugin, Identifier) => Boolean) { def unapply(resolved: LogicalPlan): Option[TableIdentifier] = resolved match { - case ResolvedIdentifier(catalog, ident) if isSessionCatalog(catalog) => + case ResolvedIdentifier(catalog, ident) if identifierPredicate(catalog, ident) => if (ident.namespace().length != 1) { if (ident.namespace().length >= 1 && ident.namespace().last.equalsIgnoreCase(CatalogManager.BUILTIN_NAMESPACE)) { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/DescribeRelationJsonCommand.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/DescribeRelationJsonCommand.scala index 8746fdbc92499..d86ed3f260e92 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/DescribeRelationJsonCommand.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/DescribeRelationJsonCommand.scala @@ -72,9 +72,9 @@ case class DescribeRelationJsonCommand( throw QueryCompilationErrors.descPartitionNotAllowedOnView(v.identifier.name()) } // Resolve `v.info` to a `CatalogTable` so the JSON renderer below can read v1-shaped - // fields uniformly. Session-catalog views carry the original `CatalogTable` inside - // `V1View`; non-session v2 views carry a plain `View` and are projected to a - // `CatalogTable` via `V1Table.toCatalogTable`, the same conversion the + // fields uniformly. Views handled by the v1 session catalog carry the original + // `CatalogTable` inside `V1View`; ViewCatalog-backed views carry a plain `View` and are + // projected to a `CatalogTable` via `V1Table.toCatalogTable`, the same conversion the // `CreateTableLike` strategy case in `DataSourceV2Strategy` uses. val metadata = v.info match { case v1Info: V1View => v1Info.v1Table diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala index df050742ec666..9d29399a90b1d 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala @@ -591,7 +591,7 @@ object ResolvedChildHelper { child match { case ResolvedTempView(_, metadata) => metadata // v1 inspection commands always see a v1 (`V1View`) view here -- the v2 strategy - // handles non-session views before this method is reached. + // handles ViewCatalog-backed views before this method is reached. case ResolvedPersistentView(_, _, info: V1View) => info.v1Table case ResolvedTable(_, _, t: V1Table, _) => t.v1Table case _ if (catalog.isTempView(table)) => diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala index 8805bfe75298f..f5fd8cf45f968 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala @@ -104,11 +104,11 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat location, session.sharedState.hadoopConf) } - // Strategy cases that target v2 views read `ResolvedPersistentView.info` directly. For - // session-catalog (v1) views the payload is a `V1View` wrapping the original - // `CatalogTable`; v2 catalogs supply a regular `View` from the catalog. - // `ResolveSessionCatalog` rewrites session-catalog views to v1 commands before this strategy - // fires, so v2 cases that don't expect a `V1View` won't see one. + // Strategy cases that target v2 views read `ResolvedPersistentView.info` directly. For views + // handled by the v1 session catalog, the payload is a `V1View` wrapping the original + // `CatalogTable`; ViewCatalog implementations supply a regular `View` from the catalog. + // `ResolveSessionCatalog` rewrites only the former to v1 commands before this strategy fires, + // so v2 cases that don't expect a `V1View` won't see one. private def qualifyLocInTableSpec(tableSpec: TableSpec): TableSpec = { val newLoc = tableSpec.location.map { loc => @@ -283,9 +283,9 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat // CREATE TABLE ... LIKE ... for a v2 catalog target. // Source is an already-resolved Table object; no extra catalog round-trip is needed. // Views are wrapped in V1Table so the exec can extract schema and provider uniformly -- - // session-catalog (v1) views unwrap to their original `CatalogTable`; non-session v2 - // views go through `V1Table.toCatalogTable` to synthesize an equivalent `CatalogTable` - // from the resolved `View`. + // views handled by the v1 session catalog unwrap to their original `CatalogTable`; + // ViewCatalog-backed views go through `V1Table.toCatalogTable` to synthesize an equivalent + // `CatalogTable` from the resolved `View`. case CreateTableLike( ResolvedIdentifier(catalog, ident), source, locationStr, provider, serdeInfo, properties, ifNotExists) => @@ -339,15 +339,14 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat CreateV2ViewExec(catalog.asInstanceOf[ViewCatalog], ident, userSpecifiedColumns, comment, collation, properties, sqlText, child, allowExisting, replace, viewSchemaMode) :: Nil - // CREATE VIEW ... WITH METRICS on a non-session v2 catalog. Routes the metric-view path - // through `CreateV2MetricViewExec`, which extends `V2ViewPreparation` to share the - // `IF NOT EXISTS` short-circuit, `OR REPLACE`, and cross-type-collision decoding with - // `CreateV2ViewExec`. Session-catalog dispatch happens earlier in `ResolveSessionCatalog`, - // which rewrites `CreateMetricView` (the parser's v1/v2-agnostic logical plan) to - // `CreateMetricViewCommand` for v1 execution. + // CREATE VIEW ... WITH METRICS on a ViewCatalog. Routes the metric-view path through + // `CreateV2MetricViewExec`, which extends `V2ViewPreparation` to share the `IF NOT EXISTS` + // short-circuit, `OR REPLACE`, and cross-type-collision decoding with `CreateV2ViewExec`. + // Session catalogs without ViewCatalog are rewritten to `CreateMetricViewCommand` earlier + // in `ResolveSessionCatalog`. case CreateMetricView( ResolvedIdentifier(catalog, ident), userSpecifiedColumns, comment, properties, - originalText, allowExisting, replace) if !CatalogV2Util.isSessionCatalog(catalog) => + originalText, allowExisting, replace) => val viewCatalog = catalog match { case vc: ViewCatalog => vc case _ => throw QueryCompilationErrors.missingCatalogViewsAbilityError(catalog) @@ -379,9 +378,8 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat AlterV2ViewExec(catalog.asInstanceOf[ViewCatalog], ident, rpv.info, originalText, query) :: Nil - // View DDL / inspection on a non-session v2 catalog that the v1 rewrite in - // `ResolveSessionCatalog` can't handle (its `ResolvedViewIdentifier` matcher is gated on - // `isSessionCatalog`). Routed to dedicated v2 execs that read the typed `View` + // View DDL / inspection on a ViewCatalog that the v1 rewrite in `ResolveSessionCatalog` + // leaves unchanged. Routed to dedicated v2 execs that read the typed `View` // resolved at analysis time directly from `ResolvedPersistentView.info` -- no re-loading // at exec time. case SetViewProperties(rpv @ ResolvedPersistentView(catalog, ident, _), props) => @@ -455,9 +453,8 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat output, rpv.info, DescribeColumn.extractColumnNameParts(column), isExtended) :: Nil // Plans that resolve through `UnresolvedTableOrView` reach here with a - // `ResolvedPersistentView` child for non-session v2 views (the v1 rewrite in - // `ResolveSessionCatalog` no longer matches them because `ResolvedViewIdentifier` is gated - // on `isSessionCatalog`). Pin each with `UNSUPPORTED_FEATURE.TABLE_OPERATION` so users get + // `ResolvedPersistentView` child for ViewCatalog-backed views. Pin each with + // `UNSUPPORTED_FEATURE.TABLE_OPERATION` so users get // a clean `AnalysisException` instead of a generic "No plan for ..." assertion from the // planner. Tracked for follow-up real handlers in SPARK-52729. case RefreshTable(ResolvedPersistentView(catalog, ident, _)) => @@ -476,8 +473,8 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat // `UnresolvedTable` (not `UnresolvedTableOrView`), so `CheckAnalysis` surfaces // `EXPECT_TABLE_NOT_VIEW.NO_ALTERNATIVE` before planning. No strategy case needed. - // DROP VIEW on a non-session ViewCatalog. The v1 rewrite in `ResolveSessionCatalog` skips - // ViewCatalog catalogs, so they fall through here. `DropViewExec` calls + // DROP VIEW on a ViewCatalog. The v1 rewrite in `ResolveSessionCatalog` skips ViewCatalog + // catalogs, so they fall through here. `DropViewExec` calls // `ViewCatalog.dropView` and surfaces `EXPECT_VIEW_NOT_TABLE` if the identifier resolves to // a table in a mixed catalog. case DropView(r @ ResolvedIdentifier(catalog: ViewCatalog, ident), ifExists) => @@ -685,11 +682,14 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat // SHOW VIEWS on a v2 ViewCatalog. `ResolveSessionCatalog` rewrites the SHOW VIEWS plan to // v1 `ShowViewsCommand` only when the catalog is NOT a `ViewCatalog`; non-`ViewCatalog` // catalogs (session or not) are rejected with `MISSING_CATALOG_ABILITY.VIEWS` there. So - // this case sees `ViewCatalog` catalogs (typically non-session, since the default - // `V2SessionCatalog` is not a `ViewCatalog`; a session-catalog override that mixes in - // `ViewCatalog` would also reach here). + // this case sees `ViewCatalog` catalogs, including custom session-catalog implementations. case ShowViews(ResolvedNamespace(catalog: ViewCatalog, ns, _), pattern, output) => - ShowViewsExec(output, catalog, ns, pattern) :: Nil + val v1SessionCatalog = if (CatalogV2Util.isSessionCatalog(catalog)) { + Some(session.sessionState.catalog) + } else { + None + } + ShowViewsExec(output, catalog, ns, pattern, v1SessionCatalog) :: Nil case ShowTablesExtended( ResolvedNamespace(catalog, ns, _), diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowViewsExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowViewsExec.scala index 00927f05842ad..51e7414254e93 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowViewsExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowViewsExec.scala @@ -17,35 +17,103 @@ package org.apache.spark.sql.execution.datasources.v2 -import scala.collection.mutable.ArrayBuffer +import scala.collection.mutable import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.analysis.{NoSuchNamespaceException, NoSuchTableException} +import org.apache.spark.sql.catalyst.catalog.SessionCatalog import org.apache.spark.sql.catalyst.expressions.Attribute import org.apache.spark.sql.catalyst.util.StringUtils +import org.apache.spark.sql.connector.catalog.{CatalogExtension, Identifier, RelationCatalog, V1View, ViewCatalog} import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.NamespaceHelper -import org.apache.spark.sql.connector.catalog.ViewCatalog import org.apache.spark.sql.execution.LeafExecNode /** * Physical plan node for SHOW VIEWS on a v2 [[ViewCatalog]]. Enumerates view identifiers via - * [[ViewCatalog#listViews]]. v2 catalogs have no temp views, so the {@code isTemporary} column - * is always false -- mirroring v1 {@code ShowViewsCommand}, which sets {@code isTemporary=true} - * only for local/global temp views that live in the session catalog. + * [[ViewCatalog#listViews]]. When the ViewCatalog is installed as the session catalog, the + * built-in session catalog contributes temporary views and persistent views that the active + * catalog positively resolves as delegated V1 views. */ case class ShowViewsExec( output: Seq[Attribute], catalog: ViewCatalog, namespace: Seq[String], - pattern: Option[String]) extends V2CommandExec with LeafExecNode { + pattern: Option[String], + v1SessionCatalog: Option[SessionCatalog] = None) extends V2CommandExec with LeafExecNode { override protected def run(): Seq[InternalRow] = { - val rows = new ArrayBuffer[InternalRow]() - catalog.listViews(namespace.toArray).foreach { ident => - val nameMatches = - pattern.forall(p => StringUtils.filterPattern(Seq(ident.name), p).nonEmpty) - if (nameMatches) { - rows += toCatalystRow(ident.namespace().quoted, ident.name(), false) + val rows = new mutable.ArrayBuffer[InternalRow]() + val seen = mutable.HashSet.empty[(String, String, Boolean)] + val delegatesSessionNamespace = catalog match { + case extension: CatalogExtension => extension.namespaceExists(namespace.toArray) + case _ => false + } + + def addView(viewNamespace: String, name: String, isTemporary: Boolean): Unit = { + if (seen.add((viewNamespace, name, isTemporary))) { + rows += toCatalystRow(viewNamespace, name, isTemporary) + } + } + + def isDelegatedV1View(ident: org.apache.spark.sql.catalyst.TableIdentifier): Boolean = { + delegatesSessionNamespace && (catalog match { + case relationCatalog: RelationCatalog => + try { + relationCatalog.loadRelation( + Identifier.of(ident.database.toArray, ident.table)).isInstanceOf[V1View] + } catch { + case _: NoSuchNamespaceException | _: NoSuchTableException => false + } + case _ => false + }) + } + + var v2NamespaceMissing: Option[NoSuchNamespaceException] = None + try { + catalog.listViews(namespace.toArray).foreach { ident => + val nameMatches = + pattern.forall(p => StringUtils.filterPattern(Seq(ident.name), p).nonEmpty) + if (nameMatches) { + addView(ident.namespace().quoted, ident.name(), isTemporary = false) + } } + } catch { + case e: NoSuchNamespaceException => v2NamespaceMissing = Some(e) } + + var v1NamespaceAttempted = false + var v1NamespaceMissing: Option[NoSuchNamespaceException] = None + v1SessionCatalog.foreach { sessionCatalog => + sessionCatalog.listLocalTempViews(pattern.getOrElse("*")).foreach { ident => + addView(ident.database.toArray.quoted, ident.table, isTemporary = true) + } + if (namespace.length == 1) { + v1NamespaceAttempted = true + val database = namespace.head + try { + sessionCatalog.listViews(database, pattern.getOrElse("*")).foreach { ident => + val isTemporary = sessionCatalog.isTempView(ident) + if (isTemporary || isDelegatedV1View(ident)) { + addView(ident.database.toArray.quoted, ident.table, isTemporary) + } + } + } catch { + case e: NoSuchNamespaceException => + v1NamespaceMissing = Some(e) + // A namespace can exist only in the custom ViewCatalog. Local temporary views still + // belong to the session and should remain visible for that valid V2 namespace. + sessionCatalog.listTempViews(database, pattern.getOrElse("*")).foreach { view => + val ident = view.identifier + addView(ident.database.toArray.quoted, ident.table, isTemporary = true) + } + } + } + } + + if (v2NamespaceMissing.isDefined && (!delegatesSessionNamespace || + !v1NamespaceAttempted || v1NamespaceMissing.isDefined)) { + throw v2NamespaceMissing.get + } + rows.toSeq } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ViewInspectionExecs.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ViewInspectionExecs.scala index 2bf4664bc4740..98e919f75167b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ViewInspectionExecs.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ViewInspectionExecs.scala @@ -33,9 +33,10 @@ import org.apache.spark.sql.errors.QueryCompilationErrors * matching the way v2 table inspection execs (e.g. `ShowCreateTableExec`, `DescribeTableExec`) * consume the [[org.apache.spark.sql.connector.catalog.Table]] attached to `ResolvedTable`. * - * Only non-session v2 views land here; the session-catalog path is rewritten to v1 commands by - * `ResolveSessionCatalog` before strategy fires. The catalog name and identifier are passed - * alongside `viewInfo` for output formatting (qualified names, EXTENDED block headers). + * ViewCatalog-backed views land here; session catalogs without ViewCatalog are rewritten to v1 + * commands by `ResolveSessionCatalog` before strategy fires. The catalog name and identifier + * are passed alongside `viewInfo` for output formatting (qualified names, EXTENDED block + * headers). */ /** diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2MetadataViewSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2MetadataViewSuite.scala index a80b2e0c21d7f..b92e81d363ad8 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2MetadataViewSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2MetadataViewSuite.scala @@ -17,11 +17,16 @@ package org.apache.spark.sql.connector +import java.util.Locale + import org.apache.spark.SparkConf import org.apache.spark.sql.{AnalysisException, Row} -import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException, NoSuchViewException, TableAlreadyExistsException, ViewAlreadyExistsException} -import org.apache.spark.sql.connector.catalog.{DelegatingTable, Identifier, Relation, RelationCatalog, Table, TableCatalog, TableChange, TableInfo, TableSummary, V1Table, View, ViewCatalog} +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.analysis.{NoSuchNamespaceException, NoSuchTableException, NoSuchViewException, TableAlreadyExistsException, ViewAlreadyExistsException} +import org.apache.spark.sql.connector.catalog.{DelegatingCatalogExtension, DelegatingTable, Identifier, Relation, RelationCatalog, Table, TableCatalog, TableChange, TableInfo, TableSummary, V1Table, V1View, View, ViewCatalog} +import org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAME import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.metricview.serde.{Column, DimensionExpression, MetricView, MetricViewFactory, SQLSource} import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.sql.types.StructType import org.apache.spark.sql.util.CaseInsensitiveStringMap @@ -126,6 +131,214 @@ class DataSourceV2MetadataViewSuite extends SharedSparkSession { // ALTER VIEW behavior tests live in the per-catalog triplet // `sql.execution.command.{,v1/,v2/}.AlterViewAsSuite{,Base}`. + test("ViewCatalog installed as spark_catalog uses v2 view commands") { + val catalogManager = spark.sessionState.catalogManager + val viewName = s"$SESSION_CATALOG_NAME.default.session_v" + val ident = Identifier.of(Array("default"), "session_v") + val metricViewName = s"$SESSION_CATALOG_NAME.default.session_mv" + val metricIdent = Identifier.of(Array("default"), "session_mv") + val v1OnlyViewName = s"$SESSION_CATALOG_NAME.default.v1_only_v" + + catalogManager.reset() + try { + sql(s"DROP VIEW IF EXISTS $v1OnlyViewName") + sql(s"CREATE VIEW $v1OnlyViewName AS SELECT 0 AS id") + withSQLConf( + SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION.key -> + classOf[TestingRelationCatalog].getName) { + catalogManager.reset() + val catalog = catalogManager.catalog(SESSION_CATALOG_NAME) + .asInstanceOf[TestingRelationCatalog] + def storedView(identifier: Identifier): View = { + catalog.getStoredView(identifier.namespace(), identifier.name()) + } + def normalizeSql(text: String): String = { + text.replace("`", "") + .trim + .stripSuffix(";") + .replaceAll("\\s+", " ") + .toLowerCase(Locale.ROOT) + } + try { + sql(s"CREATE VIEW $viewName TBLPROPERTIES ('k' = 'v1') AS SELECT 1 AS id") + assert(storedView(ident).properties().get("k") == "v1") + val listedViewNames = sql(s"SHOW VIEWS IN $SESSION_CATALOG_NAME.default") + .collect().map(_.getString(1)).toSet + assert(listedViewNames.contains(ident.name())) + assert(!listedViewNames.contains("v1_only_v"), + s"non-delegated V1 view leaked into SHOW VIEWS: $listedViewNames") + + sql(s"ALTER VIEW $viewName SET TBLPROPERTIES ('k' = 'v2')") + assert(storedView(ident).properties().get("k") == "v2") + + sql(s"ALTER VIEW $viewName AS SELECT 2 AS id") + assert(normalizeSql(storedView(ident).queryText()) == "select 2 as id") + val showCreate = normalizeSql(sql(s"SHOW CREATE TABLE $viewName").head().getString(0)) + assert(showCreate.contains(s"create view $viewName")) + + sql(s"DROP VIEW $viewName") + assert(!catalog.viewExists(ident)) + + val metricView = MetricView( + "0.1", + SQLSource("SELECT 1 AS id"), + where = None, + select = Seq(Column("id", DimensionExpression("id"), 0))) + val yaml = MetricViewFactory.toYAML(metricView) + sql( + s"""CREATE VIEW $metricViewName + |WITH METRICS + |LANGUAGE YAML + |AS + |$$$$ + |$yaml + |$$$$""".stripMargin) + assert(storedView(metricIdent).properties().get(TableCatalog.PROP_TABLE_TYPE) == + TableSummary.METRIC_VIEW_TABLE_TYPE) + sql(s"DROP VIEW $metricViewName") + assert(!catalog.viewExists(metricIdent)) + } finally { + sql(s"DROP VIEW IF EXISTS $viewName") + sql(s"DROP VIEW IF EXISTS $metricViewName") + } + } + } finally { + catalogManager.reset() + sql(s"DROP VIEW IF EXISTS $v1OnlyViewName") + catalogManager.reset() + } + } + + test("session ViewCatalog preserves delegated v1 and temporary views") { + val catalogManager = spark.sessionState.catalogManager + val delegatedView = s"$SESSION_CATALOG_NAME.default.delegated_v" + val customView = s"$SESSION_CATALOG_NAME.default.custom_v" + val tempView = "session_temp_v" + val globalTempView = "session_global_temp_v" + val customOnlyIdent = Identifier.of(Array("custom_only"), "custom_only_v") + val multiPartIdent = Identifier.of(Array("multi", "part"), "multi_part_v") + val globalCustomIdent = Identifier.of(Array("global_temp"), globalTempView) + + catalogManager.reset() + try { + sql(s"DROP VIEW IF EXISTS $delegatedView") + sql(s"CREATE VIEW $delegatedView TBLPROPERTIES ('origin' = 'v1') AS SELECT 1 AS id") + + withSQLConf( + SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION.key -> + classOf[TestingDelegatingRelationCatalog].getName) { + catalogManager.reset() + val customCatalog = catalogManager.catalog(SESSION_CATALOG_NAME) + .asInstanceOf[TestingDelegatingRelationCatalog] + try { + sql(s"ALTER VIEW $delegatedView SET TBLPROPERTIES ('updated' = 'yes')") + val metadata = spark.sessionState.catalog.getTableMetadata( + TableIdentifier("delegated_v", Some("default"))) + assert(metadata.properties.get("updated").contains("yes")) + + val propertyKeys = sql(s"SHOW TBLPROPERTIES $delegatedView") + .collect().map(_.getString(0)) + assert(!propertyKeys.exists(_.startsWith("view.")), + s"v1-internal properties leaked through v2 inspection: ${propertyKeys.mkString(", ")}") + + sql(s"CREATE VIEW $customView AS SELECT 2 AS id") + customCatalog.createView( + customOnlyIdent, + new View.Builder() + .withSchema(new StructType().add("id", "int")) + .withQueryText("SELECT 4 AS id") + .build()) + customCatalog.createView( + multiPartIdent, + new View.Builder() + .withSchema(new StructType().add("id", "int")) + .withQueryText("SELECT 6 AS id") + .build()) + customCatalog.createView( + globalCustomIdent, + new View.Builder() + .withSchema(new StructType().add("id", "int")) + .withQueryText("SELECT 7 AS id") + .build()) + sql(s"CREATE TEMP VIEW $tempView AS SELECT 3 AS id") + sql(s"CREATE GLOBAL TEMP VIEW $globalTempView AS SELECT 5 AS id") + val rows = sql(s"SHOW VIEWS IN $SESSION_CATALOG_NAME.default").collect() + assert(rows.exists(r => r.getString(0) == "default" && + r.getString(1) == "delegated_v" && !r.getBoolean(2)), + s"delegated v1 view missing from SHOW VIEWS: ${rows.mkString(", ")}") + assert(rows.exists(r => r.getString(0) == "default" && + r.getString(1) == "custom_v" && !r.getBoolean(2)), + s"custom v2 view missing from SHOW VIEWS: ${rows.mkString(", ")}") + assert(rows.exists(r => r.getString(1) == tempView && r.getBoolean(2)), + s"temporary view missing from SHOW VIEWS: ${rows.mkString(", ")}") + + val filteredRows = sql( + s"SHOW VIEWS IN $SESSION_CATALOG_NAME.default LIKE 'delegated_v'").collect() + assert(filteredRows.map(_.getString(1)).toSet == Set("delegated_v")) + + val customOnlyRows = sql( + s"SHOW VIEWS IN $SESSION_CATALOG_NAME.custom_only").collect() + assert(customOnlyRows.exists(_.getString(1) == customOnlyIdent.name()), + s"custom-only namespace was rejected by v1 fallback: ${customOnlyRows.mkString(", ")}") + assert(customOnlyRows.exists(r => r.getString(1) == tempView && r.getBoolean(2)), + s"temporary view missing from custom-only namespace: ${customOnlyRows.mkString(", ")}") + + val multiPartRows = sql( + s"SHOW VIEWS IN $SESSION_CATALOG_NAME.multi.part").collect() + assert(multiPartRows.exists(_.getString(1) == multiPartIdent.name()), + s"custom view missing from multi-part namespace: ${multiPartRows.mkString(", ")}") + assert(multiPartRows.exists(r => r.getString(1) == tempView && r.getBoolean(2)), + s"temporary view missing from multi-part namespace: ${multiPartRows.mkString(", ")}") + + val globalTempRows = sql( + s"SHOW VIEWS IN $SESSION_CATALOG_NAME.global_temp").collect() + val collidingRows = globalTempRows.filter(_.getString(1) == globalTempView) + assert(collidingRows.map(_.getBoolean(2)).toSet == Set(false, true), + s"persistent/global-temp collision was not preserved: ${globalTempRows.mkString(", ")}") + + sql(s"DROP VIEW $delegatedView") + assert(!spark.sessionState.catalog.tableExists( + TableIdentifier("delegated_v", Some("default")))) + } finally { + sql(s"DROP VIEW IF EXISTS $customView") + sql(s"DROP VIEW IF EXISTS $tempView") + sql(s"DROP VIEW IF EXISTS global_temp.$globalTempView") + customCatalog.dropView(customOnlyIdent) + customCatalog.dropView(multiPartIdent) + customCatalog.dropView(globalCustomIdent) + catalogManager.reset() + } + } + } finally { + catalogManager.reset() + sql(s"DROP VIEW IF EXISTS $delegatedView") + catalogManager.reset() + } + } + + test("session ViewCatalog does not inherit hidden V1 namespaces") { + val catalogManager = spark.sessionState.catalogManager + val v1OnlyNamespace = "v1_only" + + catalogManager.reset() + try { + sql(s"DROP DATABASE IF EXISTS $v1OnlyNamespace CASCADE") + sql(s"CREATE DATABASE $v1OnlyNamespace") + withSQLConf( + SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION.key -> + classOf[TestingFilteredNamespaceRelationCatalog].getName) { + catalogManager.reset() + intercept[NoSuchNamespaceException] { + sql(s"SHOW VIEWS IN $SESSION_CATALOG_NAME.$v1OnlyNamespace").collect() + } + } + } finally { + catalogManager.reset() + sql(s"DROP DATABASE IF EXISTS $v1OnlyNamespace CASCADE") + catalogManager.reset() + } + } + // --- Pure ViewCatalog (no TableCatalog mixin) --------------------------- test("read view from a pure ViewCatalog (no TableCatalog mixin)") { @@ -568,6 +781,93 @@ class TestingRelationCatalog extends RelationCatalog { override def name(): String = catalogName } +/** + * A session-catalog extension that keeps new v2 views in its own store while delegating tables + * and pre-existing v1 views to the built-in session catalog. + */ +class TestingDelegatingRelationCatalog extends DelegatingCatalogExtension with RelationCatalog { + private val views = + new java.util.concurrent.ConcurrentHashMap[(Seq[String], String), View]() + private val supportedNamespaces = Set( + Seq("default"), Seq("custom_only"), Seq("multi", "part"), Seq("global_temp")) + + private def key(ident: Identifier): (Seq[String], String) = { + (ident.namespace().toSeq, ident.name()) + } + + private def delegatedTableExists(ident: Identifier): Boolean = { + try { + delegate.asInstanceOf[TableCatalog].tableExists(ident) + } catch { + case _: NoSuchNamespaceException => false + } + } + + override def loadRelation(ident: Identifier): Relation = { + Option(views.get(key(ident))).getOrElse { + delegate.asInstanceOf[TableCatalog].loadTable(ident) match { + case v1: V1Table if v1.v1Table.isViewLike => new V1View(v1.v1Table) + case table => table + } + } + } + + override def loadTable(ident: Identifier): Table = loadRelation(ident) match { + case table: Table => table + case _ => throw new NoSuchTableException(ident) + } + + override def listViews(namespace: Array[String]): Array[Identifier] = { + val target = namespace.toSeq + if (!supportedNamespaces.contains(target)) { + throw new NoSuchNamespaceException(namespace) + } + val identifiers = new java.util.ArrayList[Identifier]() + views.forEach { (viewKey, _) => + if (viewKey._1 == target) { + identifiers.add(Identifier.of(viewKey._1.toArray, viewKey._2)) + } + } + identifiers.toArray(new Array[Identifier](0)) + } + + override def createView(ident: Identifier, info: View): View = { + if (delegatedTableExists(ident) || views.putIfAbsent(key(ident), info) != null) { + throw new ViewAlreadyExistsException(ident) + } + info + } + + override def replaceView(ident: Identifier, info: View): View = { + if (!views.containsKey(key(ident))) { + throw new NoSuchViewException(ident) + } + views.put(key(ident), info) + info + } + + override def dropView(ident: Identifier): Boolean = views.remove(key(ident)) != null + + override def renameView(oldIdent: Identifier, newIdent: Identifier): Unit = { + val oldKey = key(oldIdent) + val newKey = key(newIdent) + val existing = views.get(oldKey) + if (existing == null) { + throw new NoSuchViewException(oldIdent) + } + if (delegatedTableExists(newIdent) || views.putIfAbsent(newKey, existing) != null) { + throw new ViewAlreadyExistsException(newIdent) + } + views.remove(oldKey) + } +} + +class TestingFilteredNamespaceRelationCatalog extends TestingDelegatingRelationCatalog { + override def namespaceExists(namespace: Array[String]): Boolean = { + namespace.toSeq != Seq("v1_only") && super.namespaceExists(namespace) + } +} + /** * A v2 catalog that does not implement ViewCatalog. Used by capability-gate tests: the gate * fires in `Analyzer.lookupTableOrView(viewOnly=true)` for ALTER VIEW and in