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

This reverts commit fbedfb1.

Author: dongjoon-hyun

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

@@ -258,10 +258,7 @@ object Analyzer {
     "spark.sql.expressionTreeChangeLog.level"
   )
 
-  def retainResolutionConfigsForAnalysis(
-      newConf: SQLConf,
-      existingConf: SQLConf,
-      createSparkVersion: String = ""): Unit = {
+  def retainResolutionConfigsForAnalysis(newConf: SQLConf, existingConf: SQLConf): Unit = {
     val retainedConfigs = existingConf.getAllConfs.filter { case (key, _) =>
       // Also apply catalog configs
       RETAINED_ANALYSIS_FLAGS.contains(key) || key.startsWith("spark.sql.catalog.")
@@ -270,25 +267,6 @@ 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,13 +41,7 @@ object ViewResolution {
           view
         )
       }
-      SQLConf.withExistingConf(
-        View.effectiveSQLConf(
-          configs = view.desc.viewSQLConfigs,
-          isTempView = view.isTempView,
-          createSparkVersion = view.desc.createVersion
-        )
-      ) {
+      SQLConf.withExistingConf(View.effectiveSQLConf(view.desc.viewSQLConfigs, view.isTempView)) {
         resolveChild(view.child)
       }
     }

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

@@ -982,13 +982,7 @@ class SessionCatalog(
       objectType = Some("VIEW"),
       objectName = Some(metadata.qualifiedName)
     )
-    val parsedPlan = SQLConf.withExistingConf(
-      View.effectiveSQLConf(
-        configs = viewConfigs,
-        isTempView = isTempView,
-        createSparkVersion = metadata.createVersion
-      )
-    ) {
+    val parsedPlan = SQLConf.withExistingConf(View.effectiveSQLConf(viewConfigs, isTempView)) {
         CurrentOrigin.withOrigin(origin) {
           parser.parseQuery(viewText)
         }
@@ -1016,11 +1010,7 @@ 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(
-          configs = metadata.viewSQLConfigs,
-          isTempView = isTempView,
-          createSparkVersion = metadata.createVersion
-        )
+        val viewConf = View.effectiveSQLConf(metadata.viewSQLConfigs, isTempView)
         val normalizeColName: String => String = if (viewConf.caseSensitiveAnalysis) {
           identity
         } else {
@@ -1627,7 +1617,6 @@ 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,10 +835,7 @@ case class View(
 }
 
 object View {
-  def effectiveSQLConf(
-      configs: Map[String, String],
-      isTempView: Boolean,
-      createSparkVersion: String = ""): SQLConf = {
+  def effectiveSQLConf(configs: Map[String, String], isTempView: Boolean): SQLConf = {
     val activeConf = SQLConf.get
     // For temporary view, we always use captured sql configs
     if (activeConf.useCurrentSQLConfigsForView && !isTempView) return activeConf
@@ -847,12 +844,7 @@ object View {
     for ((k, v) <- configs) {
       sqlConf.settings.put(k, v)
     }
-    Analyzer.retainResolutionConfigsForAnalysis(
-      newConf = sqlConf,
-      existingConf = activeConf,
-      createSparkVersion = createSparkVersion
-    )
-
+    Analyzer.retainResolutionConfigsForAnalysis(newConf = sqlConf, existingConf = activeConf)
     sqlConf
   }
 }

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

@@ -5594,15 +5594,6 @@ 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,8 +19,6 @@ 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}
@@ -89,25 +87,10 @@ 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)
-
-    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
+    modifiedConfs.map { case (key, value) => s"$SQL_CONFIG_PREFIX$key" -> value }
   }
 
   /**

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

@@ -481,19 +481,16 @@ object ViewHelper extends SQLConfHelper with Logging {
   }
 
   /**
-   * 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.
+   * Convert the view SQL configs to `properties`.
    */
   private def sqlConfigsToProps(conf: SQLConf): Map[String, String] = {
     val modifiedConfs = getModifiedConf(conf)
-
-    val alwaysCaptured = Seq(SQLConf.SESSION_LOCAL_TIMEZONE, SQLConf.ANSI_ENABLED)
+    // 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)
       .filter(c => !modifiedConfs.contains(c.key))
-      .map(c => (c.key, conf.getConf(c).toString))
+      .map(c => (c.key, conf.getConf(c)))
 
     val props = new mutable.HashMap[String, String]
     for ((key, value) <- modifiedConfs ++ alwaysCaptured) {