From 4aad22e53c64d9be7a5030673d3bb2622fa98c2d Mon Sep 17 00:00:00 2001 From: "chen.wang" Date: Fri, 20 Mar 2026 00:42:15 +0000 Subject: [PATCH 1/8] Support metric view creation and querying with V2 catalogs (Unity Catalog) - Route metric view creation to V2 catalog when the target is not the session catalog, extracting source table dependency from YAML and passing it as a table property for the connector to forward to UC - Use parseMultipartIdentifier instead of parseTableIdentifier in Source.apply to support three-part names (catalog.schema.table) in metric view YAML source references - Replace misleading invalidLiteralForWindowDurationError with accurate SparkException.internalError messages for YAML parsing failures --- .../serde/MetricViewCanonical.scala | 2 +- .../metricview/util/MetricViewPlanner.scala | 7 +- .../command/metricViewCommands.scala | 71 ++++++++++++++++++- 3 files changed, 73 insertions(+), 7 deletions(-) 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..574d789f0ce8 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 { 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/execution/command/metricViewCommands.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/metricViewCommands.scala index 8c21a908ddf3..0e477cfb00ae 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 @@ -23,7 +23,10 @@ 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.connector.catalog.CatalogV2Util +import org.apache.spark.sql.connector.expressions.Transform import org.apache.spark.sql.errors.QueryCompilationErrors +import org.apache.spark.sql.metricview.serde.{AssetSource, MetricViewFactory} import org.apache.spark.sql.metricview.util.MetricViewPlanner import org.apache.spark.sql.types.StructType @@ -39,13 +42,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 +76,60 @@ 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 sourceTableFullName = extractSourceTable(originalText) + + val tableProperties = new java.util.HashMap[String, String]() + tableProperties.put("table_type", "METRIC_VIEW") + tableProperties.put("view_definition", originalText) + sourceTableFullName.foreach(tableProperties.put("view.dependency", _)) + comment.foreach(tableProperties.put("comment", _)) + properties.foreach { case (k, v) => tableProperties.put(k, v) } + + val columns = CatalogV2Util.structTypeToV2Columns(schema) + + tableCatalog.createTable( + ident, columns, Array.empty[Transform], tableProperties) + Seq.empty + } + + /** + * Extracts the source table three-part name from the metric view YAML. + * Only supports AssetSource (not SQLSource). Returns None if not extractable. + */ + private def extractSourceTable(yaml: String): Option[String] = { + try { + val metricView = MetricViewFactory.fromYAML(yaml) + metricView.from match { + case asset: AssetSource => Some(asset.name) + case _ => None + } + } catch { + case _: Exception => None + } + } + override protected def withNewChildInternal(newChild: LogicalPlan): LogicalPlan = { copy(child = newChild) } From 286a82b27e619701b3150f6dc45819b6b93de2ef Mon Sep 17 00:00:00 2001 From: "chen.wang" Date: Tue, 31 Mar 2026 01:09:53 +0000 Subject: [PATCH 2/8] Add METRIC_VIEW as first-class table type and use AnalysisContext for metric view ID propagation - Add CatalogTableType.METRIC_VIEW and METRIC_VIEW_TABLE_TYPE constant - Extend TableInfo with tableType, viewDefinition, and viewDependencies fields - Add Dependency, DependencyList, TableDependency, FunctionDependency classes - Add metricViewId field to AnalysisContext for definer's rights credential vending - Update V1Table to map METRIC_VIEW to the V2 table type - Update isMetricView() for backward compatibility with legacy property - Use structured TableInfo in CreateMetricViewCommand instead of flat properties --- .../sql/connector/catalog/Dependency.java | 41 +++++++++++++++ .../sql/connector/catalog/DependencyList.java | 47 +++++++++++++++++ .../connector/catalog/FunctionDependency.java | 38 ++++++++++++++ .../connector/catalog/TableDependency.java | 38 ++++++++++++++ .../sql/connector/catalog/TableInfo.java | 52 +++++++++++++++++++ .../sql/connector/catalog/TableSummary.java | 1 + .../sql/catalyst/analysis/Analyzer.scala | 8 ++- .../sql/catalyst/catalog/interface.scala | 9 ++-- .../spark/sql/connector/catalog/V1Table.scala | 1 + .../command/metricViewCommands.scala | 23 ++++---- .../spark/sql/execution/command/views.scala | 11 ++-- 11 files changed, 248 insertions(+), 21 deletions(-) create mode 100644 sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Dependency.java create mode 100644 sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/DependencyList.java create mode 100644 sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/FunctionDependency.java create mode 100644 sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableDependency.java 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..f5feb55cada6 --- /dev/null +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Dependency.java @@ -0,0 +1,41 @@ +/* + * 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}. + * This mirrors the Databricks Unity Catalog dependency model where each dependency + * identifies a specific securable object that the parent object depends on. + * + * @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..233927848b97 --- /dev/null +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/DependencyList.java @@ -0,0 +1,47 @@ +/* + * 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. + *

+ * This mirrors the Databricks Unity Catalog {@code DependencyList} model: + *

+ * + * @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/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index faf6a5e07ed1..d6c17268463e 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala @@ -156,6 +156,7 @@ case class AnalysisContext( referredTempVariableNames: Seq[Seq[String]] = Seq.empty, outerPlan: Option[LogicalPlan] = None, collation: Option[String] = None, + metricViewId: Option[String] = None, /** * This is a bridge state between this fixed-point [[Analyzer]] and a single-pass [[Resolver]]. @@ -193,7 +194,12 @@ object AnalysisContext { value.get.setSinglePassResolverBridgeState(prevSinglePassResolverBridgeState) } - private def set(context: AnalysisContext): Unit = value.set(context) + private[sql] def set(context: AnalysisContext): Unit = value.set(context) + + def setMetricViewId(viewId: String): Unit = { + val ctx = get + set(ctx.copy(metricViewId = Some(viewId))) + } def withAnalysisContext[A](viewDesc: CatalogTable)(f: => A): A = { val originContext = value.get() 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..e5756db9930c 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 @@ -731,11 +731,13 @@ object CatalogTable { 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, either by its table type or + * by the legacy property (for backward compatibility with older metric views). */ def isMetricView(table: CatalogTable): Boolean = { - table.tableType == CatalogTableType.VIEW && - table.properties.get(VIEW_WITH_METRICS).contains("true") + table.tableType == CatalogTableType.METRIC_VIEW || + (table.tableType == CatalogTableType.VIEW && + table.properties.get(VIEW_WITH_METRICS).contains("true")) } // Convert the current catalog and namespace to properties. @@ -1051,6 +1053,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/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 0e477cfb00ae..0b2d6b1ca752 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 @@ -23,8 +23,7 @@ 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.connector.catalog.CatalogV2Util -import org.apache.spark.sql.connector.expressions.Transform +import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Dependency, DependencyList, TableInfo, TableSummary} import org.apache.spark.sql.errors.QueryCompilationErrors import org.apache.spark.sql.metricview.serde.{AssetSource, MetricViewFactory} import org.apache.spark.sql.metricview.util.MetricViewPlanner @@ -97,20 +96,26 @@ case class CreateMetricViewCommand( } val schema = ViewHelper.aliasPlan(sparkSession, analyzed, userSpecifiedColumns).schema - + val columns = CatalogV2Util.structTypeToV2Columns(schema) val sourceTableFullName = extractSourceTable(originalText) val tableProperties = new java.util.HashMap[String, String]() - tableProperties.put("table_type", "METRIC_VIEW") - tableProperties.put("view_definition", originalText) - sourceTableFullName.foreach(tableProperties.put("view.dependency", _)) comment.foreach(tableProperties.put("comment", _)) properties.foreach { case (k, v) => tableProperties.put(k, v) } - val columns = CatalogV2Util.structTypeToV2Columns(schema) + val deps = sourceTableFullName.map(name => + DependencyList.of(Dependency.table(name)) + ).orNull + + val tableInfo = new TableInfo.Builder() + .withColumns(columns) + .withProperties(tableProperties) + .withTableType(TableSummary.METRIC_VIEW_TABLE_TYPE) + .withViewDefinition(originalText) + .withViewDependencies(deps) + .build() - tableCatalog.createTable( - ident, columns, Array.empty[Transform], tableProperties) + tableCatalog.createTable(ident, tableInfo) Seq.empty } 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, From 5049b16f273d289b84644a22e0242bbbbaf40140 Mon Sep 17 00:00:00 2001 From: "chen.wang" Date: Tue, 7 Apr 2026 19:33:04 +0000 Subject: [PATCH 3/8] extract dependencies from SQL source --- .../command/metricViewCommands.scala | 62 +++++++++--- .../spark/sql/execution/MetricViewSuite.scala | 95 +++++++++++++++++++ 2 files changed, 142 insertions(+), 15 deletions(-) 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 0b2d6b1ca752..69bec11eecd2 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,11 +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, TableInfo, TableSummary} import org.apache.spark.sql.errors.QueryCompilationErrors -import org.apache.spark.sql.metricview.serde.{AssetSource, MetricViewFactory} +import org.apache.spark.sql.execution.datasources.LogicalRelation +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation +import org.apache.spark.sql.metricview.serde.{AssetSource, MetricViewFactory, SQLSource} import org.apache.spark.sql.metricview.util.MetricViewPlanner import org.apache.spark.sql.types.StructType @@ -97,15 +100,17 @@ case class CreateMetricViewCommand( val schema = ViewHelper.aliasPlan(sparkSession, analyzed, userSpecifiedColumns).schema val columns = CatalogV2Util.structTypeToV2Columns(schema) - val sourceTableFullName = extractSourceTable(originalText) + val sourceTableNames = extractSourceTables(originalText, analyzed) val tableProperties = new java.util.HashMap[String, String]() comment.foreach(tableProperties.put("comment", _)) properties.foreach { case (k, v) => tableProperties.put(k, v) } - val deps = sourceTableFullName.map(name => - DependencyList.of(Dependency.table(name)) - ).orNull + val deps = if (sourceTableNames.nonEmpty) { + DependencyList.of(sourceTableNames.map(Dependency.table): _*) + } else { + null + } val tableInfo = new TableInfo.Builder() .withColumns(columns) @@ -119,19 +124,15 @@ case class CreateMetricViewCommand( Seq.empty } - /** - * Extracts the source table three-part name from the metric view YAML. - * Only supports AssetSource (not SQLSource). Returns None if not extractable. - */ - private def extractSourceTable(yaml: String): Option[String] = { + private def extractSourceTables(yaml: String, analyzed: LogicalPlan): Seq[String] = { try { val metricView = MetricViewFactory.fromYAML(yaml) metricView.from match { - case asset: AssetSource => Some(asset.name) - case _ => None + case asset: AssetSource => Seq(asset.name) + case _: SQLSource => MetricViewHelper.collectTableDependencies(analyzed) } } catch { - case _: Exception => None + case _: Exception => Seq.empty } } @@ -143,6 +144,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/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") + } + } } From 8a36654990c1a933a240de157c6916d214d21ba0 Mon Sep 17 00:00:00 2001 From: "chen.wang" Date: Mon, 13 Apr 2026 19:21:57 +0000 Subject: [PATCH 4/8] Remove metricViewId from AnalysisContext The metricViewId field and setMetricViewId method were only needed for definer's rights credential vending. With invoker's rights, the connector no longer propagates metric view identity through the AnalysisContext. Revert to upstream state. --- .../org/apache/spark/sql/catalyst/analysis/Analyzer.scala | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index d6c17268463e..faf6a5e07ed1 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala @@ -156,7 +156,6 @@ case class AnalysisContext( referredTempVariableNames: Seq[Seq[String]] = Seq.empty, outerPlan: Option[LogicalPlan] = None, collation: Option[String] = None, - metricViewId: Option[String] = None, /** * This is a bridge state between this fixed-point [[Analyzer]] and a single-pass [[Resolver]]. @@ -194,12 +193,7 @@ object AnalysisContext { value.get.setSinglePassResolverBridgeState(prevSinglePassResolverBridgeState) } - private[sql] def set(context: AnalysisContext): Unit = value.set(context) - - def setMetricViewId(viewId: String): Unit = { - val ctx = get - set(ctx.copy(metricViewId = Some(viewId))) - } + private def set(context: AnalysisContext): Unit = value.set(context) def withAnalysisContext[A](viewDesc: CatalogTable)(f: => A): A = { val originContext = value.get() From 4af2d1596cd9fcae4f4761aa214a365b5d04567e Mon Sep 17 00:00:00 2001 From: "chen.wang" Date: Tue, 14 Apr 2026 21:22:24 +0000 Subject: [PATCH 5/8] Forward createTable(Identifier, TableInfo) in DelegatingCatalogExtension DelegatingCatalogExtension forwards the older createTable overloads (StructType and Column[] variants) but was missing the TableInfo overload added in 4.1.0. Without this, catalog extensions like DeltaCatalog silently drop tableType, viewDefinition, and viewDependencies by falling through to the default implementation on TableCatalog which only passes columns/partitions/properties. --- .../sql/connector/catalog/DelegatingCatalogExtension.java | 6 ++++++ 1 file changed, 6 insertions(+) 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, From 7dabfd1b1d3de7a980934252543bc94fdc4cd540 Mon Sep 17 00:00:00 2001 From: "chen.wang" Date: Fri, 17 Apr 2026 20:45:50 +0000 Subject: [PATCH 6/8] Address review comments on metric-view V2 support Review comment fixes: - Drop the "This mirrors the Databricks Unity Catalog..." sentence from Dependency.java and DependencyList.java javadocs; these types are Spark V2 catalog APIs and the description should be generic. - Simplify CatalogTable.isMetricView to just check tableType == METRIC_VIEW, and remove the unused VIEW_WITH_METRICS constant. This PR stopped writing the legacy property, so the fallback is dead code. - In CreateMetricViewCommand.createMetricViewInV2Catalog, drop the extractSourceTables helper that branched on AssetSource/SQLSource. The analyzed plan has already resolved both source types, so MetricViewHelper.collectTableDependencies(analyzed) alone produces fully-qualified 3-part names for the dependency list. - On the V2 path, additionally capture: - metric_view.from.type / metric_view.from.name / metric_view.from.sql / metric_view.where via a new MetricView.getProperties method (mirrors the canonical metric view properties used by other platforms, e.g. Databricks). - view.catalogAndNamespace.*, view.sqlConfig.*, view.schemaMode, and query-column-name props via ViewHelper.generateViewProperties so the V2 metric view can be reloaded and re-parsed with the same catalog/namespace and SQL configs as the V1 path. - Use TableCatalog.PROP_COMMENT instead of the raw "comment" string for consistency. New tests: - MetricViewV2CatalogSuite exercises createMetricViewInV2Catalog via a non-session catalog (MetricViewRecordingCatalog built on InMemoryTableCatalog that captures the full TableInfo passed to createTable). Covers asset source, SQL source, and the new metric_view.* / view.* properties plus comment round-trip. --- .../sql/connector/catalog/Dependency.java | 2 - .../sql/connector/catalog/DependencyList.java | 1 - .../sql/catalyst/catalog/interface.scala | 12 +- .../serde/MetricViewCanonical.scala | 31 ++- .../command/metricViewCommands.scala | 36 +-- .../execution/MetricViewV2CatalogSuite.scala | 205 ++++++++++++++++++ 6 files changed, 256 insertions(+), 31 deletions(-) create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/MetricViewV2CatalogSuite.scala 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 index f5feb55cada6..b681b78f1033 100644 --- 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 @@ -23,8 +23,6 @@ * Represents a dependency of a SQL object such as a view or metric view. *

* A dependency is one of: {@link TableDependency} or {@link FunctionDependency}. - * This mirrors the Databricks Unity Catalog dependency model where each dependency - * identifies a specific securable object that the parent object depends on. * * @since 4.2.0 */ 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 index 233927848b97..6a1092bd94d7 100644 --- 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 @@ -24,7 +24,6 @@ /** * A list of dependencies for a SQL object such as a view or metric view. *

- * This mirrors the Databricks Unity Catalog {@code DependencyList} model: *