[SPARK-58895][SQL] Route custom session view catalogs through V2 commands

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,

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

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 <view> 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 <view> 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,22 +914,30 @@ 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.
    * Rejects multi-part namespaces and builtin namespace targets with appropriate errors.
    *
    * @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)) {

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

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)) =>

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, _),