[SPARK-56920][SQL] Support METRIC_VIEW creation on V2 catalogs

common/utils/src/main/resources/error/error-conditions.json

@@ -4123,6 +4123,12 @@
     },
     "sqlState" : "KD002"
   },
+  "INVALID_METRIC_VIEW_YAML" : {
+    "message" : [
+      "Failed to parse metric view YAML: <message>"
+    ],
+    "sqlState" : "42K0L"
+  },
   "INVALID_NAME_IN_USE_COMMAND" : {
     "message" : [
       "Invalid name '<name>' in <command> command. Reason: <reason>"
@@ -8261,6 +8267,11 @@
           "The table <tableName> is a Spark data source table. Please use SHOW CREATE TABLE without AS SERDE instead."
         ]
       },
+      "ON_METRIC_VIEW" : {
+        "message" : [
+          "The command is not supported on a metric view <tableName>."
+        ]
+      },
       "ON_TEMPORARY_VIEW" : {
         "message" : [
           "The command is not supported on a temporary view <tableName>."

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

@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.catalog;
+
+import java.util.List;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * Represents a dependency of a SQL object such as a view or metric view.
+ * <p>
+ * A dependency is one of: {@link TableDependency} or {@link FunctionDependency}. The
+ * {@code sealed} declaration enforces this structurally.
+ * <p>
+ * Note: today the only producer in Spark itself is metric-view dependency extraction, which
+ * emits {@link TableDependency} only. {@link FunctionDependency} and the
+ * {@link #function(String...)} factory are exposed as groundwork for future producers
+ * (e.g. SQL UDF dependency tracking); consumers iterating a {@link DependencyList} received
+ * from Spark today should expect to see only {@link TableDependency} instances.
+ *
+ * @since 4.2.0
+ */
+@Evolving
+public sealed interface Dependency permits TableDependency, FunctionDependency {
+
+  /**
+   * Construct a {@link TableDependency} from the structural multi-part name of the dependent
+   * table. {@code nameParts} should contain at least one element; for catalog-managed tables
+   * the first element is typically the catalog name and subsequent elements are namespace
+   * components followed by the table name.
+   */
+  static TableDependency table(String... nameParts) {
+    return new TableDependency(List.of(nameParts));
+  }
+
+  /**
+   * Construct a {@link FunctionDependency} from the structural multi-part name of the
+   * dependent function. {@code nameParts} should contain at least one element; for
+   * catalog-managed functions the first element is typically the catalog name and subsequent
+   * elements are namespace components followed by the function name.
+   */
+  static FunctionDependency function(String... nameParts) {
+    return new FunctionDependency(List.of(nameParts));
+  }
+}

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

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.catalog;
+
+import java.util.List;
+import java.util.Objects;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * A list of dependencies for a SQL object such as a view or metric view.
+ * <p>
+ * <ul>
+ *   <li>When {@code null}, the dependency information is not provided.</li>
+ *   <li>When the list is empty, dependencies are provided but the object has none.</li>
+ *   <li>When the list is non-empty, each entry describes one dependency.</li>
+ * </ul>
+ * <p>
+ * {@code dependencies} is held as an immutable {@link List} so the record's auto-generated
+ * {@code equals}/{@code hashCode} delegate to per-element value semantics on the contained
+ * {@link Dependency} entries.
+ *
+ * @param dependencies list of dependencies (immutable copy made)
+ * @since 4.2.0
+ */
+@Evolving
+public record DependencyList(List<Dependency> dependencies) {
+
+  public DependencyList {
+    Objects.requireNonNull(dependencies, "dependencies must not be null");
+    dependencies = List.copyOf(dependencies);
+  }
+
+  public static DependencyList of(Dependency... dependencies) {
+    return new DependencyList(List.of(dependencies));
+  }
+}

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

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.catalog;
+
+import java.util.List;
+import java.util.Objects;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * A function dependency of a SQL object.
+ * <p>
+ * The dependent function is identified by its structural multi-part name. See
+ * {@link TableDependency} for the parts-form contract.
+ * <p>
+ * {@code nameParts} is held as an immutable {@link List} so the record's auto-generated
+ * {@code equals}/{@code hashCode} delegate to per-element value semantics.
+ *
+ * @param nameParts structural multi-part identifier (immutable copy made; never empty)
+ * @since 4.2.0
+ */
+@Evolving
+public record FunctionDependency(List<String> nameParts) implements Dependency {
+  public FunctionDependency {
+    Objects.requireNonNull(nameParts, "nameParts must not be null");
+    if (nameParts.isEmpty()) {
+      throw new IllegalArgumentException("nameParts must not be empty");
+    }
+    nameParts = List.copyOf(nameParts);
+  }
+}

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

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.catalog;
+
+import java.util.List;
+import java.util.Objects;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * A table dependency of a SQL object.
+ * <p>
+ * The dependent table is identified by its structural multi-part name. {@code nameParts}
+ * arity matches the catalog's namespace depth plus one for the table name -- for a catalog
+ * with single-level namespaces the parts are {@code [catalog, schema, table]}; for a catalog
+ * with multi-level namespaces (e.g. Iceberg with {@code db1.db2}) the parts are
+ * {@code [catalog, db1, db2, ..., table]}; for v1 sources resolved through the session
+ * catalog, producers should normalize to {@code [spark_catalog, db, table]} so consumers see
+ * a stable arity per source kind. The structural form preserves arity and is unambiguous
+ * against quoted identifiers containing a literal {@code .}; consumers that need a flat
+ * string should join the parts themselves with a quoting scheme appropriate to their wire
+ * format.
+ * <p>
+ * {@code nameParts} is held as an immutable {@link List} so the record's auto-generated
+ * {@code equals}/{@code hashCode} delegate to per-element value semantics.
+ *
+ * @param nameParts structural multi-part identifier (immutable copy made; never empty)
+ * @since 4.2.0
+ */
+@Evolving
+public record TableDependency(List<String> nameParts) implements Dependency {
+  public TableDependency {
+    Objects.requireNonNull(nameParts, "nameParts must not be null");
+    if (nameParts.isEmpty()) {
+      throw new IllegalArgumentException("nameParts must not be empty");
+    }
+    nameParts = List.copyOf(nameParts);
+  }
+}

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

@@ -27,6 +27,7 @@ public interface TableSummary {
     String EXTERNAL_TABLE_TYPE = "EXTERNAL";
     String VIEW_TABLE_TYPE = "VIEW";
     String FOREIGN_TABLE_TYPE = "FOREIGN";
+    String METRIC_VIEW_TABLE_TYPE = "METRIC_VIEW";
 
     Identifier identifier();
     String tableType();

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

@@ -48,6 +48,7 @@ public class ViewInfo extends TableInfo {
   private final Map<String, String> sqlConfigs;
   private final String schemaMode;
   private final String[] queryColumnNames;
+  private final DependencyList viewDependencies;
 
   protected ViewInfo(Builder builder) {
     super(builder);
@@ -57,11 +58,11 @@ protected ViewInfo(Builder builder) {
     this.sqlConfigs = Collections.unmodifiableMap(builder.sqlConfigs);
     this.schemaMode = builder.schemaMode;
     this.queryColumnNames = builder.queryColumnNames;
-    // Force PROP_TABLE_TYPE = VIEW so that `properties()` reflects the typed ViewInfo
-    // classification. Catalogs and generic viewers reading PROP_TABLE_TYPE from the properties
-    // bag (e.g. TableCatalog.listTableSummaries default impl, DESCRIBE) see "VIEW" without
-    // requiring authors to remember to call withTableType(VIEW).
-    properties().put(TableCatalog.PROP_TABLE_TYPE, TableSummary.VIEW_TABLE_TYPE);
+    this.viewDependencies = builder.viewDependencies;
+    // Default PROP_TABLE_TYPE = VIEW so `properties()` reflects the typed ViewInfo
+    // classification. Callers can refine to a more specific view kind (for example,
+    // METRIC_VIEW) by calling BaseBuilder.withTableType(...) on the builder before build().
+    properties().putIfAbsent(TableCatalog.PROP_TABLE_TYPE, TableSummary.VIEW_TABLE_TYPE);
   }
 
   /** The SQL text of the view. */
@@ -102,13 +103,22 @@ protected ViewInfo(Builder builder) {
    */
   public String[] queryColumnNames() { return queryColumnNames; }
 
+  /**
+   * Returns the structured list of objects this view depends on (source tables and functions),
+   * or {@code null} if no dependency list was supplied. Unlike other view metadata which is
+   * encoded into {@link #properties()}, dependency lists are a first-class field because their
+   * nested structure does not round-trip cleanly through flat string properties.
+   */
+  public DependencyList viewDependencies() { return viewDependencies; }
+
   public static class Builder extends BaseBuilder<Builder> {
     private String queryText;
     private String currentCatalog;
     private String[] currentNamespace = new String[0];
     private Map<String, String> sqlConfigs = new HashMap<>();
     private String schemaMode;
     private String[] queryColumnNames = new String[0];
+    private DependencyList viewDependencies = null;
 
     @Override
     protected Builder self() { return this; }
@@ -143,6 +153,16 @@ public Builder withQueryColumnNames(String[] queryColumnNames) {
       return this;
     }
 
+    /**
+     * Sets the structured dependency list for this view. Source tables and functions referenced
+     * by the view text should be recorded here so downstream consumers (e.g. catalogs persisting
+     * lineage) can access them without re-analyzing the view body.
+     */
+    public Builder withViewDependencies(DependencyList viewDependencies) {
+      this.viewDependencies = viewDependencies;
+      return this;
+    }
+
     @Override
     public ViewInfo build() {
       Objects.requireNonNull(columns, "columns should not be null");

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1226,7 +1226,8 @@ class Analyzer(
                 ) {
                   CatalogV2Util.loadTable(catalog, ident).map {
                     case v1Table: V1Table if CatalogV2Util.isSessionCatalog(catalog) &&
-                      v1Table.v1Table.tableType == CatalogTableType.VIEW =>
+                      (v1Table.v1Table.tableType == CatalogTableType.VIEW ||
+                        v1Table.v1Table.tableType == CatalogTableType.METRIC_VIEW) =>
                       val v1Ident = v1Table.catalogTable.identifier
                       val v2Ident = Identifier.of(v1Ident.database.toArray, v1Ident.identifier)
                       ResolvedPersistentView(

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala

@@ -398,7 +398,8 @@ class RelationResolution(
       timeTravelSpec: Option[TimeTravelSpec]): Option[LogicalPlan] = {
     def createDataSourceV1Scan(v1Table: CatalogTable): LogicalPlan = {
       if (isStreaming) {
-        if (v1Table.tableType == CatalogTableType.VIEW) {
+        if (v1Table.tableType == CatalogTableType.VIEW ||
+            v1Table.tableType == CatalogTableType.METRIC_VIEW) {
           throw QueryCompilationErrors.permanentViewNotSupportedByStreamingReadingAPIError(
             ident.quoted
           )

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/InMemoryCatalog.scala

@@ -394,7 +394,10 @@ class InMemoryCatalog(
 
   override def listViews(db: String, pattern: String): Seq[String] = synchronized {
     requireDbExists(db)
-    val views = catalog(db).tables.filter(_._2.table.tableType == CatalogTableType.VIEW).keySet
+    val views = catalog(db).tables.filter { case (_, t) =>
+      t.table.tableType == CatalogTableType.VIEW ||
+        t.table.tableType == CatalogTableType.METRIC_VIEW
+    }.keySet
     StringUtils.filterPattern(views.toSeq.sorted, pattern)
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

@@ -1256,7 +1256,8 @@ class SessionCatalog(
       import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
       val ident = nameParts.asTableIdentifier
       try {
-        getTempViewOrPermanentTableMetadata(ident).tableType == CatalogTableType.VIEW
+        val t = getTempViewOrPermanentTableMetadata(ident).tableType
+        t == CatalogTableType.VIEW || t == CatalogTableType.METRIC_VIEW
       } catch {
         case _: NoSuchTableException => false
         case _: NoSuchNamespaceException => false