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

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

@@ -256,7 +256,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.")
@@ -265,6 +268,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

@@ -43,7 +43,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

@@ -1002,7 +1002,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)
         }
@@ -1030,7 +1036,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 {
@@ -1642,6 +1652,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

@@ -864,7 +864,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
@@ -873,7 +876,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

@@ -6243,6 +6243,15 @@ object SQLConf {
       .createWithDefault(false)
   }
 
+  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

@@ -493,16 +493,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,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
+
+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 {
+
+  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
+    )
+  }
+}