szehon-ho commented on code in PR #14984: URL: https://github.com/apache/iceberg/pull/14984#discussion_r3724915306
########## spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckViews.scala: ########## @@ -0,0 +1,126 @@ +/* + * 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.catalyst.analysis + +import org.apache.spark.sql.catalyst.expressions.SubqueryExpression +import org.apache.spark.sql.catalyst.plans.logical.AlterViewAs +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan +import org.apache.spark.sql.catalyst.plans.logical.Project +import org.apache.spark.sql.catalyst.plans.logical.SubqueryAlias +import org.apache.spark.sql.catalyst.plans.logical.View +import org.apache.spark.sql.catalyst.plans.logical.views.CreateIcebergView +import org.apache.spark.sql.catalyst.plans.logical.views.ResolvedV2View +import org.apache.spark.sql.connector.catalog.ViewCatalog +import org.apache.spark.sql.errors.QueryCompilationErrors +import org.apache.spark.sql.execution.command.ViewHelper + +object CheckViews extends (LogicalPlan => Unit) { + + import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._ + + override def apply(plan: LogicalPlan): Unit = { + plan foreach { + // RewriteViewCommands replaces Spark's CreateView before CheckViewReferences runs, so apply + // Spark's shared post-analysis checks to the Iceberg node here. Cycle detection remains + // Iceberg-specific because ResolveViews expands V2 views without logical View nodes. Review Comment: The first sentence is the real reason and still holds. The trailing clause is stale though: `ResolveViews` no longer expands V2 views, so they do arrive as logical `View` nodes. Cycle detection still needs to live here, but because `RewriteViewCommands` replaces `CreateView` so `CheckViewReferences` never runs — not because of the plan shape. On your question of whether this can move to Spark: see my reply on the thread above. ########## spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala: ########## @@ -0,0 +1,149 @@ +/* + * 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.catalyst.analysis + +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.analysis.ViewUtil.IcebergViewHelper +import org.apache.spark.sql.catalyst.expressions.Alias +import org.apache.spark.sql.catalyst.expressions.SubqueryExpression +import org.apache.spark.sql.catalyst.expressions.UpCast +import org.apache.spark.sql.catalyst.parser.ParseException +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan +import org.apache.spark.sql.catalyst.plans.logical.Project +import org.apache.spark.sql.catalyst.plans.logical.SubqueryAlias +import org.apache.spark.sql.catalyst.plans.logical.views.ResolvedV2View +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.catalyst.trees.CurrentOrigin +import org.apache.spark.sql.catalyst.trees.Origin +import org.apache.spark.sql.connector.catalog.CatalogManager +import org.apache.spark.sql.connector.catalog.CatalogPlugin +import org.apache.spark.sql.connector.catalog.LookupCatalog +import org.apache.spark.sql.connector.catalog.View +import org.apache.spark.sql.errors.QueryCompilationErrors + +case class ResolveViews(spark: SparkSession) extends Rule[LogicalPlan] with LookupCatalog { + + import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._ + + protected lazy val catalogManager: CatalogManager = spark.sessionState.catalogManager + + override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators { + case u @ UnresolvedRelation(nameParts, _, _) + if catalogManager.v1SessionCatalog.isTempView(nameParts) => + u + + case u @ UnresolvedRelation(parts @ CatalogAndIdentifier(catalog, ident), _, _) => + ViewUtil + .loadView(catalog, ident) + .map(createViewRelation(parts, catalog, _)) + .getOrElse(u) + + case u @ UnresolvedTableOrView(CatalogAndIdentifier(catalog, ident), _, _, _) => + ViewUtil + .loadView(catalog, ident) + .map(_ => ResolvedV2View(catalog.asViewCatalog, ident)) + .getOrElse(u) + } + + private def createViewRelation( + nameParts: Seq[String], + catalog: CatalogPlugin, + view: View): LogicalPlan = { + val parsed = parseViewText(nameParts.quoted, view.queryText) + + // Apply any necessary rewrites to preserve correct resolution + val viewCatalogAndNamespace: Seq[String] = + Option(view.currentCatalog).getOrElse(catalog.name()) +: view.currentNamespace.toSeq + val rewritten = rewriteIdentifiers(parsed, viewCatalogAndNamespace) + + // Apply the field aliases and column comments + // This logic differs from how Spark handles views in SessionCatalog.fromCatalogTable. + // This is more strict because it doesn't allow resolution by field name. + val aliases = view.schema.fields.zipWithIndex.map { case (expected, pos) => + val attr = GetColumnByOrdinal(pos, expected.dataType) + Alias(UpCast(attr, expected.dataType), expected.name)(explicitMetadata = + Some(expected.metadata)) + }.toIndexedSeq + + // Preserve Iceberg's positional schema mapping as an alias over a projection. This differs Review Comment: If the expansion above is unreachable, this comment is no longer accurate — Spark produces a logical `View` node rather than this `SubqueryAlias(Project(...))` shape. ########## spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveViews.scala: ########## @@ -0,0 +1,149 @@ +/* + * 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.catalyst.analysis + +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.analysis.ViewUtil.IcebergViewHelper +import org.apache.spark.sql.catalyst.expressions.Alias +import org.apache.spark.sql.catalyst.expressions.SubqueryExpression +import org.apache.spark.sql.catalyst.expressions.UpCast +import org.apache.spark.sql.catalyst.parser.ParseException +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan +import org.apache.spark.sql.catalyst.plans.logical.Project +import org.apache.spark.sql.catalyst.plans.logical.SubqueryAlias +import org.apache.spark.sql.catalyst.plans.logical.views.ResolvedV2View +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.catalyst.trees.CurrentOrigin +import org.apache.spark.sql.catalyst.trees.Origin +import org.apache.spark.sql.connector.catalog.CatalogManager +import org.apache.spark.sql.connector.catalog.CatalogPlugin +import org.apache.spark.sql.connector.catalog.LookupCatalog +import org.apache.spark.sql.connector.catalog.View +import org.apache.spark.sql.errors.QueryCompilationErrors + +case class ResolveViews(spark: SparkSession) extends Rule[LogicalPlan] with LookupCatalog { + + import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._ + + protected lazy val catalogManager: CatalogManager = spark.sessionState.catalogManager + + override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators { + case u @ UnresolvedRelation(nameParts, _, _) + if catalogManager.v1SessionCatalog.isTempView(nameParts) => + u + + case u @ UnresolvedRelation(parts @ CatalogAndIdentifier(catalog, ident), _, _) => + ViewUtil + .loadView(catalog, ident) + .map(createViewRelation(parts, catalog, _)) + .getOrElse(u) + + case u @ UnresolvedTableOrView(CatalogAndIdentifier(catalog, ident), _, _, _) => + ViewUtil + .loadView(catalog, ident) + .map(_ => ResolvedV2View(catalog.asViewCatalog, ident)) + .getOrElse(u) + } + + private def createViewRelation( Review Comment: This expansion path looks unreachable on 4.2. `ResolveRelations` runs ahead of `extendedResolutionRules`, and now that `BaseCatalog` implements `RelationCatalog`, `loadRelation` falls back to `loadView` and `RelationResolution.createRelation` turns the view into a V1 scan before `ResolveViews` ever sees an unresolved relation. The test diff seems to confirm it empirically rather than just by reasoning: invalid view text now reports `[PARSE_SYNTAX_ERROR] ... SQL of VIEW` from Spark's `parseQuery`, instead of Iceberg's `invalidViewNameError` from `parseViewText` below. If that is right, `createViewRelation` (64), `parseViewText` (90), `rewriteIdentifiers` (103) and `isBuiltinFunction` (146) can all go; `ResolvedV2View` is still needed by `RewriteViewCommands`. This also answers your other question on this file — as far as I can tell essentially none of the expansion logic is still needed, and none of the deletion depends on a Spark-side change. One consequence worth calling out separately: dropping `rewriteIdentifiers` means Iceberg stops rewriting `system.bucket`-style references and defers to Spark's function resolution. Is that the accepted outcome, or should it be fixed upstream? ########## spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkView.java: ########## @@ -0,0 +1,186 @@ +/* + * 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.iceberg.spark.source; + +import static org.apache.iceberg.TableProperties.FORMAT_VERSION; + +import java.util.Map; +import java.util.Set; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; +import org.apache.iceberg.spark.SparkSchemaUtil; +import org.apache.iceberg.view.BaseView; +import org.apache.iceberg.view.SQLViewRepresentation; +import org.apache.iceberg.view.ViewOperations; +import org.apache.spark.sql.connector.catalog.DependencyList; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.connector.catalog.TableSummary; +import org.apache.spark.sql.connector.catalog.View; + +/** + * Converts Iceberg view metadata to Spark's {@link View} representation. + * + * <p>Keeps this conversion in Iceberg instead of relying on Spark's built-in view handling so + * reserved properties and Iceberg defaults are exposed consistently to Spark commands. + */ +public class SparkView { + + public static final String PROP_CREATE_ENGINE_VERSION = "create_engine_version"; + public static final String PROP_ENGINE_VERSION = "engine_version"; + public static final String QUERY_COLUMN_NAMES = "spark.query-column-names"; + public static final String SQL_CONFIG_PREFIX = "spark.sql-config."; + public static final String VIEW_SCHEMA_MODE = "spark.view-schema-mode"; + public static final String VIEW_DEPENDENCIES = "spark.view-dependencies"; + private static final Set<String> INTERNAL_PROPERTIES = + ImmutableSet.of( + TableCatalog.PROP_PROVIDER, + TableCatalog.PROP_LOCATION, + TableCatalog.PROP_TABLE_TYPE, + FORMAT_VERSION, + PROP_CREATE_ENGINE_VERSION, + PROP_ENGINE_VERSION, + QUERY_COLUMN_NAMES, + VIEW_SCHEMA_MODE, + VIEW_DEPENDENCIES); + + private SparkView() {} + + public static View toView(String catalogName, org.apache.iceberg.view.View icebergView) { + SQLViewRepresentation sqlRepr = icebergView.sqlFor("spark"); + Preconditions.checkState(sqlRepr != null, "Cannot load SQL for view %s", icebergView.name()); + + Namespace defaultNamespace = icebergView.currentVersion().defaultNamespace(); + String defaultCatalog = icebergView.currentVersion().defaultCatalog(); + + View.Builder builder = + new View.Builder() + .withQueryText(sqlRepr.sql()) + .withCurrentCatalog(defaultCatalog != null ? defaultCatalog : catalogName) + .withCurrentNamespace( + defaultNamespace != null ? defaultNamespace.levels() : new String[0]) + .withSchema(SparkSchemaUtil.convert(icebergView.schema())) + .withQueryColumnNames(queryColumnNames(icebergView.properties())) + .withProperties(properties(icebergView)); + + String schemaMode = icebergView.properties().get(VIEW_SCHEMA_MODE); Review Comment: `withSchemaMode` is only called when `spark.view-schema-mode` is present. For any view created by an earlier Iceberg release, by Trino or Flink, or directly through the Java API, the property is absent, so `schemaMode()` stays null, nothing lands in the `CatalogTable`, and `CatalogTable.viewSchemaModeFromProperties` defaults to `SchemaBinding` (`interface.scala:831`). Meanwhile the test diff shows a freshly created view emitting `WITH SCHEMA COMPENSATION`. So two views with identical DDL behave differently depending on which engine wrote them, and the stricter mode is the one that lands on the legacy population. `SchemaBinding` up-casts view output to the recorded schema and fails on anything not up-castable, so if a referenced table's column widened after view creation, a read that worked on 4.1 can now fail. Now that Spark's expansion is the only read path, this one line governs how every pre-existing and cross-engine view is read, so I think the default should be chosen explicitly here rather than inherited from Spark. Ideally with a test that builds a view via `viewCatalog().buildView(...)` with no Spark properties, changes a column type, then reads it. ########## spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckViews.scala: ########## @@ -0,0 +1,126 @@ +/* + * 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.catalyst.analysis + +import org.apache.spark.sql.catalyst.expressions.SubqueryExpression +import org.apache.spark.sql.catalyst.plans.logical.AlterViewAs +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan +import org.apache.spark.sql.catalyst.plans.logical.Project +import org.apache.spark.sql.catalyst.plans.logical.SubqueryAlias +import org.apache.spark.sql.catalyst.plans.logical.View +import org.apache.spark.sql.catalyst.plans.logical.views.CreateIcebergView +import org.apache.spark.sql.catalyst.plans.logical.views.ResolvedV2View +import org.apache.spark.sql.connector.catalog.ViewCatalog +import org.apache.spark.sql.errors.QueryCompilationErrors +import org.apache.spark.sql.execution.command.ViewHelper + +object CheckViews extends (LogicalPlan => Unit) { + + import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._ + + override def apply(plan: LogicalPlan): Unit = { + plan foreach { + // RewriteViewCommands replaces Spark's CreateView before CheckViewReferences runs, so apply + // Spark's shared post-analysis checks to the Iceberg node here. Cycle detection remains + // Iceberg-specific because ResolveViews expands V2 views without logical View nodes. + case c: CreateIcebergView if c.isAnalyzed => + c.child match { + case resolvedIdent @ ResolvedIdentifier(_: ViewCatalog, _) => + val viewIdent: Seq[String] = + resolvedIdent.catalog.name() +: resolvedIdent.identifier.asMultipartIdentifier + ViewHelper.verifyTemporaryObjectsNotExists( + isTemporary = false, + viewIdent, + c.query, + c.referredTempFunctions) + ViewHelper.verifyAutoGeneratedAliasesNotExists(c.query, isTemporary = false, viewIdent) + verifyColumnCount(resolvedIdent, c.columnAliases, c.query) + if (c.replace) { + checkCyclicViewReference(viewIdent, c.query, Seq(viewIdent)) + } + + case _ => // OK + } + + case AlterViewAs(ResolvedV2View(_, _), _, _, _, _) => + throw new IcebergAnalysisException( + "ALTER VIEW <viewName> AS is not supported. Use CREATE OR REPLACE VIEW instead") + + case _ => // OK + } + } + + private def verifyColumnCount( + ident: ResolvedIdentifier, + columns: Seq[String], + query: LogicalPlan): Unit = { + if (columns.nonEmpty) { + val viewNameParts = ident.catalog.name() +: ident.identifier.asMultipartIdentifier + if (columns.length > query.output.length) { + throw QueryCompilationErrors.cannotCreateViewNotEnoughColumnsError( + viewNameParts, + columns, + query) + } else if (columns.length < query.output.length) { + throw QueryCompilationErrors.cannotCreateViewTooManyColumnsError( + viewNameParts, + columns, + query) + } + } + } + + private def checkCyclicViewReference( + viewIdent: Seq[String], + plan: LogicalPlan, + cyclePath: Seq[Seq[String]]): Unit = { + plan match { + case sub @ SubqueryAlias(_, Project(_, _)) => Review Comment: With `ResolveViews` no longer producing this shape, this branch looks dead — the `v1View: View` case below is what fires now. Cycle detection itself is fine, and the switch to `desc.fullIdent` is required rather than cosmetic, since `fullIdent` prefers `multipartIdentifier` while `identifier` goes through `asLegacyTableIdentifier` and flattens multi-level namespaces. But this case and the `SubqueryAlias` import at line 25 can probably go with the rest. ########## spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkView.java: ########## @@ -0,0 +1,186 @@ +/* + * 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.iceberg.spark.source; + +import static org.apache.iceberg.TableProperties.FORMAT_VERSION; + +import java.util.Map; +import java.util.Set; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; +import org.apache.iceberg.spark.SparkSchemaUtil; +import org.apache.iceberg.view.BaseView; +import org.apache.iceberg.view.SQLViewRepresentation; +import org.apache.iceberg.view.ViewOperations; +import org.apache.spark.sql.connector.catalog.DependencyList; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.connector.catalog.TableSummary; +import org.apache.spark.sql.connector.catalog.View; + +/** + * Converts Iceberg view metadata to Spark's {@link View} representation. + * + * <p>Keeps this conversion in Iceberg instead of relying on Spark's built-in view handling so + * reserved properties and Iceberg defaults are exposed consistently to Spark commands. + */ +public class SparkView { + + public static final String PROP_CREATE_ENGINE_VERSION = "create_engine_version"; + public static final String PROP_ENGINE_VERSION = "engine_version"; + public static final String QUERY_COLUMN_NAMES = "spark.query-column-names"; + public static final String SQL_CONFIG_PREFIX = "spark.sql-config."; + public static final String VIEW_SCHEMA_MODE = "spark.view-schema-mode"; + public static final String VIEW_DEPENDENCIES = "spark.view-dependencies"; + private static final Set<String> INTERNAL_PROPERTIES = + ImmutableSet.of( + TableCatalog.PROP_PROVIDER, + TableCatalog.PROP_LOCATION, + TableCatalog.PROP_TABLE_TYPE, + FORMAT_VERSION, + PROP_CREATE_ENGINE_VERSION, + PROP_ENGINE_VERSION, + QUERY_COLUMN_NAMES, + VIEW_SCHEMA_MODE, + VIEW_DEPENDENCIES); + + private SparkView() {} + + public static View toView(String catalogName, org.apache.iceberg.view.View icebergView) { + SQLViewRepresentation sqlRepr = icebergView.sqlFor("spark"); + Preconditions.checkState(sqlRepr != null, "Cannot load SQL for view %s", icebergView.name()); + + Namespace defaultNamespace = icebergView.currentVersion().defaultNamespace(); + String defaultCatalog = icebergView.currentVersion().defaultCatalog(); + + View.Builder builder = + new View.Builder() + .withQueryText(sqlRepr.sql()) + .withCurrentCatalog(defaultCatalog != null ? defaultCatalog : catalogName) + .withCurrentNamespace( + defaultNamespace != null ? defaultNamespace.levels() : new String[0]) + .withSchema(SparkSchemaUtil.convert(icebergView.schema())) + .withQueryColumnNames(queryColumnNames(icebergView.properties())) + .withProperties(properties(icebergView)); + + String schemaMode = icebergView.properties().get(VIEW_SCHEMA_MODE); + Map<String, String> sqlConfigs = sqlConfigs(icebergView.properties()); + DependencyList dependencies = viewDependencies(icebergView.properties()); + + return applyOptionalFields(builder, schemaMode, sqlConfigs, dependencies).build(); + } + + /** Applies optional Spark view fields to a view builder when present. */ + public static View.Builder applyOptionalFields( + View.Builder builder, + String schemaMode, + Map<String, String> sqlConfigs, + DependencyList dependencies) { + if (schemaMode != null) { + builder.withSchemaMode(schemaMode); + } + + if (sqlConfigs != null) { + builder.withSqlConfigs(sqlConfigs); + } + + if (dependencies != null) { + builder.withViewDependencies(dependencies); + } + + return builder; + } + + /** + * Returns reserved properties that preserve Spark view metadata not represented by Iceberg view + * metadata fields. + */ + public static Map<String, String> internalProperties(View view) { + ImmutableMap.Builder<String, String> propsBuilder = ImmutableMap.builder(); + String tableType = view.properties().get(TableCatalog.PROP_TABLE_TYPE); + if (tableType != null && !TableSummary.VIEW_TABLE_TYPE.equals(tableType)) { + propsBuilder.put(TableCatalog.PROP_TABLE_TYPE, tableType); + } + + propsBuilder.put(QUERY_COLUMN_NAMES, String.join(",", view.queryColumnNames())); Review Comment: Joined with `,` here and read back with `split(",")` at line 144, so a query column name containing a comma round-trips as two names. `SessionCatalog.fromCatalogTable` then asserts the count matches the schema ("Corrupted view metadata detected for view"): ```sql CREATE VIEW v AS SELECT id AS `a,b` FROM t; SELECT * FROM v; ``` The encoding predates this PR, but it was latent on 4.1 because the positional expansion never read the property back — now it is on the read path. Since this becomes on-disk view metadata, it would be good to fix before the format is in the wild. `SparkViewDependenciesParser` in this PR is the precedent: a JSON array, with a fallback read of the legacy comma format. ########## spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/IcebergAlterV2ViewSetPropertiesExec.scala: ########## @@ -0,0 +1,69 @@ +/* + * 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.iceberg.spark.Spark3Util +import org.apache.iceberg.spark.source.SparkView +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.analysis.ViewUtil +import org.apache.spark.sql.catalyst.expressions.Attribute +import org.apache.spark.sql.connector.catalog.Identifier +import org.apache.spark.sql.connector.catalog.ViewCatalog + +/** + * Executes ALTER VIEW SET TBLPROPERTIES for Spark V2 views. + * + * Uses a custom command instead of Spark's built-in implementation so Iceberg catalogs commit + * property-only metadata updates and reject changes to reserved view properties. + */ +case class IcebergAlterV2ViewSetPropertiesExec( + catalog: ViewCatalog, + ident: Identifier, + properties: Map[String, String]) + extends LeafV2CommandExec { + + override lazy val output: Seq[Attribute] = Nil + + override protected def run(): Seq[InternalRow] = { + properties.keys.foreach(verifyNonReservedPropertyIsSet) + + val icebergViewCatalog = + ViewUtil + .icebergViewCatalog(catalog, ident) + .getOrElse( + throw new IllegalStateException( + s"Cannot load underlying Iceberg view catalog for view: $ident")) + val view = icebergViewCatalog.loadView(Spark3Util.identifierToTableIdentifier(ident)) + val update = view.updateProperties() + properties.foreach { case (key, value) => update.set(key, value) } + update.commit() Review Comment: Spark's `AlterV2ViewSetPropertiesExec` calls `CommandUtils.uncacheTableOrView(session, ResolvedIdentifier(catalog, identifier))` before committing, specifically to match v1's `invalidateCachedTable`. This exec commits without it, as do the unset and rename ones, so cached query plans referencing the view can keep serving stale entries. I have not run a repro for this, so worth confirming before acting on it. ########## spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkView.java: ########## @@ -0,0 +1,186 @@ +/* + * 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.iceberg.spark.source; + +import static org.apache.iceberg.TableProperties.FORMAT_VERSION; + +import java.util.Map; +import java.util.Set; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; +import org.apache.iceberg.spark.SparkSchemaUtil; +import org.apache.iceberg.view.BaseView; +import org.apache.iceberg.view.SQLViewRepresentation; +import org.apache.iceberg.view.ViewOperations; +import org.apache.spark.sql.connector.catalog.DependencyList; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.connector.catalog.TableSummary; +import org.apache.spark.sql.connector.catalog.View; + +/** + * Converts Iceberg view metadata to Spark's {@link View} representation. + * + * <p>Keeps this conversion in Iceberg instead of relying on Spark's built-in view handling so + * reserved properties and Iceberg defaults are exposed consistently to Spark commands. + */ +public class SparkView { + + public static final String PROP_CREATE_ENGINE_VERSION = "create_engine_version"; Review Comment: Not blocking, just noting the direction: Spark 4.2 removed `ViewCatalog.PROP_*` and `RESERVED_PROPERTIES` entirely, so these constants and `RESERVED_PROPERTIES` below are now maintained independently of Spark. That is self-consistent given the inspection execs are staying, but it does mean the list needs a deliberate owner, since Spark's `TABLE_RESERVED_PROPERTIES` will keep moving. ########## spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java: ########## @@ -0,0 +1,996 @@ +/* + * 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.iceberg.spark; + +import static org.apache.iceberg.TableProperties.GC_ENABLED; +import static org.apache.iceberg.TableProperties.GC_ENABLED_DEFAULT; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.CachingCatalog; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.EnvironmentContext; +import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.MetadataTableType; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Transaction; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.SupportsNamespaces; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.catalog.ViewCatalog; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.hadoop.HadoopCatalog; +import org.apache.iceberg.hadoop.HadoopTables; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.base.Splitter; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.rest.RESTCatalog; +import org.apache.iceberg.spark.actions.SparkActions; +import org.apache.iceberg.spark.source.SparkChangelogTable; +import org.apache.iceberg.spark.source.SparkTable; +import org.apache.iceberg.spark.source.SparkView; +import org.apache.iceberg.spark.source.StagedSparkTable; +import org.apache.iceberg.util.Pair; +import org.apache.iceberg.util.PropertyUtil; +import org.apache.iceberg.view.ViewBuilder; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.catalyst.analysis.NamespaceAlreadyExistsException; +import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException; +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; +import org.apache.spark.sql.catalyst.analysis.NoSuchViewException; +import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException; +import org.apache.spark.sql.catalyst.analysis.ViewAlreadyExistsException; +import org.apache.spark.sql.catalyst.analysis.ViewUtil; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.NamespaceChange; +import org.apache.spark.sql.connector.catalog.StagedTable; +import org.apache.spark.sql.connector.catalog.Table; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.connector.catalog.TableChange; +import org.apache.spark.sql.connector.catalog.TableChange.ColumnChange; +import org.apache.spark.sql.connector.catalog.TableChange.RemoveProperty; +import org.apache.spark.sql.connector.catalog.TableChange.SetProperty; +import org.apache.spark.sql.connector.catalog.TableSummary; +import org.apache.spark.sql.connector.catalog.View; +import org.apache.spark.sql.connector.expressions.Transform; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A Spark TableCatalog implementation that wraps an Iceberg {@link Catalog}. + * + * <p>This supports the following catalog configuration options: + * + * <ul> + * <li><code>type</code> - catalog type, "hive" or "hadoop" or "rest". To specify a non-hive or + * hadoop catalog, use the <code>catalog-impl</code> option. + * <li><code>uri</code> - the Hive Metastore URI for Hive catalog or REST URI for REST catalog + * <li><code>warehouse</code> - the warehouse path (Hadoop catalog only) + * <li><code>catalog-impl</code> - a custom {@link Catalog} implementation to use + * <li><code>io-impl</code> - a custom {@link org.apache.iceberg.io.FileIO} implementation to use + * <li><code>metrics-reporter-impl</code> - a custom {@link + * org.apache.iceberg.metrics.MetricsReporter} implementation to use + * <li><code>default-namespace</code> - a namespace to use as the default + * <li><code>cache-enabled</code> - whether to enable catalog cache + * <li><code>cache.case-sensitive</code> - whether the catalog cache should compare table + * identifiers in a case sensitive way + * <li><code>cache.expiration-interval-ms</code> - interval in millis before expiring tables from + * catalog cache. Refer to {@link CatalogProperties#CACHE_EXPIRATION_INTERVAL_MS} for further + * details and significant values. + * <li><code>table-default.$tablePropertyKey</code> - table property $tablePropertyKey default at + * catalog level + * <li><code>table-override.$tablePropertyKey</code> - table property $tablePropertyKey enforced + * at catalog level + * </ul> + * + * <p> + */ +public class SparkCatalog extends BaseCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(SparkCatalog.class); + private static final Set<String> DEFAULT_NS_KEYS = ImmutableSet.of(TableCatalog.PROP_OWNER); + private static final Splitter COMMA = Splitter.on(","); + private static final Pattern AT_TIMESTAMP = Pattern.compile("at_timestamp_(\\d+)"); + private static final Pattern SNAPSHOT_ID = Pattern.compile("snapshot_id_(\\d+)"); + private static final Pattern BRANCH = Pattern.compile("branch_(.*)"); + private static final Pattern TAG = Pattern.compile("tag_(.*)"); + + private enum ViewCommit { + CREATE, + REPLACE, + CREATE_OR_REPLACE + } + + private String catalogName = null; + private Catalog icebergCatalog = null; + private SupportsNamespaces asNamespaceCatalog = null; + private ViewCatalog asViewCatalog = null; + private String[] defaultNamespace = null; + private HadoopTables tables; + private boolean restCatalogPurge; + private boolean isRestCatalog; + + /** + * Build an Iceberg {@link Catalog} to be used by this Spark catalog adapter. + * + * @param name Spark's catalog name + * @param options Spark's catalog options + * @return an Iceberg catalog + */ + protected Catalog buildIcebergCatalog(String name, CaseInsensitiveStringMap options) { + Configuration conf = SparkUtil.hadoopConfCatalogOverrides(SparkSession.active(), name); + Map<String, String> optionsMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + optionsMap.putAll(options.asCaseSensitiveMap()); + optionsMap.put(CatalogProperties.APP_ID, SparkSession.active().sparkContext().applicationId()); + optionsMap.put(CatalogProperties.USER, SparkSession.active().sparkContext().sparkUser()); + return CatalogUtil.buildIcebergCatalog(name, optionsMap, conf); + } + + /** + * Build an Iceberg {@link TableIdentifier} for the given Spark identifier. + * + * @param identifier Spark's identifier + * @return an Iceberg identifier + */ + protected TableIdentifier buildIdentifier(Identifier identifier) { + return Spark3Util.identifierToTableIdentifier(identifier); + } + + @Override + public Table loadTable(Identifier ident) throws NoSuchTableException { + return load(ident, null /* no time travel */); + } + + @Override + public Table loadTable(Identifier ident, String version) throws NoSuchTableException { + return load(ident, TimeTravel.version(version)); + } + + @Override + public Table loadTable(Identifier ident, long timestampMicros) throws NoSuchTableException { + return load(ident, TimeTravel.timestampMicros(timestampMicros)); + } + + @Override + public boolean tableExists(Identifier ident) { + if (isPathIdentifier(ident)) { + return tables.exists(((PathIdentifier) ident).location()); + } else { + return icebergCatalog.tableExists(buildIdentifier(ident)); + } + } + + @Override + public Table createTable( + Identifier ident, StructType schema, Transform[] transforms, Map<String, String> properties) + throws TableAlreadyExistsException { + Schema icebergSchema = SparkSchemaUtil.convert(schema); + try { + Catalog.TableBuilder builder = newBuilder(ident, icebergSchema); + org.apache.iceberg.Table icebergTable = + builder + .withPartitionSpec(Spark3Util.toPartitionSpec(icebergSchema, transforms)) + .withLocation(properties.get("location")) + .withProperties(Spark3Util.rebuildCreateProperties(properties)) + .create(); + return new SparkTable(icebergTable); + } catch (AlreadyExistsException e) { + throw new TableAlreadyExistsException(ident); + } + } + + @Override + public StagedTable stageCreate( + Identifier ident, StructType schema, Transform[] transforms, Map<String, String> properties) + throws TableAlreadyExistsException { + Schema icebergSchema = SparkSchemaUtil.convert(schema); + try { + Catalog.TableBuilder builder = newBuilder(ident, icebergSchema); + Transaction transaction = + builder + .withPartitionSpec(Spark3Util.toPartitionSpec(icebergSchema, transforms)) + .withLocation(properties.get("location")) + .withProperties(Spark3Util.rebuildCreateProperties(properties)) + .createTransaction(); + return new StagedSparkTable(transaction); + } catch (AlreadyExistsException e) { + throw new TableAlreadyExistsException(ident); + } + } + + @Override + public StagedTable stageReplace( + Identifier ident, StructType schema, Transform[] transforms, Map<String, String> properties) + throws NoSuchTableException { + Schema icebergSchema = SparkSchemaUtil.convert(schema); + try { + Catalog.TableBuilder builder = newBuilder(ident, icebergSchema); + Transaction transaction = + builder + .withPartitionSpec(Spark3Util.toPartitionSpec(icebergSchema, transforms)) + .withLocation(properties.get("location")) + .withProperties(Spark3Util.rebuildCreateProperties(properties)) + .replaceTransaction(); + return new StagedSparkTable(transaction); + } catch (org.apache.iceberg.exceptions.NoSuchTableException e) { + throw new NoSuchTableException(ident); + } + } + + @Override + public StagedTable stageCreateOrReplace( + Identifier ident, StructType schema, Transform[] transforms, Map<String, String> properties) { + Schema icebergSchema = SparkSchemaUtil.convert(schema); + Catalog.TableBuilder builder = newBuilder(ident, icebergSchema); + Transaction transaction = + builder + .withPartitionSpec(Spark3Util.toPartitionSpec(icebergSchema, transforms)) + .withLocation(properties.get("location")) + .withProperties(Spark3Util.rebuildCreateProperties(properties)) + .createOrReplaceTransaction(); + return new StagedSparkTable(transaction); + } + + @Override + public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchTableException { + SetProperty setLocation = null; + SetProperty setSnapshotId = null; + SetProperty pickSnapshotId = null; + List<TableChange> propertyChanges = Lists.newArrayList(); + List<TableChange> schemaChanges = Lists.newArrayList(); + + for (TableChange change : changes) { + if (change instanceof SetProperty) { + SetProperty set = (SetProperty) change; + if (TableCatalog.PROP_LOCATION.equalsIgnoreCase(set.property())) { + setLocation = set; + } else if ("current-snapshot-id".equalsIgnoreCase(set.property())) { + setSnapshotId = set; + } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(set.property())) { + pickSnapshotId = set; + } else if ("sort-order".equalsIgnoreCase(set.property())) { + throw new UnsupportedOperationException( + "Cannot specify the 'sort-order' because it's a reserved table " + + "property. Please use the command 'ALTER TABLE ... WRITE ORDERED BY' to specify write sort-orders."); + } else if ("identifier-fields".equalsIgnoreCase(set.property())) { + throw new UnsupportedOperationException( + "Cannot specify the 'identifier-fields' because it's a reserved table property. " + + "Please use the command 'ALTER TABLE ... SET IDENTIFIER FIELDS' to specify identifier fields."); + } else { + propertyChanges.add(set); + } + } else if (change instanceof RemoveProperty) { + propertyChanges.add(change); + } else if (change instanceof ColumnChange) { + schemaChanges.add(change); + } else { + throw new UnsupportedOperationException("Cannot apply unknown table change: " + change); + } + } + + try { + org.apache.iceberg.Table table = icebergCatalog.loadTable(buildIdentifier(ident)); + commitChanges( + table, setLocation, setSnapshotId, pickSnapshotId, propertyChanges, schemaChanges); + return new SparkTable(table); + } catch (org.apache.iceberg.exceptions.NoSuchTableException e) { + throw new NoSuchTableException(ident); + } + } + + @Override + public boolean dropTable(Identifier ident) { + return catalogDropTable(ident); + } + + @Override + public boolean purgeTable(Identifier ident) { + try { + org.apache.iceberg.Table table = icebergCatalog.loadTable(buildIdentifier(ident)); + ValidationException.check( + PropertyUtil.propertyAsBoolean(table.properties(), GC_ENABLED, GC_ENABLED_DEFAULT), + "Cannot purge table: GC is disabled (deleting files may corrupt other tables)"); + String metadataFileLocation = + ((HasTableOperations) table).operations().current().metadataFileLocation(); + + if (isRestCatalog && !isPathIdentifier(ident)) { + if (restCatalogPurge) { + return icebergCatalog.dropTable(buildIdentifier(ident), true); + } else { + LOG.info( + "Set '{}' to true to use the REST catalog's capabilities to purge the table.", + SparkCatalogProperties.REST_CATALOG_PURGE); + } + } + + boolean dropped = catalogDropTable(ident); + + if (dropped) { + // check whether the metadata file exists because HadoopCatalog/HadoopTables + // will drop the warehouse directly and ignore the `purge` argument + boolean metadataFileExists = table.io().newInputFile(metadataFileLocation).exists(); + + if (metadataFileExists) { + SparkActions.get().deleteReachableFiles(metadataFileLocation).io(table.io()).execute(); + } + } + + return dropped; + } catch (org.apache.iceberg.exceptions.NoSuchTableException e) { + return false; + } + } + + private boolean catalogDropTable(Identifier ident) { + if (isPathIdentifier(ident)) { + return tables.dropTable(((PathIdentifier) ident).location(), false /* don't purge data */); + } else { + return icebergCatalog.dropTable(buildIdentifier(ident), false /* don't purge data */); + } + } + + @Override + public void renameTable(Identifier from, Identifier to) + throws NoSuchTableException, TableAlreadyExistsException { + try { + checkNotPathIdentifier(from, "renameTable"); + checkNotPathIdentifier(to, "renameTable"); + icebergCatalog.renameTable(buildIdentifier(from), buildIdentifier(to)); + } catch (org.apache.iceberg.exceptions.NoSuchTableException e) { + throw new NoSuchTableException(from); + } catch (AlreadyExistsException e) { + throw new TableAlreadyExistsException(to); + } + } + + @Override + public void invalidateTable(Identifier ident) { + if (!isPathIdentifier(ident)) { + icebergCatalog.invalidateTable(buildIdentifier(ident)); + } + } + + @Override + public Identifier[] listTables(String[] namespace) { + return icebergCatalog.listTables(Namespace.of(namespace)).stream() + .map(ident -> Identifier.of(ident.namespace().levels(), ident.name())) + .toArray(Identifier[]::new); + } + + @Override + public TableSummary[] listTableSummaries(String[] namespace) { + // Build summaries directly from the catalog listings to avoid loading every table, which the + // default TableCatalog.listTableSummaries implementation would do. Iceberg tables are always + // reported as EXTERNAL (see BaseSparkTable#properties). + List<TableSummary> summaries = Lists.newArrayList(); + + // Most catalogs return only tables from listTables, but HiveCatalog with list-all-tables=true + // returns every metastore entry, including views. RelationCatalog requires this method to + // return tables only; listRelationSummaries adds views to this result. + Set<Identifier> viewIdents = Sets.newHashSet(listViews(namespace)); + + for (Identifier ident : listTables(namespace)) { + if (!viewIdents.contains(ident)) { + summaries.add(TableSummary.of(ident, TableSummary.EXTERNAL_TABLE_TYPE)); + } + } + + return summaries.toArray(new TableSummary[0]); + } + + @Override + public String[] defaultNamespace() { + if (defaultNamespace != null) { + return defaultNamespace; + } + + return new String[0]; + } + + @Override + public String[][] listNamespaces() { + if (asNamespaceCatalog != null) { + return asNamespaceCatalog.listNamespaces().stream() + .map(Namespace::levels) + .toArray(String[][]::new); + } + + return new String[0][]; + } + + @Override + public String[][] listNamespaces(String[] namespace) throws NoSuchNamespaceException { + if (asNamespaceCatalog != null) { + try { + return asNamespaceCatalog.listNamespaces(Namespace.of(namespace)).stream() + .map(Namespace::levels) + .toArray(String[][]::new); + } catch (org.apache.iceberg.exceptions.NoSuchNamespaceException e) { + throw new NoSuchNamespaceException(namespace); + } + } + + throw new NoSuchNamespaceException(namespace); + } + + @Override + public boolean namespaceExists(String[] namespace) { + return asNamespaceCatalog != null + && asNamespaceCatalog.namespaceExists(Namespace.of(namespace)); + } + + @Override + public Map<String, String> loadNamespaceMetadata(String[] namespace) + throws NoSuchNamespaceException { + if (asNamespaceCatalog != null) { + try { + return asNamespaceCatalog.loadNamespaceMetadata(Namespace.of(namespace)); + } catch (org.apache.iceberg.exceptions.NoSuchNamespaceException e) { + throw new NoSuchNamespaceException(namespace); + } + } + + throw new NoSuchNamespaceException(namespace); + } + + @Override + public void createNamespace(String[] namespace, Map<String, String> metadata) + throws NamespaceAlreadyExistsException { + if (asNamespaceCatalog != null) { + try { + if (asNamespaceCatalog instanceof HadoopCatalog + && DEFAULT_NS_KEYS.equals(metadata.keySet())) { + // Hadoop catalog will reject metadata properties, but Spark automatically adds "owner". + // If only the automatic properties are present, replace metadata with an empty map. + asNamespaceCatalog.createNamespace(Namespace.of(namespace), ImmutableMap.of()); + } else { + asNamespaceCatalog.createNamespace(Namespace.of(namespace), metadata); + } + } catch (AlreadyExistsException e) { + throw new NamespaceAlreadyExistsException(namespace); + } + } else { + throw new UnsupportedOperationException( + "Namespaces are not supported by catalog: " + catalogName); + } + } + + @Override + public void alterNamespace(String[] namespace, NamespaceChange... changes) + throws NoSuchNamespaceException { + if (asNamespaceCatalog != null) { + Map<String, String> updates = Maps.newHashMap(); + Set<String> removals = Sets.newHashSet(); + for (NamespaceChange change : changes) { + if (change instanceof NamespaceChange.SetProperty) { + NamespaceChange.SetProperty set = (NamespaceChange.SetProperty) change; + updates.put(set.property(), set.value()); + } else if (change instanceof NamespaceChange.RemoveProperty) { + removals.add(((NamespaceChange.RemoveProperty) change).property()); + } else { + throw new UnsupportedOperationException( + "Cannot apply unknown namespace change: " + change); + } + } + + try { + if (!updates.isEmpty()) { + asNamespaceCatalog.setProperties(Namespace.of(namespace), updates); + } + + if (!removals.isEmpty()) { + asNamespaceCatalog.removeProperties(Namespace.of(namespace), removals); + } + + } catch (org.apache.iceberg.exceptions.NoSuchNamespaceException e) { + throw new NoSuchNamespaceException(namespace); + } + } else { + throw new NoSuchNamespaceException(namespace); + } + } + + @Override + public boolean dropNamespace(String[] namespace, boolean cascade) + throws NoSuchNamespaceException { + if (asNamespaceCatalog != null) { + try { + return asNamespaceCatalog.dropNamespace(Namespace.of(namespace)); + } catch (org.apache.iceberg.exceptions.NoSuchNamespaceException e) { + throw new NoSuchNamespaceException(namespace); + } + } + + return false; + } + + @Override + public Identifier[] listViews(String... namespace) { + if (null != asViewCatalog) { + return asViewCatalog.listViews(Namespace.of(namespace)).stream() + .map(ident -> Identifier.of(ident.namespace().levels(), ident.name())) + .toArray(Identifier[]::new); + } + + return new Identifier[0]; + } + + @Override + public boolean viewExists(Identifier ident) { + return asViewCatalog != null && asViewCatalog.viewExists(buildIdentifier(ident)); + } + + @Override + public View loadView(Identifier ident) throws NoSuchViewException { + if (null != asViewCatalog) { + try { + org.apache.iceberg.view.View icebergView = asViewCatalog.loadView(buildIdentifier(ident)); + return SparkView.toView(catalogName, icebergView); + } catch (org.apache.iceberg.exceptions.NoSuchViewException e) { + throw new NoSuchViewException(ident); + } + } + + throw new NoSuchViewException(ident); + } + + @Override + public View createView(Identifier ident, View view) + throws ViewAlreadyExistsException, NoSuchNamespaceException { + try { + return commitView(ident, view, ViewCommit.CREATE); + } catch (NoSuchViewException e) { + throw unexpectedViewCommitException("create", ident, "reported that the view is missing", e); + } + } + + @Override + public View replaceView(Identifier ident, View view) throws NoSuchViewException { + try { + return commitView(ident, view, ViewCommit.REPLACE); + } catch (NoSuchNamespaceException e) { + throw unexpectedViewCommitException( + "replace", ident, "reported that the namespace is missing", e); + } catch (ViewAlreadyExistsException e) { + throw unexpectedViewCommitException( + "replace", ident, "reported that the view already exists", e); + } + } + + @Override + public View createOrReplaceView(Identifier ident, View view) + throws ViewAlreadyExistsException, NoSuchNamespaceException { + try { + return commitView(ident, view, ViewCommit.CREATE_OR_REPLACE); + } catch (NoSuchViewException e) { + throw unexpectedViewCommitException( + "create or replace", ident, "reported that the view is missing", e); + } + } + + private static RuntimeException unexpectedViewCommitException( + String operation, Identifier ident, String failure, Exception cause) { + return new IllegalStateException( + String.format( + "Cannot %s view %s because the underlying catalog %s", operation, ident, failure), + cause); + } + + private View commitView(Identifier ident, View view, ViewCommit viewCommit) + throws ViewAlreadyExistsException, NoSuchNamespaceException, NoSuchViewException { + Preconditions.checkArgument(view != null, "Invalid view metadata: null"); + + if (null == asViewCatalog) { + if (viewCommit == ViewCommit.REPLACE && tableExists(ident)) { + throw new NoSuchViewException(ident); + } + + throw new UnsupportedOperationException( + "View operations are not supported by catalog: " + catalogName); + } + + View normalizedView = normalizeViewCurrentCatalog(catalogName, view); + String[] currentNamespace = normalizedView.currentNamespace(); + Map<String, String> properties = normalizedView.properties(); + Schema icebergSchema = SparkSchemaUtil.convert(normalizedView.schema()); + Map<String, String> props = ViewUtil.createProperties(normalizedView); + + try { + ViewBuilder builder = + asViewCatalog + .buildView(buildIdentifier(ident)) + .withDefaultCatalog(normalizedView.currentCatalog()) + .withDefaultNamespace(Namespace.of(currentNamespace)) + .withQuery("spark", normalizedView.queryText()) + .withSchema(icebergSchema) + .withLocation(properties.get(TableCatalog.PROP_LOCATION)) + .withProperties(props); Review Comment: `ViewMetadata.Builder.setProperties` does `properties.putAll(updated)`, so properties absent from the incoming `View` survive rather than being removed. That is reachable on paths Iceberg does not control: `CREATE OR REPLACE VIEW` and Spark's `ALTER VIEW ... AS SELECT` both land in `replaceView`, so a redefinition that drops a `TBLPROPERTIES` entry silently keeps it. Diffing the incoming properties against the existing view and issuing `UpdateViewProperties.remove` for the disappeared keys would cover it. ########## spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/IcebergShowV2ViewPropertiesExec.scala: ########## @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.spark.sql.execution.datasources.v2 + +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.analysis.ViewUtil +import org.apache.spark.sql.catalyst.expressions.Attribute +import org.apache.spark.sql.connector.catalog.Identifier +import org.apache.spark.sql.connector.catalog.View +import org.apache.spark.sql.execution.LeafExecNode +import scala.jdk.CollectionConverters._ + +/** + * Executes SHOW TBLPROPERTIES for Spark V2 views. + * + * Uses a custom command instead of Spark's built-in implementation so Iceberg reserved metadata is + * filtered from user-visible view properties. + */ +case class IcebergShowV2ViewPropertiesExec( + output: Seq[Attribute], + catalogName: String, + ident: Identifier, + view: View, + propertyKey: Option[String]) + extends V2CommandExec + with LeafExecNode { + + import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._ + + override protected def run(): Seq[InternalRow] = { + val redactedProperties = conf.redactOptions(properties) + propertyKey match { + case Some(p) => + val propValue = + redactedProperties + .getOrElse(p, s"View ${catalogName}.${ident.quoted} does not have property: $p") Review Comment: `ident` is quoted but `catalogName` is not. Spark's equivalent builds the whole name as `(catalog.name() +: ident.asMultipartIdentifier).map(quoteIfNeeded).mkString(".")`, so a catalog name that needs quoting renders differently here. ########## spark/v4.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckViews.scala: ########## @@ -28,39 +28,27 @@ import org.apache.spark.sql.catalyst.plans.logical.View import org.apache.spark.sql.catalyst.plans.logical.views.CreateIcebergView import org.apache.spark.sql.catalyst.plans.logical.views.ResolvedV2View import org.apache.spark.sql.connector.catalog.ViewCatalog -import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.util.SchemaUtils object CheckViews extends (LogicalPlan => Unit) { import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._ override def apply(plan: LogicalPlan): Unit = { plan foreach { - case CreateIcebergView( - resolvedIdent @ ResolvedIdentifier(_: ViewCatalog, _), - _, - query, - columnAliases, - _, - _, - _, - _, - _, - replace, - _, - _) => - verifyColumnCount(resolvedIdent, columnAliases, query) - SchemaUtils.checkColumnNameDuplication( - query.schema.fieldNames.toIndexedSeq, - SQLConf.get.resolver) - if (replace) { - val viewIdent: Seq[String] = - resolvedIdent.catalog.name() +: resolvedIdent.identifier.asMultipartIdentifier - checkCyclicViewReference(viewIdent, query, Seq(viewIdent)) + case c: CreateIcebergView => Review Comment: For create, Spark is mostly ready — the validation you are asking about exists natively as `CheckViewReferences`, and this PR already delegates the create itself to Spark's `CreateV2ViewExec`. The catch is that `CheckViewReferences` only fires on Spark's own `CreateView` plan, which `RewriteViewCommands` replaces, and I do not think we can stop replacing it while `spark_catalog` needs to work: `ResolveSessionCatalog` routes view DDL by catalog name (`ResolvedViewIdentifier` matches only when `isSessionCatalog(catalog)`), so a `RelationCatalog` installed as `spark_catalog` lands on the V1 path rather than reaching the v2 view execs. I traced that statically rather than running it, so worth confirming. If it holds, `CheckViews` has to stay until Spark routes by capability, and that seems worth a JIRA. The ALTER property execs look like a firmer no, though. Spark expresses a property change as a full `replaceView` — `viewInfoBuilderFrom(existing).withProperties(merged).build()` — because `ViewCatalog` has no partial-update operation, and that is lossy against Iceberg's model in two ways. `UNSET` becomes a silent no-op, since `ViewMetadata.Builder.setProperties` does `putAll` and omitted keys survive. And the rebuild only supplies `withQuery("spark", ...)`, while `ViewBuilder.replace()` writes the new version with `addAllRepresentations(representations)`, so setting an unrelated property on a view that also carries a Trino or Flink representation discards it. Spark-only views hit neither, which is why the tests are green. `ALTER VIEW ... AS SELECT` is fine to delegate — the query genuinely changed there, so a new version is correct. It is specifically the property-only path that needs `updateProperties()`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
