diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/DelegatingCatalogExtension.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/DelegatingCatalogExtension.java index 786821514822..e71fb87c7041 100644 --- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/DelegatingCatalogExtension.java +++ b/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, diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Dependency.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Dependency.java new file mode 100644 index 000000000000..b681b78f1033 --- /dev/null +++ b/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. + *

+ * 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); + } +} diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/DependencyList.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/DependencyList.java new file mode 100644 index 000000000000..6a1092bd94d7 --- /dev/null +++ b/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. + *

+ *

+ * + * @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); + } +} diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/FunctionDependency.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/FunctionDependency.java new file mode 100644 index 000000000000..dd76190788c8 --- /dev/null +++ b/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. + *

+ * 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"); + } +} diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableDependency.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableDependency.java new file mode 100644 index 000000000000..5e44397139f4 --- /dev/null +++ b/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. + *

+ * 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"); + } +} diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableInfo.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableInfo.java index a5b4e333afa8..51750f85ff31 100644 --- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableInfo.java +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableInfo.java @@ -30,6 +30,9 @@ public class TableInfo { private final Map 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 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); diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableSummary.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableSummary.java index 8f46a372342a..17a4f23bdd1f 100644 --- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableSummary.java +++ b/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(); diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala index 8ad01e54ae9b..f9a4f5e185bb 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala +++ b/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) } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/V1Table.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/V1Table.scala index eee6ddf3e58f..5f5c38124bf6 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/V1Table.scala +++ b/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 } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/metricview/serde/MetricViewCanonical.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/metricview/serde/MetricViewCanonical.scala index 2e76a13741d0..2a6f4d7c57c5 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/metricview/serde/MetricViewCanonical.scala +++ b/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) +} diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/metricview/util/MetricViewPlanner.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/metricview/util/MetricViewPlanner.scala index 121d908eda90..b2d39057750e 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/metricview/util/MetricViewPlanner.scala +++ b/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)) 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 7efd2e111317..f4d8c46af311 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 @@ -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) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala index 3face404378c..0eabdcea4296 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala @@ -232,7 +232,8 @@ case class DropTableCommand( // If the command DROP VIEW is to drop a table or DROP TABLE is to drop a view // issue an exception. catalog.getTableMetadata(tableName).tableType match { - case CatalogTableType.VIEW if !isView => + // Both VIEW and METRIC_VIEW are conceptually views and must be dropped via DROP VIEW. + case CatalogTableType.VIEW | CatalogTableType.METRIC_VIEW if !isView => throw QueryCompilationErrors.wrongCommandForObjectTypeError( operation = "DROP TABLE", requiredType = s"${CatalogTableType.EXTERNAL.name} or ${CatalogTableType.MANAGED.name}", @@ -240,10 +241,11 @@ case class DropTableCommand( foundType = catalog.getTableMetadata(tableName).tableType.name, alternative = "DROP VIEW" ) - case o if o != CatalogTableType.VIEW && isView => + case o if o != CatalogTableType.VIEW && o != CatalogTableType.METRIC_VIEW && isView => throw QueryCompilationErrors.wrongCommandForObjectTypeError( operation = "DROP VIEW", - requiredType = CatalogTableType.VIEW.name, + requiredType = + s"${CatalogTableType.VIEW.name} or ${CatalogTableType.METRIC_VIEW.name}", objectName = catalog.getTableMetadata(tableName).qualifiedName, foundType = o.name, alternative = "DROP TABLE" diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/metricViewCommands.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/metricViewCommands.scala index 8c21a908ddf3..1e1a402e171f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/metricViewCommands.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/metricViewCommands.scala @@ -21,9 +21,14 @@ import org.apache.spark.SparkException import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.sql.catalyst.{QueryPlanningTracker, TableIdentifier} import org.apache.spark.sql.catalyst.analysis.{ResolvedIdentifier, SchemaUnsupported} -import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType} -import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan +import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType, HiveTableRelation} +import org.apache.spark.sql.catalyst.expressions.SubqueryExpression +import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, View} +import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Dependency, DependencyList, TableCatalog, TableInfo, TableSummary} import org.apache.spark.sql.errors.QueryCompilationErrors +import org.apache.spark.sql.execution.datasources.LogicalRelation +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation +import org.apache.spark.sql.metricview.serde.MetricViewFactory import org.apache.spark.sql.metricview.util.MetricViewPlanner import org.apache.spark.sql.types.StructType @@ -39,13 +44,21 @@ case class CreateMetricViewCommand( import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._ override def run(sparkSession: SparkSession): Seq[Row] = { - val catalog = sparkSession.sessionState.catalog - val name = child match { + child match { + case v: ResolvedIdentifier if !CatalogV2Util.isSessionCatalog(v.catalog) => + createMetricViewInV2Catalog(sparkSession, v) case v: ResolvedIdentifier => - v.identifier.asTableIdentifier + createMetricViewInSessionCatalog(sparkSession, v) case _ => throw SparkException.internalError( s"Failed to resolve identifier for creating metric view") } + } + + private def createMetricViewInSessionCatalog( + sparkSession: SparkSession, + resolved: ResolvedIdentifier): Seq[Row] = { + val catalog = sparkSession.sessionState.catalog + val name = resolved.identifier.asTableIdentifier val analyzed = MetricViewHelper.analyzeMetricViewText(sparkSession, name, originalText) if (userSpecifiedColumns.nonEmpty) { @@ -65,6 +78,66 @@ case class CreateMetricViewCommand( ignoreIfExists = allowExisting) Seq.empty } + + private def createMetricViewInV2Catalog( + sparkSession: SparkSession, + resolved: ResolvedIdentifier): Seq[Row] = { + val tableCatalog = resolved.catalog.asTableCatalog + val ident = resolved.identifier + val name = ident.asTableIdentifier + + val analyzed = MetricViewHelper.analyzeMetricViewText(sparkSession, name, originalText) + + if (userSpecifiedColumns.nonEmpty) { + if (userSpecifiedColumns.length > analyzed.output.length) { + throw QueryCompilationErrors.cannotCreateViewNotEnoughColumnsError( + name, userSpecifiedColumns.map(_._1), analyzed) + } else if (userSpecifiedColumns.length < analyzed.output.length) { + throw QueryCompilationErrors.cannotCreateViewTooManyColumnsError( + name, userSpecifiedColumns.map(_._1), analyzed) + } + } + + val schema = ViewHelper.aliasPlan(sparkSession, analyzed, userSpecifiedColumns).schema + val columns = CatalogV2Util.structTypeToV2Columns(schema) + val sourceTableNames = MetricViewHelper.collectTableDependencies(analyzed) + + // Mirror what the V1 path produces via ViewHelper.prepareTable: capture the + // create-time catalog/namespace and SQL configs so the view text can be + // re-parsed and resolved consistently when the metric view is loaded later. + val viewProps = ViewHelper.generateViewProperties( + properties, sparkSession, + analyzed.schema.fieldNames, schema.fieldNames, + SchemaUnsupported) + + // Describe this metric view's source and filter as table properties so + // catalogs and tools can inspect them without re-parsing the YAML. + val metricView = MetricViewFactory.fromYAML(originalText) + val metricViewProps = metricView.getProperties + + val tableProperties = new java.util.HashMap[String, String]() + comment.foreach(tableProperties.put(TableCatalog.PROP_COMMENT, _)) + viewProps.foreach { case (k, v) => tableProperties.put(k, v) } + metricViewProps.foreach { case (k, v) => tableProperties.put(k, v) } + + val deps = if (sourceTableNames.nonEmpty) { + DependencyList.of(sourceTableNames.map(Dependency.table): _*) + } else { + null + } + + val tableInfo = new TableInfo.Builder() + .withColumns(columns) + .withProperties(tableProperties) + .withTableType(TableSummary.METRIC_VIEW_TABLE_TYPE) + .withViewDefinition(originalText) + .withViewDependencies(deps) + .build() + + tableCatalog.createTable(ident, tableInfo) + Seq.empty + } + override protected def withNewChildInternal(newChild: LogicalPlan): LogicalPlan = { copy(child = newChild) } @@ -73,6 +146,37 @@ case class CreateMetricViewCommand( case class AlterMetricViewCommand(child: LogicalPlan, originalText: String) object MetricViewHelper { + + /** + * Walks the analyzed plan to collect direct table/view dependencies. + * Stops recursion at relation leaf nodes and persistent View nodes so that only + * direct (not transitive) dependencies are recorded. + */ + private[execution] def collectTableDependencies(plan: LogicalPlan): Seq[String] = { + val tables = scala.collection.mutable.ArrayBuffer.empty[String] + def traverse(p: LogicalPlan): Unit = p match { + case v: View if !v.isTempView => + tables += v.desc.identifier.unquotedString + case r: DataSourceV2Relation if r.catalog.isDefined && r.identifier.isDefined => + val cat = r.catalog.get.name() + val ns = r.identifier.get.namespace().mkString(".") + val name = r.identifier.get.name() + tables += s"$cat.$ns.$name" + case r: HiveTableRelation => + tables += r.tableMeta.identifier.unquotedString + case r: LogicalRelation if r.catalogTable.isDefined => + tables += r.catalogTable.get.identifier.unquotedString + case other => + other.children.foreach(traverse) + other.expressions.foreach(_.foreach { + case s: SubqueryExpression => traverse(s.plan) + case _ => + }) + } + traverse(plan) + tables.distinct.toSeq + } + def analyzeMetricViewText( session: SparkSession, name: TableIdentifier, diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala index 95d76c72d295..c2c2292a3aaa 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala @@ -799,19 +799,14 @@ object ViewHelper extends SQLConfHelper with Logging with CapturesConfig { val newProperties = generateViewProperties( properties, session, analyzedPlan.schema.fieldNames, aliasedSchema.fieldNames, viewSchemaMode) - // Add property to indicate if this is a metric view - val finalProperties = if (isMetricView) { - newProperties + (CatalogTable.VIEW_WITH_METRICS -> "true") - } else { - newProperties - } + val tableType = if (isMetricView) CatalogTableType.METRIC_VIEW else CatalogTableType.VIEW CatalogTable( identifier = name, - tableType = CatalogTableType.VIEW, + tableType = tableType, storage = CatalogStorageFormat.empty, schema = aliasedSchema, - properties = finalProperties, + properties = newProperties, viewOriginalText = originalText, viewText = originalText, comment = comment, 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 91e753096a23..66d2e5681286 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 @@ -33,7 +33,7 @@ import org.apache.spark.sql.catalyst.planning.PhysicalOperation import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.util.{toPrettySQL, GeneratedColumn, IdentityColumn, ResolveDefaultColumns, ResolveTableConstraints, V2ExpressionBuilder} import org.apache.spark.sql.classic.SparkSession -import org.apache.spark.sql.connector.catalog.{Identifier, StagingTableCatalog, SupportsDeleteV2, SupportsNamespaces, SupportsPartitionManagement, SupportsWrite, TableCapability, TableCatalog, TruncatableTable} +import org.apache.spark.sql.connector.catalog.{Identifier, StagingTableCatalog, SupportsDeleteV2, SupportsNamespaces, SupportsPartitionManagement, SupportsWrite, TableCapability, TableCatalog, TableSummary, TruncatableTable} import org.apache.spark.sql.connector.catalog.TableChange import org.apache.spark.sql.connector.catalog.index.SupportsIndex import org.apache.spark.sql.connector.expressions.{FieldReference, LiteralValue} @@ -97,6 +97,23 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat cacheManager.uncacheTableOrView(session, nameParts, cascade = true) } + /** + * Returns true if `ident` resolves to a metric view in `catalog`. Returns false if the table + * does not exist or carries no `table_type` property identifying it as a metric view. + */ + private def isMetricView(catalog: TableCatalog, ident: Identifier): Boolean = { + if (!catalog.tableExists(ident)) { + return false + } + val table = catalog.loadTable(ident) + TableSummary.METRIC_VIEW_TABLE_TYPE.equals( + table.properties().get(TableCatalog.PROP_TABLE_TYPE)) + } + + private def qualifiedName(catalog: TableCatalog, ident: Identifier): String = { + (catalog.name() +: ident.namespace() :+ ident.name()).mkString(".") + } + private def makeQualifiedDBObjectPath(location: String): String = { CatalogUtils.makeQualifiedDBObjectPath(session.sharedState.conf.get(WAREHOUSE_PATH), location, session.sharedState.hadoopConf) @@ -396,8 +413,38 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat } case DropTable(r: ResolvedIdentifier, ifExists, purge) => + val tableCatalog = r.catalog.asTableCatalog + // Metric views are stored as tables but must be dropped via DROP VIEW. Reject the + // command at planning time when the resolved target is a metric view so the user + // gets a clear "use DROP VIEW" error before any drop side-effect runs. + if (isMetricView(tableCatalog, r.identifier)) { + throw QueryCompilationErrors.wrongCommandForObjectTypeError( + operation = "DROP TABLE", + requiredType = + s"${TableSummary.EXTERNAL_TABLE_TYPE} or ${TableSummary.MANAGED_TABLE_TYPE}", + objectName = qualifiedName(tableCatalog, r.identifier), + foundType = TableSummary.METRIC_VIEW_TABLE_TYPE, + alternative = "DROP VIEW" + ) + } val invalidateFunc = () => CommandUtils.uncacheTableOrView(session, r) - DropTableExec(r.catalog.asTableCatalog, r.identifier, ifExists, purge, invalidateFunc) :: Nil + DropTableExec(tableCatalog, r.identifier, ifExists, purge, invalidateFunc) :: Nil + + case DropView(r: ResolvedIdentifier, ifExists) => + val tableCatalog = r.catalog.asTableCatalog + if (!tableCatalog.tableExists(r.identifier) || isMetricView(tableCatalog, r.identifier)) { + // Either the target does not exist (let DropMetricViewExec honor IF EXISTS) or it + // is a metric view (the only V2 object DROP VIEW currently supports). + DropMetricViewExec(tableCatalog, r.identifier, ifExists) :: Nil + } else { + throw QueryCompilationErrors.wrongCommandForObjectTypeError( + operation = "DROP VIEW", + requiredType = TableSummary.METRIC_VIEW_TABLE_TYPE, + objectName = qualifiedName(tableCatalog, r.identifier), + foundType = TableSummary.MANAGED_TABLE_TYPE, + alternative = "DROP TABLE" + ) + } case _: NoopCommand => LocalTableScanExec(Nil, Nil, None) :: Nil diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DropMetricViewExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DropMetricViewExec.scala new file mode 100644 index 000000000000..45404aeb0298 --- /dev/null +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DropMetricViewExec.scala @@ -0,0 +1,49 @@ +/* + * 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.execution.datasources.v2 + +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.Attribute +import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog} +import org.apache.spark.sql.errors.QueryCompilationErrors +import org.apache.spark.util.ArrayImplicits._ + +/** + * Physical plan node for `DROP VIEW` against a non-session V2 catalog when the resolved + * target is a metric view. Metric views are stored as tables on the V2 catalog (via + * `createTable(Identifier, TableInfo)`) but are conceptually views, so `DROP VIEW` is the + * supported DDL for them. The strategy layer is responsible for routing only metric-view + * drops to this exec; this class only performs the drop and honors `IF EXISTS` semantics. + */ +case class DropMetricViewExec( + catalog: TableCatalog, + ident: Identifier, + ifExists: Boolean) extends LeafV2CommandExec { + + override def run(): Seq[InternalRow] = { + if (catalog.tableExists(ident)) { + catalog.dropTable(ident) + } else if (!ifExists) { + throw QueryCompilationErrors.noSuchTableError( + (catalog.name() +: ident.namespace() :+ ident.name()).toImmutableArraySeq) + } + Seq.empty + } + + override def output: Seq[Attribute] = Seq.empty +} diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalog.scala index d21b5c730f0c..0d35380c61fd 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalog.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalog.scala @@ -334,8 +334,10 @@ class V2SessionCatalog(catalog: SessionCatalog) private def dropTableInternal(ident: Identifier, purge: Boolean = false): Boolean = { try { loadTable(ident) match { - case V1Table(v1Table) if v1Table.tableType == CatalogTableType.VIEW && - !SQLConf.get.getConf(SQLConf.DROP_TABLE_VIEW_ENABLED) => + case V1Table(v1Table) + if (v1Table.tableType == CatalogTableType.VIEW || + v1Table.tableType == CatalogTableType.METRIC_VIEW) && + !SQLConf.get.getConf(SQLConf.DROP_TABLE_VIEW_ENABLED) => throw QueryCompilationErrors.wrongCommandForObjectTypeError( operation = "DROP TABLE", requiredType = s"${CatalogTableType.EXTERNAL.name} or" + diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/MetricViewSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/MetricViewSuite.scala index 5e6033aeaa75..98bb2ee3b3cc 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/MetricViewSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/MetricViewSuite.scala @@ -18,6 +18,8 @@ package org.apache.spark.sql.execution import org.apache.spark.sql.{AnalysisException, QueryTest} +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.execution.command.MetricViewHelper import org.apache.spark.sql.metricview.serde.{AssetSource, Column, DimensionExpression, MeasureExpression, MetricView, MetricViewFactory, SQLSource} import org.apache.spark.sql.test.{SharedSparkSession, SQLTestUtils} @@ -433,4 +435,97 @@ abstract class MetricViewSuite extends QueryTest with SQLTestUtils { checkAnswer(unionDf.distinct(), df) } } + + private def analyzeAndCollectDeps(yaml: String): Seq[String] = { + val viewName = TableIdentifier("dep_test_view", Some("default")) + val analyzed = MetricViewHelper.analyzeMetricViewText(spark, viewName, yaml) + MetricViewHelper.collectTableDependencies(analyzed) + } + + test("SQL source dependency extraction - single table") { + val columns = Seq( + Column("region", DimensionExpression("region"), 0), + Column("count_sum", MeasureExpression("sum(count)"), 1) + ) + val metricView = MetricView( + "0.1", SQLSource(s"SELECT * FROM $testTableName"), None, columns) + val yaml = MetricViewFactory.toYAML(metricView) + val deps = analyzeAndCollectDeps(yaml) + assert(deps.length == 1, s"Expected 1 dependency, got ${deps.length}: $deps") + assert(deps.head.endsWith(testTableName), + s"Expected dependency to reference $testTableName, got ${deps.head}") + } + + test("SQL source dependency extraction - JOIN (multiple tables)") { + val secondTable = "test_table_customers" + withTable(secondTable) { + Seq((1, "Alice"), (2, "Bob")).toDF("id", "name") + .write.saveAsTable(secondTable) + + val columns = Seq( + Column("name", DimensionExpression("name"), 0), + Column("count_sum", MeasureExpression("sum(count)"), 1) + ) + val joinSql = + s"SELECT c.name, t.count FROM $testTableName t JOIN $secondTable c ON t.count = c.id" + val metricView = MetricView("0.1", SQLSource(joinSql), None, columns) + val yaml = MetricViewFactory.toYAML(metricView) + val deps = analyzeAndCollectDeps(yaml) + + assert(deps.length == 2, s"Expected 2 dependencies, got ${deps.length}: $deps") + assert(deps.exists(_.endsWith(testTableName)), + s"Expected dependency on $testTableName in $deps") + assert(deps.exists(_.endsWith(secondTable)), + s"Expected dependency on $secondTable in $deps") + } + } + + test("SQL source dependency extraction - subquery") { + val columns = Seq( + Column("region", DimensionExpression("region"), 0), + Column("count_sum", MeasureExpression("sum(count)"), 1) + ) + val subquerySql = + s"SELECT * FROM $testTableName WHERE count > (SELECT avg(count) FROM $testTableName)" + val metricView = MetricView("0.1", SQLSource(subquerySql), None, columns) + val yaml = MetricViewFactory.toYAML(metricView) + val deps = analyzeAndCollectDeps(yaml) + + assert(deps.length == 1, + s"Expected 1 dependency (deduplicated), got ${deps.length}: $deps") + assert(deps.head.endsWith(testTableName), + s"Expected dependency to reference $testTableName, got ${deps.head}") + } + + test("SQL source dependency extraction - self-join deduplication") { + val columns = Seq( + Column("region", DimensionExpression("a_region"), 0), + Column("count_sum", MeasureExpression("sum(a_count)"), 1) + ) + val selfJoinSql = + s"""SELECT a.region AS a_region, a.count AS a_count + |FROM $testTableName a JOIN $testTableName b ON a.region = b.region""".stripMargin + val metricView = MetricView("0.1", SQLSource(selfJoinSql), None, columns) + val yaml = MetricViewFactory.toYAML(metricView) + val deps = analyzeAndCollectDeps(yaml) + + assert(deps.length == 1, + s"Expected 1 dependency (self-join deduplicated), got ${deps.length}: $deps") + assert(deps.head.endsWith(testTableName), + s"Expected dependency to reference $testTableName, got ${deps.head}") + } + + test("AssetSource dependency extraction still works") { + val columns = Seq( + Column("region", DimensionExpression("region"), 0), + Column("count_sum", MeasureExpression("sum(count)"), 1) + ) + val metricView = MetricView("0.1", AssetSource(testTableName), None, columns) + val yaml = MetricViewFactory.toYAML(metricView) + val mv = MetricViewFactory.fromYAML(yaml) + mv.from match { + case asset: AssetSource => assert(asset.name == testTableName) + case other => fail(s"Expected AssetSource, got $other") + } + } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/MetricViewV2CatalogSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/MetricViewV2CatalogSuite.scala new file mode 100644 index 000000000000..ddc87d679903 --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/MetricViewV2CatalogSuite.scala @@ -0,0 +1,259 @@ +/* + * 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.execution + +import java.util.concurrent.ConcurrentHashMap + +import org.apache.spark.sql.QueryTest +import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryTableCatalog, Table, TableCatalog, TableInfo, TableSummary} +import org.apache.spark.sql.metricview.serde.{AssetSource, Column, DimensionExpression, MeasureExpression, MetricView, MetricViewFactory, SQLSource} +import org.apache.spark.sql.test.SharedSparkSession + +/** + * Tests that exercise [[org.apache.spark.sql.execution.command.CreateMetricViewCommand]] on a + * non-session V2 catalog. Ensures the TableInfo produced for the V2 catalog carries the + * metric view type, view definition, view dependencies, and the `metric_view.*` and + * `view.*` table properties expected by downstream catalogs. + */ +class MetricViewV2CatalogSuite extends QueryTest with SharedSparkSession { + + import testImplicits._ + + private val testCatalogName = "testcat" + private val testNamespace = "ns" + private val sourceTableName = "events" + private val fullSourceTableName = + s"$testCatalogName.$testNamespace.$sourceTableName" + private val metricViewName = "mv" + private val fullMetricViewName = + s"$testCatalogName.$testNamespace.$metricViewName" + + private val metricViewColumns = Seq( + Column("region", DimensionExpression("region"), 0), + Column("count_sum", MeasureExpression("sum(count)"), 1)) + + private val testTableData = Seq( + ("region_1", 1, 5.0), + ("region_2", 2, 10.0)) + + override protected def beforeAll(): Unit = { + super.beforeAll() + spark.conf.set( + s"spark.sql.catalog.$testCatalogName", + classOf[MetricViewRecordingCatalog].getName) + } + + override protected def afterAll(): Unit = { + spark.conf.unset(s"spark.sql.catalog.$testCatalogName") + super.afterAll() + } + + private def withTestCatalogTables(body: => Unit): Unit = { + MetricViewRecordingCatalog.reset() + testTableData.toDF("region", "count", "price") + .createOrReplaceTempView("metric_view_v2_source") + try { + sql( + s"""CREATE TABLE $fullSourceTableName + |USING foo AS SELECT * FROM metric_view_v2_source""".stripMargin) + body + } finally { + sql(s"DROP VIEW IF EXISTS $fullMetricViewName") + sql(s"DROP TABLE IF EXISTS $fullSourceTableName") + spark.catalog.dropTempView("metric_view_v2_source") + MetricViewRecordingCatalog.reset() + } + } + + private def createMetricView( + name: String, + metricView: MetricView, + comment: Option[String] = None): String = { + val yaml = MetricViewFactory.toYAML(metricView) + val commentClause = comment.map(c => s"\nCOMMENT '$c'").getOrElse("") + sql( + s"""CREATE VIEW $name + |WITH METRICS$commentClause + |LANGUAGE YAML + |AS + |$$$$ + |$yaml + |$$$$""".stripMargin) + yaml + } + + private def capturedTableInfo(): TableInfo = { + val ident = Identifier.of(Array(testNamespace), metricViewName) + val info = MetricViewRecordingCatalog.captured.get(ident) + assert(info != null, + s"Expected TableInfo for $ident to be captured by the V2 catalog") + info + } + + test("V2 catalog receives METRIC_VIEW table type and view definition") { + withTestCatalogTables { + val metricView = MetricView( + "0.1", + AssetSource(fullSourceTableName), + where = None, + select = metricViewColumns) + val yaml = createMetricView(fullMetricViewName, metricView) + + val info = capturedTableInfo() + assert(info.tableType() === TableSummary.METRIC_VIEW_TABLE_TYPE) + assert(info.viewDefinition().contains(yaml.trim)) + val deps = info.viewDependencies() + assert(deps != null) + assert(deps.dependencies().length === 1) + val tableDep = deps.dependencies()(0).asInstanceOf[ + org.apache.spark.sql.connector.catalog.TableDependency] + assert(tableDep.tableFullName() === fullSourceTableName) + } + } + + test("V2 catalog path populates metric_view.* and view.* properties") { + withTestCatalogTables { + val metricView = MetricView( + "0.1", + AssetSource(fullSourceTableName), + where = Some("count > 0"), + select = metricViewColumns) + createMetricView(fullMetricViewName, metricView) + + val props = capturedTableInfo().properties() + + // metric_view.* descriptive properties (mirrors DBR SingleSourceMetricView). + assert(props.get(MetricView.PROP_FROM_TYPE) === "ASSET") + assert(props.get(MetricView.PROP_FROM_NAME) === fullSourceTableName) + assert(props.get(MetricView.PROP_FROM_SQL) === null) + assert(props.get(MetricView.PROP_WHERE) === "count > 0") + + // view.* properties produced by ViewHelper.generateViewProperties, needed + // to re-parse the view text consistently when the metric view is loaded. + import org.apache.spark.sql.catalyst.catalog.CatalogTable + assert(props.containsKey(CatalogTable.VIEW_CATALOG_AND_NAMESPACE), + s"Expected ${CatalogTable.VIEW_CATALOG_AND_NAMESPACE} in $props") + val sqlConfigKeys = props.keySet().stream() + .filter(_.startsWith(CatalogTable.VIEW_SQL_CONFIG_PREFIX)) + .count() + assert(sqlConfigKeys > 0, + s"Expected at least one ${CatalogTable.VIEW_SQL_CONFIG_PREFIX}* property in $props") + } + } + + test("DROP VIEW succeeds on a V2 metric view; DROP TABLE rejects it") { + withTestCatalogTables { + val metricView = MetricView( + "0.1", + AssetSource(fullSourceTableName), + where = None, + select = metricViewColumns) + createMetricView(fullMetricViewName, metricView) + val ident = Identifier.of(Array(testNamespace), metricViewName) + + // DROP TABLE on a metric view must fail with WRONG_COMMAND_FOR_OBJECT_TYPE. + val dropTableEx = intercept[org.apache.spark.sql.AnalysisException] { + sql(s"DROP TABLE $fullMetricViewName") + } + assert(dropTableEx.getCondition === "WRONG_COMMAND_FOR_OBJECT_TYPE") + assert(dropTableEx.getMessage.contains("DROP VIEW")) + + // The metric view must still exist after the failed DROP TABLE. + assert(MetricViewRecordingCatalog.captured.containsKey(ident)) + + // DROP VIEW on a metric view must succeed. + sql(s"DROP VIEW $fullMetricViewName") + // The catalog no longer carries the underlying table either. + assert(!sql(s"SHOW TABLES IN $testCatalogName.$testNamespace") + .where("tableName = 'mv'") + .collect().nonEmpty) + } + } + + test("DROP VIEW IF EXISTS on a non-existent V2 metric view is a no-op") { + withTestCatalogTables { + sql(s"DROP VIEW IF EXISTS $testCatalogName.$testNamespace.does_not_exist") + } + } + + test("DROP VIEW on a regular V2 table fails with WRONG_COMMAND_FOR_OBJECT_TYPE") { + withTestCatalogTables { + // Source table is a regular table, not a metric view. + val ex = intercept[org.apache.spark.sql.AnalysisException] { + sql(s"DROP VIEW $fullSourceTableName") + } + assert(ex.getCondition === "WRONG_COMMAND_FOR_OBJECT_TYPE") + assert(ex.getMessage.contains("DROP TABLE")) + } + } + + test("V2 catalog path captures SQL source and comment") { + withTestCatalogTables { + val metricView = MetricView( + "0.1", + SQLSource(s"SELECT * FROM $fullSourceTableName"), + where = None, + select = metricViewColumns) + createMetricView(fullMetricViewName, metricView, comment = Some("my mv")) + + val info = capturedTableInfo() + assert(info.tableType() === TableSummary.METRIC_VIEW_TABLE_TYPE) + + val props = info.properties() + assert(props.get(MetricView.PROP_FROM_TYPE) === "SQL") + assert(props.get(MetricView.PROP_FROM_NAME) === null) + assert(props.get(MetricView.PROP_FROM_SQL) === + s"SELECT * FROM $fullSourceTableName") + assert(props.get(TableCatalog.PROP_COMMENT) === "my mv") + + // SQL source still produces the source-table dependency via the analyzed plan. + val deps = info.viewDependencies() + assert(deps != null && deps.dependencies().length === 1) + val tableDep = deps.dependencies()(0).asInstanceOf[ + org.apache.spark.sql.connector.catalog.TableDependency] + assert(tableDep.tableFullName() === fullSourceTableName) + } + } +} + +/** + * Minimal V2 catalog used by [[MetricViewV2CatalogSuite]] to capture the full + * [[TableInfo]] passed to `createTable(Identifier, TableInfo)` so tests can + * assert on `tableType`, `viewDefinition`, `viewDependencies`, and properties. + * + * It also propagates `tableInfo.tableType()` into the stored table's properties as + * `TableCatalog.PROP_TABLE_TYPE`, mirroring how production V2 catalogs surface the + * type so that DROP VIEW / DROP TABLE routing on metric views can be exercised. + */ +class MetricViewRecordingCatalog extends InMemoryTableCatalog { + override def createTable(ident: Identifier, tableInfo: TableInfo): Table = { + MetricViewRecordingCatalog.captured.put(ident, tableInfo) + val propsWithType = new java.util.HashMap[String, String](tableInfo.properties()) + Option(tableInfo.tableType()).foreach { t => + propsWithType.put(TableCatalog.PROP_TABLE_TYPE, t) + } + super.createTable(ident, tableInfo.columns(), tableInfo.partitions(), propsWithType) + } +} + +object MetricViewRecordingCatalog { + val captured: ConcurrentHashMap[Identifier, TableInfo] = + new ConcurrentHashMap[Identifier, TableInfo]() + + def reset(): Unit = captured.clear() +} diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala index 8d7eeff6edea..9c3161b6972b 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala @@ -1091,7 +1091,7 @@ abstract class DDLSuite extends QueryTest with DDLSuiteBase { "alternative" -> "DROP TABLE", "operation" -> "DROP VIEW", "foundType" -> "EXTERNAL", - "requiredType" -> "VIEW", + "requiredType" -> "VIEW or METRIC_VIEW", "objectName" -> "spark_catalog.dbx.tab1") ) } diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala index a73736fbde7a..4357cef91caf 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala @@ -1144,7 +1144,7 @@ class HiveDDLSuite "alternative" -> "DROP TABLE", "operation" -> "DROP VIEW", "foundType" -> "MANAGED", - "requiredType" -> "VIEW", + "requiredType" -> "VIEW or METRIC_VIEW", "objectName" -> s"$SESSION_CATALOG_NAME.default.tab1" ) )