[SPARK-53348][SQL][4.0] Always persist ANSI value when creating a view or assume it when querying if not stored

### What changes were proposed in this pull request?
I propose that we always store ANSI value when creating a view because otherwise users can be affected by unwanted behavior. For example if user creates a view on version that has ANSI = false by default he expects this not to fail.

```
CREATE VIEW view AS SELECT CAST('abc' AS INT) AS a;
SELECT * FROM view;
```

But if user queries the view on the version which has ANSI = true by default, above query is going to fail (because when we don't store the value, and we store it only if explicitly set, we use the default one). Number of this and similar use cases is huge, because ANSI impact area is huge and thus I propose that we always store the value.

If the value is not stored, I propose that we use createVersion field to determine whether the ANSI value should be true (Spark 4.0.0 and above) or false (lower than Spark 4.0.0). If the createVersion field wasn't stored during view creation, I propose that we assume that the ANSI = false because number of those views is incomparable larger than the ones expecting ANSI = true

### Why are the changes needed?
To improve user experience.

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

### How was this patch tested?
Added suite.

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

Closes #52147 from mihailoale-db/ansi40backport.

Authored-by: mihailoale-db <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: mihailoale-db

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -258,7 +258,10 @@ object Analyzer {
     "spark.sql.expressionTreeChangeLog.level"
   )
 
-  def retainResolutionConfigsForAnalysis(newConf: SQLConf, existingConf: SQLConf): Unit = {
+  def retainResolutionConfigsForAnalysis(
+      newConf: SQLConf,
+      existingConf: SQLConf,
+      createSparkVersion: String = ""): Unit = {
     val retainedConfigs = existingConf.getAllConfs.filter { case (key, _) =>
       // Also apply catalog configs
       RETAINED_ANALYSIS_FLAGS.contains(key) || key.startsWith("spark.sql.catalog.")
@@ -267,6 +270,25 @@ object Analyzer {
     retainedConfigs.foreach { case (k, v) =>
       newConf.settings.put(k, v)
     }
+
+    trySetAnsiValue(newConf, createSparkVersion)
+  }
+
+  /**
+   * In case ANSI value wasn't persisted for a view or a UDF, we set it to `true` in case Spark
+   * version used to create the view is 4.0.0 or higher. We set it to `false` in case Spark version
+   * is lower than 4.0.0 or if the Spark version wasn't stored (in that case we assume that the
+   * value is `false`)
+   */
+  def trySetAnsiValue(sqlConf: SQLConf, createSparkVersion: String = ""): Unit = {
+    if (conf.getConf(SQLConf.ASSUME_ANSI_FALSE_IF_NOT_PERSISTED) &&
+      !sqlConf.settings.containsKey(SQLConf.ANSI_ENABLED.key)) {
+      if (createSparkVersion.startsWith("4.")) {
+        sqlConf.settings.put(SQLConf.ANSI_ENABLED.key, "true")
+      } else {
+        sqlConf.settings.put(SQLConf.ANSI_ENABLED.key, "false")
+      }
+    }
   }
 }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ViewResolution.scala

@@ -41,7 +41,13 @@ object ViewResolution {
           view
         )
       }
-      SQLConf.withExistingConf(View.effectiveSQLConf(view.desc.viewSQLConfigs, view.isTempView)) {
+      SQLConf.withExistingConf(
+        View.effectiveSQLConf(
+          configs = view.desc.viewSQLConfigs,
+          isTempView = view.isTempView,
+          createSparkVersion = view.desc.createVersion
+        )
+      ) {
         resolveChild(view.child)
       }
     }

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

@@ -982,7 +982,13 @@ class SessionCatalog(
       objectType = Some("VIEW"),
       objectName = Some(metadata.qualifiedName)
     )
-    val parsedPlan = SQLConf.withExistingConf(View.effectiveSQLConf(viewConfigs, isTempView)) {
+    val parsedPlan = SQLConf.withExistingConf(
+      View.effectiveSQLConf(
+        configs = viewConfigs,
+        isTempView = isTempView,
+        createSparkVersion = metadata.createVersion
+      )
+    ) {
         CurrentOrigin.withOrigin(origin) {
           parser.parseQuery(viewText)
         }
@@ -1010,7 +1016,11 @@ class SessionCatalog(
         // Note that, the column names may have duplication, e.g. `CREATE VIEW v(x, y) AS
         // SELECT 1 col, 2 col`. We need to make sure that the matching attributes have the same
         // number of duplications, and pick the corresponding attribute by ordinal.
-        val viewConf = View.effectiveSQLConf(metadata.viewSQLConfigs, isTempView)
+        val viewConf = View.effectiveSQLConf(
+          configs = metadata.viewSQLConfigs,
+          isTempView = isTempView,
+          createSparkVersion = metadata.createVersion
+        )
         val normalizeColName: String => String = if (viewConf.caseSensitiveAnalysis) {
           identity
         } else {
@@ -1617,6 +1627,7 @@ class SessionCatalog(
     // Use captured SQL configs when parsing a SQL function.
     val conf = new SQLConf()
     function.getSQLConfigs.foreach { case (k, v) => conf.settings.put(k, v) }
+    Analyzer.trySetAnsiValue(conf)
     SQLConf.withExistingConf(conf) {
       val inputParam = function.inputParam
       val returnType = function.getScalarFuncReturnType

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

@@ -835,7 +835,10 @@ case class View(
 }
 
 object View {
-  def effectiveSQLConf(configs: Map[String, String], isTempView: Boolean): SQLConf = {
+  def effectiveSQLConf(
+      configs: Map[String, String],
+      isTempView: Boolean,
+      createSparkVersion: String = ""): SQLConf = {
     val activeConf = SQLConf.get
     // For temporary view, we always use captured sql configs
     if (activeConf.useCurrentSQLConfigsForView && !isTempView) return activeConf
@@ -844,7 +847,12 @@ object View {
     for ((k, v) <- configs) {
       sqlConf.settings.put(k, v)
     }
-    Analyzer.retainResolutionConfigsForAnalysis(newConf = sqlConf, existingConf = activeConf)
+    Analyzer.retainResolutionConfigsForAnalysis(
+      newConf = sqlConf,
+      existingConf = activeConf,
+      createSparkVersion = createSparkVersion
+    )
+
     sqlConf
   }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -5594,6 +5594,15 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
+  val ASSUME_ANSI_FALSE_IF_NOT_PERSISTED =
+    buildConf("spark.sql.assumeAnsiFalseIfNotPersisted.enabled")
+      .internal()
+      .doc("If enabled, assume ANSI mode is false if not persisted during view or UDF " +
+        "creation. Otherwise use the default value.")
+      .version("4.0.1")
+      .booleanConf
+      .createWithDefault(true)
+
   /**
    * Holds information about keys that have been deprecated.
    *

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

@@ -19,6 +19,8 @@ package org.apache.spark.sql.execution.command
 
 import java.util.Locale
 
+import scala.collection.mutable
+
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.FunctionIdentifier
 import org.apache.spark.sql.catalyst.catalog.{LanguageSQL, RoutineLanguage, UserDefinedFunctionErrors}
@@ -87,10 +89,25 @@ object CreateUserDefinedFunctionCommand {
    * [[org.apache.spark.sql.catalyst.expressions.ExpressionInfo]], all SQL configs and other
    * function properties (such as the function parameters and the function return type)
    * are saved together in a property map.
+   *
+   * Here we only capture the SQL configs that are modifiable and should be captured, i.e. not in
+   * the denyList and in the allowList. Besides mentioned ones we also capture `ANSI_ENABLED`.
+   *
+   * We need to always capture them to make sure we apply the same configs when querying the
+   * function.
    */
   def sqlConfigsToProps(conf: SQLConf): Map[String, String] = {
     val modifiedConfs = ViewHelper.getModifiedConf(conf)
-    modifiedConfs.map { case (key, value) => s"$SQL_CONFIG_PREFIX$key" -> value }
+
+    val alwaysCaptured = Seq(SQLConf.ANSI_ENABLED)
+      .filter(c => !modifiedConfs.contains(c.key))
+      .map(c => (c.key, conf.getConf(c).toString))
+
+    val props = new mutable.HashMap[String, String]
+    for ((key, value) <- modifiedConfs ++ alwaysCaptured) {
+      props.put(s"$SQL_CONFIG_PREFIX$key", value)
+    }
+    props.toMap
   }
 
   /**

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

@@ -481,16 +481,19 @@ object ViewHelper extends SQLConfHelper with Logging {
   }
 
   /**
-   * Convert the view SQL configs to `properties`.
+   * Convert the view SQL configs to `properties`. Here we only capture the SQL configs that are
+   * modifiable and should be captured, i.e. not in the denyList and in the allowList. We also
+   * capture `SESSION_LOCAL_TIMEZONE` whose default value relies on the JVM system timezone and
+   * the `ANSI_ENABLED` value.
+   *
+   * We need to always capture them to make sure we apply the same configs when querying the view.
    */
   private def sqlConfigsToProps(conf: SQLConf): Map[String, String] = {
     val modifiedConfs = getModifiedConf(conf)
-    // Some configs have dynamic default values, such as SESSION_LOCAL_TIMEZONE whose
-    // default value relies on the JVM system timezone. We need to always capture them to
-    // to make sure we apply the same configs when reading the view.
-    val alwaysCaptured = Seq(SQLConf.SESSION_LOCAL_TIMEZONE)
+
+    val alwaysCaptured = Seq(SQLConf.SESSION_LOCAL_TIMEZONE, SQLConf.ANSI_ENABLED)
       .filter(c => !modifiedConfs.contains(c.key))
-      .map(c => (c.key, conf.getConf(c)))
+      .map(c => (c.key, conf.getConf(c).toString))
 
     val props = new mutable.HashMap[String, String]
     for ((key, value) <- modifiedConfs ++ alwaysCaptured) {

sql/core/src/test/scala/org/apache/spark/sql/DefaultANSIValueSuite.scala

@@ -0,0 +1,136 @@
+/*
+ * 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
+
+import org.scalactic.source.Position
+import org.scalatest.Tag
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.SQLScalarFunction
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType, SQLFunction}
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.plans.logical.{OneRowRelation, Project, View}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.StructType
+
+/**
+ * This suite tests if default ANSI value is persisted for views and functions if not explicitly
+ * set.
+ */
+class DefaultANSIValueSuite extends QueryTest with SharedSparkSession {
+
+  override protected def test(testName: String, testTags: Tag*)(testFun: => Any)(
+    implicit pos: Position): Unit = {
+    if (!sys.env.get("SPARK_ANSI_SQL_MODE").contains("false")) {
+      super.test(testName, testTags: _*)(testFun)
+    }
+  }
+
+  protected override def sparkConf: SparkConf = {
+    super.sparkConf
+      .set(SQLConf.ASSUME_ANSI_FALSE_IF_NOT_PERSISTED.key, "true")
+  }
+
+  private val testViewName = "test_view"
+  private val testFunctionName = "test_function"
+
+  test("Default ANSI value is stored for views") {
+    withView(testViewName) {
+      testView(expectedAnsiValue = true)
+    }
+  }
+
+  test("Explicitly set ANSI value is respected over default one for views") {
+    withView(testViewName) {
+      withSQLConf("spark.sql.ansi.enabled" -> "false") {
+        testView(expectedAnsiValue = false)
+      }
+    }
+
+    withView(testViewName) {
+      withSQLConf("spark.sql.ansi.enabled" -> "true") {
+        testView(expectedAnsiValue = true)
+      }
+    }
+  }
+
+  test("Default ANSI value is stored for functions") {
+    withUserDefinedFunction(testFunctionName -> false) {
+      testFunction(expectedAnsiValue = true)
+    }
+  }
+
+  test("Explicitly set ANSI value is respected over default one for functions") {
+    withUserDefinedFunction(testFunctionName -> false) {
+      withSQLConf("spark.sql.ansi.enabled" -> "false") {
+        testFunction(expectedAnsiValue = false)
+      }
+    }
+
+    withUserDefinedFunction(testFunctionName -> false) {
+      withSQLConf("spark.sql.ansi.enabled" -> "true") {
+        testFunction(expectedAnsiValue = true)
+      }
+    }
+  }
+
+  test("ANSI value is set to false if not persisted for views") {
+    val catalogTable = new CatalogTable(
+      identifier = TableIdentifier(testViewName),
+      tableType = CatalogTableType.VIEW,
+      storage = CatalogStorageFormat(None, None, None, None, false, Map.empty),
+      schema = new StructType(),
+      properties = Map.empty[String, String]
+    )
+    val view = View(desc = catalogTable, isTempView = false, child = OneRowRelation())
+
+    val sqlConf = View.effectiveSQLConf(view.desc.viewSQLConfigs, view.isTempView)
+
+    assert(sqlConf.settings.get("spark.sql.ansi.enabled") == "false")
+  }
+
+  private def testView(expectedAnsiValue: Boolean): Unit = {
+    sql(s"CREATE VIEW $testViewName AS SELECT CAST('string' AS BIGINT) AS alias")
+
+    val viewMetadata = spark.sessionState.catalog.getTableMetadata(TableIdentifier(testViewName))
+
+    assert(
+      viewMetadata.properties("view.sqlConfig.spark.sql.ansi.enabled") == expectedAnsiValue.toString
+    )
+  }
+
+  private def testFunction(expectedAnsiValue: Boolean): Unit = {
+    sql(
+      s"""
+         |CREATE OR REPLACE FUNCTION $testFunctionName()
+         |RETURN SELECT CAST('string' AS BIGINT) AS alias
+         |""".stripMargin)
+
+    val df = sql(s"select $testFunctionName()")
+
+    assert(
+      df.queryExecution.analyzed.asInstanceOf[Project]
+        .projectList.head.asInstanceOf[Alias]
+        .child.asInstanceOf[SQLScalarFunction]
+        .function.asInstanceOf[SQLFunction]
+        .properties.get("sqlConfig.spark.sql.ansi.enabled").get == expectedAnsiValue.toString
+    )
+  }
+}