[SPARK-52417][SQL] Simplify Table properties handling in View Schema Evolution Mode

### What changes were proposed in this pull request?

When a View is created, ex CREATE VIEW v (a1 INT, a2 STRING) AS select c1, c2, it needs to save both set of columns (user-specified view schema and query output).

1. User-specified View schema are saved as View Schema (a1 INT, b2 STRING).
2. View query output is saved as Table property w/index (c1, 0) (c2, 1)

In the new Schema Evolution mode, we never allow user-specified view schema, so view schema == view query output schema.  Every time we detect the output view schema changes, we sync the view's schema with view query schema, keeping the invariant.

So we can simplify the logic for View in Schema Evolution mode to not update the Table Properties, and instead rely on the View Schema all the time.

### Why are the changes needed?
View Schema Evolution is a useful mode.  However, it requires a lot of permissions on the user querying the view, because that user needs to update the View definition.  It will simplify auth if we do not have to update table properties too, and reduce the update to only the schema.

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Existing unit test

### Was this patch authored or co-authored using generative AI tooling?
No

Closes #51107 from szehon-ho/SPARK-524167.

Authored-by: Szehon Ho <[email protected]>
Signed-off-by: Gengliang Wang <[email protected]>

Author: szehon-ho

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala

@@ -629,9 +629,15 @@ case class CatalogTable(
       if (lastAccessTime <= 0) JString("UNKNOWN")
       else JLong(lastAccessTime)
 
-    val viewQueryOutputColumns: JValue =
-      if (viewQueryColumnNames.nonEmpty) JArray(viewQueryColumnNames.map(JString).toList)
-      else JNull
+    val viewQueryOutputColumns: JValue = {
+      if (viewSchemaMode == SchemaEvolution) {
+        JArray(schema.map(_.name).map(JString).toList)
+      } else if (viewQueryColumnNames.nonEmpty) {
+        JArray(viewQueryColumnNames.map(JString).toList)
+      } else {
+        JNull
+      }
+    }
 
     val map = mutable.LinkedHashMap[String, JValue]()
 

sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala

@@ -586,13 +586,21 @@ object ViewHelper extends SQLConfHelper with Logging {
     // names.
     SchemaUtils.checkColumnNameDuplication(fieldNames.toImmutableArraySeq, conf.resolver)
 
+    val queryColumnNameProps = if (viewSchemaMode == SchemaEvolution) {
+      // If the view schema mode is SCHEMA EVOLUTION, we can avoid generating the query output
+      // column names as table properties, and always use view schema as they are always same
+      Seq()
+    } else {
+      generateQueryColumnNames(queryOutput.toImmutableArraySeq)
+    }
+
     // Generate the view default catalog and namespace, as well as captured SQL configs.
     val manager = session.sessionState.catalogManager
     removeReferredTempNames(removeSQLConfigs(removeQueryColumnNames(properties))) ++
       catalogAndNamespaceToProps(
         manager.currentCatalog.name, manager.currentNamespace.toImmutableArraySeq) ++
       sqlConfigsToProps(conf) ++
-      generateQueryColumnNames(queryOutput.toImmutableArraySeq) ++
+      queryColumnNameProps ++
       referredTempNamesToProps(tempViewNames, tempFunctionNames, tempVariableNames) ++
       viewSchemaModeToProps(viewSchemaMode)
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala

@@ -35,7 +35,6 @@ import org.apache.spark.sql.classic.SparkSession
 import org.apache.spark.sql.connector.expressions.{FieldReference, RewritableTransform}
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.execution.command.DDLUtils
-import org.apache.spark.sql.execution.command.ViewHelper.generateViewProperties
 import org.apache.spark.sql.execution.datasources.{CreateTable => CreateTableV1}
 import org.apache.spark.sql.execution.datasources.v2.FileDataSourceV2
 import org.apache.spark.sql.internal.SQLConf
@@ -705,16 +704,6 @@ object ViewSyncSchemaToMetaStore extends (LogicalPlan => Unit) {
         }
 
         if (redo) {
-          val newProperties = if (viewSchemaMode == SchemaEvolution) {
-            generateViewProperties(
-              metaData.properties,
-              session,
-              fieldNames,
-              fieldNames,
-              metaData.viewSchemaMode)
-          } else {
-            metaData.properties
-          }
           val newSchema = if (viewSchemaMode == SchemaTypeEvolution) {
             val newFields = viewQuery.schema.map {
               case StructField(name, dataType, nullable, _) =>
@@ -727,9 +716,7 @@ object ViewSyncSchemaToMetaStore extends (LogicalPlan => Unit) {
           }
           SchemaUtils.checkColumnNameDuplication(fieldNames.toImmutableArraySeq,
             session.sessionState.conf.resolver)
-          val updatedViewMeta = metaData.copy(
-            properties = newProperties,
-            schema = newSchema)
+          val updatedViewMeta = metaData.copy(schema = newSchema)
           session.sessionState.catalog.alterTable(updatedViewMeta)
         }
       case _ => // OK

sql/core/src/test/scala/org/apache/spark/sql/execution/SQLViewTestSuite.scala

@@ -22,6 +22,7 @@ import scala.jdk.CollectionConverters._
 import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest, Row}
 import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier}
 import org.apache.spark.sql.catalyst.catalog.CatalogFunction
+import org.apache.spark.sql.catalyst.catalog.CatalogTable._
 import org.apache.spark.sql.catalyst.expressions.Expression
 import org.apache.spark.sql.catalyst.plans.logical.Repartition
 import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.withDefaultTimeZone
@@ -827,6 +828,32 @@ class PersistedViewTestSuite extends SQLViewTestSuite with SharedSparkSession {
     )
   }
 
+  test("SPARK-52417: Simplify Table properties handling in View Schema Evolution Mode") {
+    withTable("t") {
+      withView("v") {
+        sql("CREATE TABLE t (c1 int)")
+        sql("CREATE VIEW v WITH SCHEMA EVOLUTION AS SELECT * from t")
+        sql("INSERT INTO t VALUES (1), (2), (3)")
+        checkAnswer(sql("SELECT * FROM v"), Seq(Row(1), Row(2), Row(3)))
+
+        val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("v"))
+        assert(table.properties.get(VIEW_SCHEMA_MODE) === Some("EVOLUTION"))
+        assert(!table.properties.exists(_._1.startsWith(VIEW_QUERY_OUTPUT_PREFIX)))
+        assert(!table.properties.exists(_._1.startsWith(VIEW_QUERY_OUTPUT_NUM_COLUMNS)))
+
+        sql("DROP TABLE t")
+        sql("CREATE TABLE t (s1 string, b1 boolean)")
+        sql("INSERT INTO t VALUES ('a', true), ('b', false), ('c', true)")
+        checkAnswer(sql("SELECT * FROM v"), Seq(Row("a", true), Row("b", false), Row("c", true)))
+
+        val updatedTable = spark.sessionState.catalog.getTableMetadata(TableIdentifier("v"))
+        assert(updatedTable.properties.get(VIEW_SCHEMA_MODE) === Some("EVOLUTION"))
+        assert(!updatedTable.properties.exists(_._1.startsWith(VIEW_QUERY_OUTPUT_PREFIX)))
+        assert(!updatedTable.properties.exists(_._1.startsWith(VIEW_QUERY_OUTPUT_NUM_COLUMNS)))
+      }
+    }
+  }
+
   def getShowCreateDDL(view: String, serde: Boolean = false): String = {
     val result = if (serde) {
       sql(s"SHOW CREATE TABLE $view AS SERDE")