[SPARK-54119] 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,58 @@
+/*
+ * 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 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(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(nameParts);
+  }
+}

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

@@ -0,0 +1,75 @@
+/*
+ * 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.Arrays;
+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 array is empty, dependencies are provided but the object has none.</li>
+ *   <li>When the array is non-empty, each entry describes one dependency.</li>
+ * </ul>
+ * <p>
+ * Records' auto-generated {@code equals}/{@code hashCode} on array fields fall through to
+ * {@link Object#equals} (reference equality), so this record overrides them to use
+ * {@link Arrays#equals(Object[], Object[])} / {@link Arrays#hashCode(Object[])} on
+ * {@code dependencies}; per-element equality delegates to the element's overridden
+ * {@code equals} ({@link TableDependency} / {@link FunctionDependency} both implement value
+ * semantics on their {@code nameParts} array). The defensive-copy accessor override clones
+ * on read so callers cannot mutate the record's internal array.
+ *
+ * @param dependencies array of dependencies; must contain no null elements (defensive
+ *                     copy made; not validated element-wise -- callers passing nulls will
+ *                     surface NPEs in downstream consumers)
+ * @since 4.2.0
+ */
+@Evolving
+public record DependencyList(Dependency[] dependencies) {
+
+  public DependencyList {
+    Objects.requireNonNull(dependencies, "dependencies must not be null");
+    dependencies = dependencies.clone();
+  }
+
+  /** Returns a defensive copy of the underlying dependencies array. */
+  @Override
+  public Dependency[] dependencies() { return dependencies.clone(); }
+
+  @Override
+  public boolean equals(Object o) {
+    return o instanceof DependencyList that && Arrays.equals(dependencies, that.dependencies);
+  }
+
+  @Override
+  public int hashCode() { return Arrays.hashCode(dependencies); }
+
+  @Override
+  public String toString() {
+    return "DependencyList[dependencies=" + Arrays.toString(dependencies) + "]";
+  }
+
+  public static DependencyList of(Dependency[] dependencies) {
+    return new DependencyList(dependencies);
+  }
+}

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

@@ -0,0 +1,68 @@
+/*
+ * 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.Arrays;
+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>
+ * Records' auto-generated {@code equals}/{@code hashCode} on array fields fall through to
+ * {@link Object#equals} (reference equality), so this record overrides them to use
+ * {@link Arrays#equals(Object[], Object[])} / {@link Arrays#hashCode(Object[])} on
+ * {@code nameParts} and give value-based semantics. The defensive-copy accessor override
+ * also clones on read so callers cannot mutate the record's internal array.
+ *
+ * @param nameParts structural multi-part identifier; must be non-empty and contain no
+ *                  null elements (defensive copy made; not validated element-wise --
+ *                  callers passing nulls will surface NPEs in downstream consumers)
+ * @since 4.2.0
+ */
+@Evolving
+public record FunctionDependency(String[] nameParts) implements Dependency {
+  public FunctionDependency {
+    Objects.requireNonNull(nameParts, "nameParts must not be null");
+    if (nameParts.length == 0) {
+      throw new IllegalArgumentException("nameParts must not be empty");
+    }
+    nameParts = nameParts.clone();
+  }
+
+  /** Returns a defensive copy of the underlying parts array. */
+  @Override
+  public String[] nameParts() { return nameParts.clone(); }
+
+  @Override
+  public boolean equals(Object o) {
+    return o instanceof FunctionDependency that && Arrays.equals(nameParts, that.nameParts);
+  }
+
+  @Override
+  public int hashCode() { return Arrays.hashCode(nameParts); }
+
+  @Override
+  public String toString() {
+    return "FunctionDependency[nameParts=" + Arrays.toString(nameParts) + "]";
+  }
+}

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

@@ -0,0 +1,76 @@
+/*
+ * 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.Arrays;
+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>
+ * Records' auto-generated {@code equals}/{@code hashCode} on array fields fall through to
+ * {@link Object#equals} (reference equality), so this record overrides them to use
+ * {@link Arrays#equals(Object[], Object[])} / {@link Arrays#hashCode(Object[])} on
+ * {@code nameParts} and give value-based semantics. The defensive-copy accessor override
+ * also clones on read so callers cannot mutate the record's internal array.
+ *
+ * @param nameParts structural multi-part identifier; must be non-empty and contain no
+ *                  null elements (defensive copy made; not validated element-wise --
+ *                  callers passing nulls will surface NPEs in downstream consumers)
+ * @since 4.2.0
+ */
+@Evolving
+public record TableDependency(String[] nameParts) implements Dependency {
+  public TableDependency {
+    Objects.requireNonNull(nameParts, "nameParts must not be null");
+    if (nameParts.length == 0) {
+      throw new IllegalArgumentException("nameParts must not be empty");
+    }
+    nameParts = nameParts.clone();
+  }
+
+  /** Returns a defensive copy of the underlying parts array. */
+  @Override
+  public String[] nameParts() { return nameParts.clone(); }
+
+  @Override
+  public boolean equals(Object o) {
+    return o instanceof TableDependency that && Arrays.equals(nameParts, that.nameParts);
+  }
+
+  @Override
+  public int hashCode() { return Arrays.hashCode(nameParts); }
+
+  @Override
+  public String toString() {
+    return "TableDependency[nameParts=" + Arrays.toString(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,7 @@ class Analyzer(
                 ) {
                   CatalogV2Util.loadTable(catalog, ident).map {
                     case v1Table: V1Table if CatalogV2Util.isSessionCatalog(catalog) &&
-                      v1Table.v1Table.tableType == CatalogTableType.VIEW =>
+                      v1Table.v1Table.isViewLike =>
                       val v1Ident = v1Table.catalogTable.identifier
                       val v2Ident = Identifier.of(v1Ident.database.toArray, v1Ident.identifier)
                       ResolvedPersistentView(