Support metric view creation and querying with V2 catalogs (OSS UC)

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

@@ -111,6 +111,12 @@ public Table createTable(
     return asTableCatalog().createTable(ident, columns, partitions, properties);
   }
 
+  @Override
+  public Table createTable(Identifier ident, TableInfo tableInfo)
+      throws TableAlreadyExistsException, NoSuchNamespaceException {
+    return asTableCatalog().createTable(ident, tableInfo);
+  }
+
   @Override
   public Table alterTable(
       Identifier ident,

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

@@ -0,0 +1,39 @@
+/*
+ * 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}.
+ *
+ * @since 4.2.0
+ */
+@Evolving
+public interface Dependency {
+
+  static TableDependency table(String tableFullName) {
+    return new TableDependency(tableFullName);
+  }
+
+  static FunctionDependency function(String functionFullName) {
+    return new FunctionDependency(functionFullName);
+  }
+}

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/DependencyList.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.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>
+ *
+ * @param dependencies array of dependencies
+ * @since 4.2.0
+ */
+@Evolving
+public record DependencyList(Dependency[] dependencies) {
+
+  public DependencyList {
+    Objects.requireNonNull(dependencies, "dependencies must not be null");
+  }
+
+  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,38 @@
+/*
+ * 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.Objects;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * A function dependency of a SQL object.
+ * <p>
+ * The dependent function is identified by its fully-qualified three-part name
+ * in the form {@code catalog_name.schema_name.function_name}.
+ *
+ * @param functionFullName fully-qualified three-part function name
+ * @since 4.2.0
+ */
+@Evolving
+public record FunctionDependency(String functionFullName) implements Dependency {
+  public FunctionDependency {
+    Objects.requireNonNull(functionFullName, "functionFullName must not be null");
+  }
+}

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

@@ -0,0 +1,38 @@
+/*
+ * 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.Objects;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * A table dependency of a SQL object.
+ * <p>
+ * The dependent table is identified by its fully-qualified three-part name
+ * in the form {@code catalog_name.schema_name.table_name}.
+ *
+ * @param tableFullName fully-qualified three-part table name
+ * @since 4.2.0
+ */
+@Evolving
+public record TableDependency(String tableFullName) implements Dependency {
+  public TableDependency {
+    Objects.requireNonNull(tableFullName, "tableFullName must not be null");
+  }
+}

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

@@ -30,6 +30,9 @@ public class TableInfo {
   private final Map<String, String> properties;
   private final Transform[] partitions;
   private final Constraint[] constraints;
+  private final String tableType;
+  private final String viewDefinition;
+  private final DependencyList viewDependencies;
 
   /**
    * Constructor for TableInfo used by the builder.
@@ -40,6 +43,9 @@ private TableInfo(Builder builder) {
     this.properties = builder.properties;
     this.partitions = builder.partitions;
     this.constraints = builder.constraints;
+    this.tableType = builder.tableType;
+    this.viewDefinition = builder.viewDefinition;
+    this.viewDependencies = builder.viewDependencies;
   }
 
   public Column[] columns() {
@@ -60,11 +66,39 @@ public Transform[] partitions() {
 
   public Constraint[] constraints() { return constraints; }
 
+  /**
+   * The table type (e.g. "MANAGED", "EXTERNAL", "VIEW", "METRIC_VIEW").
+   * May be null if the connector should infer the type from properties.
+   *
+   * @see TableSummary for table type constants
+   * @since 4.2.0
+   */
+  public String tableType() { return tableType; }
+
+  /**
+   * The view definition text. For metric views this is the YAML body;
+   * for regular views this would be the SQL text. May be null for non-view tables.
+   *
+   * @since 4.2.0
+   */
+  public String viewDefinition() { return viewDefinition; }
+
+  /**
+   * The list of dependencies for this view or metric view. May be null
+   * if dependency information is not provided.
+   *
+   * @since 4.2.0
+   */
+  public DependencyList viewDependencies() { return viewDependencies; }
+
   public static class Builder {
     private Column[] columns;
     private Map<String, String> properties = new HashMap<>();
     private Transform[] partitions = new Transform[0];
     private Constraint[] constraints = new Constraint[0];
+    private String tableType;
+    private String viewDefinition;
+    private DependencyList viewDependencies;
 
     public Builder withColumns(Column[] columns) {
       this.columns = columns;
@@ -86,6 +120,24 @@ public Builder withConstraints(Constraint[] constraints) {
       return this;
     }
 
+    /** @since 4.2.0 */
+    public Builder withTableType(String tableType) {
+      this.tableType = tableType;
+      return this;
+    }
+
+    /** @since 4.2.0 */
+    public Builder withViewDefinition(String viewDefinition) {
+      this.viewDefinition = viewDefinition;
+      return this;
+    }
+
+    /** @since 4.2.0 */
+    public Builder withViewDependencies(DependencyList viewDependencies) {
+      this.viewDependencies = viewDependencies;
+      return this;
+    }
+
     public TableInfo build() {
       Objects.requireNonNull(columns, "columns should not be null");
       return new TableInfo(this);

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/scala/org/apache/spark/sql/catalyst/catalog/interface.scala

@@ -727,15 +727,9 @@ object CatalogTable {
   val VIEW_CATALOG_AND_NAMESPACE = VIEW_PREFIX + "catalogAndNamespace.numParts"
   val VIEW_CATALOG_AND_NAMESPACE_PART_PREFIX = VIEW_PREFIX + "catalogAndNamespace.part."
 
-  // Property to indicate that a VIEW is actually a METRIC VIEW
-  val VIEW_WITH_METRICS = VIEW_PREFIX + "viewWithMetrics"
-
-  /**
-   * Check if a CatalogTable is a metric view by looking at its properties.
-   */
+  /** Check if a CatalogTable is a metric view. */
   def isMetricView(table: CatalogTable): Boolean = {
-    table.tableType == CatalogTableType.VIEW &&
-      table.properties.get(VIEW_WITH_METRICS).contains("true")
+    table.tableType == CatalogTableType.METRIC_VIEW
   }
 
   // Convert the current catalog and namespace to properties.
@@ -1051,6 +1045,7 @@ object CatalogTableType {
   val EXTERNAL = new CatalogTableType("EXTERNAL")
   val MANAGED = new CatalogTableType("MANAGED")
   val VIEW = new CatalogTableType("VIEW")
+  val METRIC_VIEW = new CatalogTableType("METRIC_VIEW")
 
   val tableTypes = Seq(EXTERNAL, MANAGED, VIEW)
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/V1Table.scala

@@ -106,6 +106,7 @@ private[sql] object V1Table {
       case CatalogTableType.EXTERNAL => Some(TableSummary.EXTERNAL_TABLE_TYPE)
       case CatalogTableType.MANAGED => Some(TableSummary.MANAGED_TABLE_TYPE)
       case CatalogTableType.VIEW => Some(TableSummary.VIEW_TABLE_TYPE)
+      case CatalogTableType.METRIC_VIEW => Some(TableSummary.METRIC_VIEW_TABLE_TYPE)
       case _ => None
     }
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/metricview/serde/MetricViewCanonical.scala

@@ -94,7 +94,7 @@ private[sql] object Source {
     if (sourceText.isEmpty) {
       throw MetricViewValidationException("Source cannot be empty")
     }
-    Try(CatalystSqlParser.parseTableIdentifier(sourceText)) match {
+    Try(CatalystSqlParser.parseMultipartIdentifier(sourceText)) match {
       case Success(_) => AssetSource(sourceText)
       case Failure(_) =>
         Try(CatalystSqlParser.parseQuery(sourceText)) match {
@@ -167,4 +167,33 @@ private[sql] case class MetricView(
     version: String,
     from: Source,
     where: Option[String] = None,
-    select: Seq[Column])
+    select: Seq[Column]) {
+
+  /**
+   * Returns a set of table properties describing this metric view's source and
+   * filter clauses. Mirrors the property keys used by the canonical metric view
+   * representation on other Spark platforms so consumers of the catalog see a
+   * consistent property layout.
+   */
+  def getProperties: Map[String, String] = {
+    val base = Map(MetricView.PROP_FROM_TYPE -> from.sourceType.toString)
+    val fromProps = from match {
+      case asset: AssetSource =>
+        base + (MetricView.PROP_FROM_NAME -> asset.name)
+      case sql: SQLSource =>
+        base + (MetricView.PROP_FROM_SQL -> MetricView.truncate(sql.sql))
+    }
+    where.fold(fromProps)(w =>
+      fromProps + (MetricView.PROP_WHERE -> MetricView.truncate(w)))
+  }
+}
+
+private[sql] object MetricView {
+  final val PROP_FROM_TYPE = "metric_view.from.type"
+  final val PROP_FROM_NAME = "metric_view.from.name"
+  final val PROP_FROM_SQL = "metric_view.from.sql"
+  final val PROP_WHERE = "metric_view.where"
+
+  private def truncate(value: String): String =
+    value.take(Constants.MAXIMUM_PROPERTY_SIZE)
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/metricview/util/MetricViewPlanner.scala

@@ -23,7 +23,6 @@ import org.apache.spark.sql.catalyst.catalog.CatalogTable
 import org.apache.spark.sql.catalyst.parser.ParserInterface
 import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan}
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
-import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.metricview.logical.MetricViewPlaceholder
 import org.apache.spark.sql.metricview.serde.{AssetSource, MetricView, MetricViewFactory, MetricViewValidationException, MetricViewYAMLParsingException, SQLSource}
 import org.apache.spark.sql.types.StructType
@@ -65,9 +64,11 @@ object MetricViewPlanner {
       MetricViewFactory.fromYAML(yaml)
     } catch {
       case e: MetricViewValidationException =>
-        throw QueryCompilationErrors.invalidLiteralForWindowDurationError()
+        throw SparkException.internalError(
+          s"Invalid metric view YAML: ${e.getMessage}", e)
       case e: MetricViewYAMLParsingException =>
-        throw QueryCompilationErrors.invalidLiteralForWindowDurationError()
+        throw SparkException.internalError(
+          s"Failed to parse metric view YAML: ${e.getMessage}", e)
     }
     val source = metricView.from match {
       case asset: AssetSource => UnresolvedRelation(sqlParser.parseMultipartIdentifier(asset.name))

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala

@@ -303,12 +303,10 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
     case DropView(DropViewInSessionCatalog(ident), ifExists) =>
       DropTableCommand(ident, ifExists, isView = true, purge = false)
 
-    case DropView(r @ ResolvedIdentifier(catalog, ident), _) =>
-      if (catalog == FakeSystemCatalog) {
-        DropTempViewCommand(ident)
-      } else {
-        throw QueryCompilationErrors.catalogOperationNotSupported(catalog, "views")
-      }
+    case DropView(ResolvedIdentifier(FakeSystemCatalog, ident), _) =>
+      DropTempViewCommand(ident)
+    // For other V2 catalogs we fall through to DataSourceV2Strategy, which routes the
+    // command to DropMetricViewExec (metric views are stored as tables on the catalog).
 
     case c @ CreateNamespace(DatabaseNameInSessionCatalog(name), _, _) if conf.useV1Command =>
       val comment = c.properties.get(SupportsNamespaces.PROP_COMMENT)