[SPARK-58936][SQL] Add partial view updates to ViewCatalog

common/utils/src/main/resources/error/error-conditions.json

@@ -9545,6 +9545,12 @@
     ],
     "sqlState" : "0A000"
   },
+  "UNSUPPORTED_VIEW_CHANGE" : {
+    "message" : [
+      "Unsupported view change: <message>"
+    ],
+    "sqlState" : "0A000"
+  },
   "UNTYPED_SCALA_UDF" : {
     "message" : [
       "You're using untyped Scala UDF, which does not have the input type information. Spark may blindly pass null to the Scala closure with primitive-type argument, and the closure will see the default value of the Java type for the null argument, e.g. `udf((x: Int) => x, IntegerType)`, the result is 0 for null input. To get rid of this error, you could:",

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

@@ -16,6 +16,9 @@
  */
 package org.apache.spark.sql.connector.catalog;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import org.apache.spark.annotation.Evolving;
 import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
 import org.apache.spark.sql.catalyst.analysis.NoSuchViewException;
@@ -110,6 +113,52 @@ View createView(Identifier ident, View info)
    */
   View replaceView(Identifier ident, View info) throws NoSuchViewException;
 
+  /**
+   * Apply a set of {@link ViewChange changes} to a view in the catalog.
+   * <p>
+   * Implementations may reject the requested changes. If any change is rejected, none of the
+   * changes should be applied to the view. The requested changes must be applied in order.
+   * <p>
+   * The default implementation loads the latest view metadata, applies the changes, and calls
+   * {@link #replaceView}. Catalogs should override this method if they can apply the changes in a
+   * single atomic operation.
+   *
+   * @param ident a view identifier
+   * @param changes changes to apply to the view
+   * @return updated metadata for the view
+   * @throws NoSuchViewException if the view does not exist
+   * @throws IllegalArgumentException if any change is rejected by the implementation
+   *
+   * @since 5.0.0
+   */
+  default View alterView(Identifier ident, ViewChange... changes) throws NoSuchViewException {
+    invalidateView(ident);
+    View current = loadView(ident);
+    Map<String, String> properties = new HashMap<>(current.properties());
+    for (ViewChange change : changes) {
+      if (change instanceof ViewChange.SetProperty set) {
+        properties.put(set.property(), set.value());
+      } else if (change instanceof ViewChange.RemoveProperty remove) {
+        properties.remove(remove.property());
+      } else {
+        throw new IllegalArgumentException("Unsupported view change: " + change);
+      }
+    }
+
+    View updated = new View.Builder()
+        .withColumns(current.columns())
+        .withProperties(properties)
+        .withQueryText(current.queryText())
+        .withCurrentCatalog(current.currentCatalog())
+        .withCurrentNamespace(current.currentNamespace())
+        .withSqlConfigs(current.sqlConfigs())
+        .withSchemaMode(current.schemaMode())
+        .withQueryColumnNames(current.queryColumnNames())
+        .withViewDependencies(current.viewDependencies())
+        .build();
+    return replaceView(ident, updated);
+  }
+
   /**
    * Create a view if one does not exist at {@code ident}, or atomically replace it if one does.
    * <p>

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

@@ -0,0 +1,112 @@
+/*
+ * 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;
+
+/**
+ * ViewChange subclasses represent requested changes to a view. These are passed to
+ * {@link ViewCatalog#alterView}.
+ *
+ * @since 5.0.0
+ */
+@Evolving
+public interface ViewChange {
+
+  /**
+   * Create a ViewChange for setting a view property.
+   *
+   * @param property the property name
+   * @param value the new property value
+   * @return a ViewChange for setting the property
+   */
+  static ViewChange setProperty(String property, String value) {
+    return new SetProperty(property, value);
+  }
+
+  /**
+   * Create a ViewChange for removing a view property.
+   * <p>
+   * If the property does not exist, the change will succeed.
+   *
+   * @param property the property name
+   * @return a ViewChange for removing the property
+   */
+  static ViewChange removeProperty(String property) {
+    return new RemoveProperty(property);
+  }
+
+  /** A ViewChange to set a view property. */
+  final class SetProperty implements ViewChange {
+    private final String property;
+    private final String value;
+
+    private SetProperty(String property, String value) {
+      this.property = property;
+      this.value = value;
+    }
+
+    public String property() { return property; }
+
+    public String value() { return value; }
+
+    @Override
+    public String toString() { return "SET PROPERTY " + property + " = " + value; }
+
+    @Override
+    public boolean equals(Object other) {
+      if (this == other) return true;
+      if (other == null || getClass() != other.getClass()) return false;
+      SetProperty that = (SetProperty) other;
+      return property.equals(that.property) && value.equals(that.value);
+    }
+
+    @Override
+    public int hashCode() { return Objects.hash(property, value); }
+  }
+
+  /**
+   * A ViewChange to remove a view property.
+   * <p>
+   * If the property does not exist, the change should succeed.
+   */
+  final class RemoveProperty implements ViewChange {
+    private final String property;
+
+    private RemoveProperty(String property) {
+      this.property = property;
+    }
+
+    public String property() { return property; }
+
+    @Override
+    public String toString() { return "REMOVE PROPERTY " + property; }
+
+    @Override
+    public boolean equals(Object other) {
+      if (this == other) return true;
+      if (other == null || getClass() != other.getClass()) return false;
+      RemoveProperty that = (RemoveProperty) other;
+      return property.equals(that.property);
+    }
+
+    @Override
+    public int hashCode() { return Objects.hash(property); }
+  }
+}

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

@@ -577,11 +577,10 @@ private[sql] object CatalogV2Util {
   }
 
   /**
-   * Construct a [[View.Builder]] seeded from an existing view's metadata. Used by ALTER
-   * VIEW execs (SET / UNSET TBLPROPERTIES, ALTER VIEW ... WITH SCHEMA BINDING) -- override
-   * the one field that changes, then `build` to produce the replacement payload for
-   * [[ViewCatalog#replaceView]]. Every other field flows through unchanged so a metadata-only
-   * mutation does not perturb the view body.
+   * Construct a [[View.Builder]] seeded from an existing view's metadata. Used by the ALTER
+   * VIEW ... WITH SCHEMA BINDING exec -- override the one field that changes, then `build` to
+   * produce the replacement payload for [[ViewCatalog#replaceView]]. Every other field flows
+   * through unchanged so a metadata-only mutation does not perturb the view body.
    */
   def viewInfoBuilderFrom(existing: View): View.Builder = {
     val builder = new View.Builder()

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala

@@ -739,6 +739,13 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
       cause = e)
   }
 
+  def unsupportedViewChangeError(e: IllegalArgumentException): Throwable = {
+    new SparkException(
+      errorClass = "UNSUPPORTED_VIEW_CHANGE",
+      messageParameters = Map("message" -> e.getMessage),
+      cause = e)
+  }
+
   def notADatasourceRDDPartitionError(split: Partition): Throwable = {
     new SparkException(
       errorClass = "_LEGACY_ERROR_TEMP_2046",

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRelationCatalog.scala

@@ -41,6 +41,8 @@ class InMemoryRelationCatalog extends RelationCatalog with SupportsNamespaces {
     new ConcurrentHashMap[(Seq[String], String), Relation]()
   private val namespaces =
     new ConcurrentHashMap[Seq[String], util.Map[String, String]]()
+  @volatile private var lastViewChanges: Seq[ViewChange] = Seq.empty
+  @volatile private var alterViewFailure: IllegalArgumentException = _
 
   override def loadRelation(ident: Identifier): Relation = {
     val key = (ident.namespace().toSeq, ident.name())
@@ -124,6 +126,14 @@ class InMemoryRelationCatalog extends RelationCatalog with SupportsNamespaces {
     info
   }
 
+  override def alterView(ident: Identifier, changes: ViewChange*): View = {
+    lastViewChanges = changes
+    if (alterViewFailure != null) {
+      throw alterViewFailure
+    }
+    super.alterView(ident, changes: _*)
+  }
+
   override def dropView(ident: Identifier): Boolean = {
     val key = (ident.namespace().toSeq, ident.name())
     val existing = store.get(key)
@@ -234,6 +244,19 @@ class InMemoryRelationCatalog extends RelationCatalog with SupportsNamespaces {
     }
   }
 
+  /** Returns the changes from the most recent alterView call. */
+  def getLastViewChanges: Seq[ViewChange] = lastViewChanges
+
+  /** Configures a failure for alterView. */
+  def failAlterViewWith(failure: IllegalArgumentException): Unit = {
+    alterViewFailure = failure
+  }
+
+  /** Clears a configured alterView failure. */
+  def clearAlterViewFailure(): Unit = {
+    alterViewFailure = null
+  }
+
   // CatalogPlugin --------------------------------------------------------------------
 
   private var catalogName: String = ""

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/ViewCatalogSuite.scala

@@ -0,0 +1,105 @@
+/*
+ * 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 scala.jdk.CollectionConverters._
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.types.IntegerType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+class ViewCatalogSuite extends SparkFunSuite {
+
+  test("default alterView applies changes in order and preserves view metadata") {
+    val catalog = new InMemoryRelationCatalog
+    catalog.initialize("test", CaseInsensitiveStringMap.empty())
+    val ident = Identifier.of(Array("ns"), "view")
+    val dependencies = DependencyList.of(
+      Array(Dependency.table(Array("source_catalog", "source_ns", "source"))))
+    val original = new View.Builder()
+      .withColumns(Array(Column.create("id", IntegerType)))
+      .withProperties(Map("first" -> "old", "second" -> "remove").asJava)
+      .withQueryText("SELECT id FROM source_catalog.source_ns.source")
+      .withCurrentCatalog("source_catalog")
+      .withCurrentNamespace(Array("source_ns"))
+      .withSqlConfigs(Map("spark.sql.ansi.enabled" -> "true").asJava)
+      .withSchemaMode("BINDING")
+      .withQueryColumnNames(Array("id"))
+      .withViewDependencies(dependencies)
+      .build()
+    catalog.createView(ident, original)
+
+    val updated = catalog.alterView(
+      ident,
+      ViewChange.setProperty("first", "intermediate"),
+      ViewChange.removeProperty("first"),
+      ViewChange.setProperty("first", "new"),
+      ViewChange.removeProperty("second"))
+
+    assert(updated.properties.get("first") === "new")
+    assert(!updated.properties.containsKey("second"))
+    assert(updated.columns.sameElements(original.columns))
+    assert(updated.queryText === original.queryText)
+    assert(updated.currentCatalog === original.currentCatalog)
+    assert(updated.currentNamespace.sameElements(original.currentNamespace))
+    assert(updated.sqlConfigs === original.sqlConfigs)
+    assert(updated.schemaMode === original.schemaMode)
+    assert(updated.queryColumnNames.sameElements(original.queryColumnNames))
+    assert(updated.viewDependencies === original.viewDependencies)
+  }
+
+  test("default alterView invalidates cached metadata before loading the current view") {
+    var invalidations = 0
+    val catalog = new InMemoryRelationCatalog {
+      private var cachedView: View = _
+
+      override def loadView(ident: Identifier): View = {
+        if (cachedView == null) {
+          cachedView = super.loadView(ident)
+        }
+        cachedView
+      }
+
+      override def invalidateView(ident: Identifier): Unit = {
+        invalidations += 1
+        cachedView = null
+      }
+    }
+    catalog.initialize("test", CaseInsensitiveStringMap.empty())
+    val ident = Identifier.of(Array("ns"), "view")
+    val original = new View.Builder()
+      .withColumns(Array(Column.create("id", IntegerType)))
+      .withProperties(Map("original" -> "value").asJava)
+      .withQueryText("SELECT 1 AS id")
+      .build()
+    catalog.createView(ident, original)
+    catalog.loadView(ident)
+
+    val concurrent = CatalogV2Util.viewInfoBuilderFrom(original)
+      .withProperties(Map("concurrent" -> "value").asJava)
+      .withQueryText("SELECT 2 AS id")
+      .build()
+    catalog.replaceView(ident, concurrent)
+
+    val updated = catalog.alterView(ident, ViewChange.setProperty("new", "value"))
+
+    assert(invalidations === 1)
+    assert(updated.queryText === concurrent.queryText)
+    assert(updated.properties.asScala === Map("concurrent" -> "value", "new" -> "value"))
+  }
+}