Another approach

Author: szehon-ho

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

@@ -140,10 +140,14 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase {
           }
           matched(ordinal)
 
+          // Try to resolve literal functions first (for DefaultValueExpression)
+        case u @ UnresolvedLiteralFunction(nameParts) => withPosition(u) {
+          LiteralFunctionResolution.resolve(nameParts).getOrElse(u)
+        }
+
         case u @ UnresolvedAttribute(nameParts) =>
           val result = withPosition(u) {
             resolveColumnByName(nameParts)
-              .orElse(LiteralFunctionResolution.resolve(nameParts))
               .map {
                 // We trim unnecessary alias here. Note that, we cannot trim the alias at top-level,
                 // as we should resolve `UnresolvedAttribute` to a named expression. The caller side
@@ -175,9 +179,6 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase {
             u.copy(child = newChild)
           }
 
-        case d @ DefaultValueExpression(c: Expression, _, _) =>
-          d.copy(child = resolveLiteralColumns(c))
-
         case _ => e.mapChildren(innerResolve(_, isTopLevel = false))
       }
       resolved.copyTagsFrom(e)
@@ -198,13 +199,6 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase {
     }
   }
 
-  private def resolveLiteralColumns(e: Expression) = {
-    e.transformWithPruning(_.containsPattern(UNRESOLVED_ATTRIBUTE)) {
-      case u @ UnresolvedAttribute(nameParts) =>
-        LiteralFunctionResolution.resolve(nameParts).getOrElse(u)
-    }
-  }
-
   // Resolves `UnresolvedAttribute` to `OuterReference`.
   protected def resolveOuterRef(e: Expression): Expression = {
     val outerPlan = AnalysisContext.get.outerPlan

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

@@ -270,6 +270,30 @@ object UnresolvedTVFAliases {
   }
 }
 
+/**
+ * Holds the name of a literal function that has yet to be resolved.
+ * Similar to UnresolvedAttribute but specifically for literal functions.
+ */
+case class UnresolvedLiteralFunction(nameParts: Seq[String]) extends Expression with Unevaluable {
+
+  def name: String =
+    nameParts.map(n => if (n.contains(".")) s"`$n`" else n).mkString(".")
+
+  override def dataType: DataType = throw new UnresolvedException("dataType")
+  override def nullable: Boolean = throw new UnresolvedException("nullable")
+  override lazy val resolved = false
+  override def children: Seq[Expression] = Nil
+
+  final override val nodePatterns: Seq[TreePattern] = Seq(UNRESOLVED_LITERAL_FUNCTION)
+
+  override def toString: String = s"'$name"
+
+  override def sql: String = nameParts.map(quoteIfNeeded).mkString(".")
+
+  override protected def withNewChildrenInternal(
+      newChildren: IndexedSeq[Expression]): Expression = this
+}
+
 /**
  * Holds the name of an attribute that has yet to be resolved.
  */

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -2932,9 +2932,9 @@ class AstBuilder extends DataTypeAstBuilder
           CurrentUser()
       }
     } else {
-      // If the parser is not in ansi mode, we should return `UnresolvedAttribute`, in case there
-      // are columns named `CURRENT_DATE` or `CURRENT_TIMESTAMP` or `CURRENT_TIME`
-      UnresolvedAttribute.quoted(ctx.name.getText)
+      // If the parser is not in ansi mode, we should return `UnresolvedLiteralFunction`,
+      // in case there are columns named `CURRENT_DATE` or `CURRENT_TIMESTAMP` or `CURRENT_TIME`
+      UnresolvedLiteralFunction(Seq(ctx.name.getText))
     }
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala

@@ -109,6 +109,7 @@ object TreePattern extends Enumeration  {
   val UNRESOLVED_DF_STAR: Value = Value
   val UNRESOLVED_FUNCTION: Value = Value
   val UNRESOLVED_IDENTIFIER: Value = Value
+  val UNRESOLVED_LITERAL_FUNCTION: Value = Value
   val UNRESOLVED_ORDINAL: Value = Value
   val UNRESOLVED_PLAN_ID: Value = Value
   val UNRESOLVED_WINDOW_EXPRESSION: Value = Value

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala

@@ -3917,7 +3917,7 @@ class DataSourceV2SQLSuiteV1Filter
       checkError(
         exception = intercept[AnalysisException] {
           sql(s"""CREATE table $t (
-           c1 STRING,
+           c1 timestamp,
            current_timestamp TIMESTAMP DEFAULT c1)""")
         },
         condition = "INVALID_DEFAULT_VALUE.NOT_CONSTANT",