[SPARK-39862][SQL] Fix two bugs in existence DEFAULT value lookups

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

@@ -518,7 +518,10 @@ case class ResolveDefaultColumns(catalog: SessionCatalog) extends Rule[LogicalPl
       case Some(r: UnresolvedCatalogRelation) => r.tableMeta.identifier
       case _ => return None
     }
-
+    // First try to get the table metadata directly. If that fails, check for views below.
+    if (catalog.tableExists(tableName)) {
+      return Some(catalog.getTableMetadata(tableName).schema)
+    }
     val lookup: LogicalPlan = try {
       catalog.lookupRelation(tableName)
     } catch {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala

@@ -499,7 +499,7 @@ case class Literal (value: Any, dataType: DataType) extends LeafExpression {
         }
       val structFields: Array[String] =
         structNames.zip(structValues).map {
-          kv => s"${kv._1}, ${kv._2}"
+          kv => s"'${kv._1}', ${kv._2}"
         }
       s"NAMED_STRUCT(${structFields.mkString(", ")})"
     case (data: ArrayBasedMapData, mapType: MapType) =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumnsUtil.scala

@@ -127,11 +127,16 @@ object ResolveDefaultColumns {
    * @param field         represents the DEFAULT column value whose "default" metadata to parse
    *                      and analyze.
    * @param statementType which type of statement we are running, such as INSERT; useful for errors.
+   * @param metadataKey   which key to look up from the column metadata; generally either
+   *                      CURRENT_DEFAULT_COLUMN_METADATA_KEY or EXISTS_DEFAULT_COLUMN_METADATA_KEY.
    * @return Result of the analysis and constant-folding operation.
    */
-  def analyze(field: StructField, statementType: String): Expression = {
+  def analyze(
+      field: StructField,
+      statementType: String,
+      metadataKey: String = CURRENT_DEFAULT_COLUMN_METADATA_KEY): Expression = {
     // Parse the expression.
-    val colText: String = field.metadata.getString(CURRENT_DEFAULT_COLUMN_METADATA_KEY)
+    val colText: String = field.metadata.getString(metadataKey)
     lazy val parser = new CatalystSqlParser()
     val parsed: Expression = try {
       parser.parseExpression(colText)
@@ -202,7 +207,7 @@ object ResolveDefaultColumns {
       val defaultValue: Option[String] = field.getExistenceDefaultValue()
       defaultValue.map { text: String =>
         val expr = try {
-          val expr = analyze(field, "")
+          val expr = analyze(field, "", EXISTS_DEFAULT_COLUMN_METADATA_KEY)
           expr match {
             case _: ExprLiteral | _: Cast => expr
           }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ExpressionTypeCheckingSuite.scala

@@ -247,9 +247,9 @@ class ExpressionTypeCheckingSuite extends SparkFunSuite with SQLHelper {
     assert(Literal.create(Array(1, 2, null), ArrayType(IntegerType)).sql ==
       "ARRAY(1, 2, CAST(NULL AS INT))")
     assert(Literal.default(StructType(Seq(StructField("col", StringType)))).sql ==
-      "NAMED_STRUCT(col, '')")
+      "NAMED_STRUCT('col', '')")
     assert(Literal.default(StructType(Seq(StructField("col", NullType)))).sql ==
-      "NAMED_STRUCT(col, NULL)")
+      "NAMED_STRUCT('col', NULL)")
     assert(Literal.create(Map(42L -> true), MapType(LongType, BooleanType)).sql ==
       "MAP(42L, true)")
     assert(Literal.create(Map(42L -> null), MapType(LongType, NullType)).sql ==

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogSuite.scala

@@ -1038,7 +1038,8 @@ abstract class CatalogTestUtils {
           .add("col2", "string")
           .add("a", IntegerType, nullable = true,
             new MetadataBuilder().putString(
-              ResolveDefaultColumns.CURRENT_DEFAULT_COLUMN_METADATA_KEY, "42").build())
+              ResolveDefaultColumns.CURRENT_DEFAULT_COLUMN_METADATA_KEY, "42")
+              .putString(ResolveDefaultColumns.EXISTS_DEFAULT_COLUMN_METADATA_KEY, "41").build())
           .add("b", StringType, nullable = false,
             new MetadataBuilder().putString(
               ResolveDefaultColumns.CURRENT_DEFAULT_COLUMN_METADATA_KEY, "\"abc\"").build())

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala

@@ -159,6 +159,9 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
       // Analyze the default column values.
       val statementType = "CREATE TABLE"
       assert(ResolveDefaultColumns.analyze(columnA, statementType).sql == "42")
+      assert(ResolveDefaultColumns
+        .analyze(columnA, statementType, ResolveDefaultColumns.EXISTS_DEFAULT_COLUMN_METADATA_KEY)
+        .sql == "41")
       assert(ResolveDefaultColumns.analyze(columnB, statementType).sql == "'abc'")
       assert(intercept[AnalysisException] {
         ResolveDefaultColumns.analyze(columnC, statementType)

sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala

@@ -1800,7 +1800,7 @@ class InsertSuite extends DataSourceTest with SharedSparkSession {
               if (testCase.insertNullsToStorage) {
                 null
               } else {
-                Row(Seq(Row(3, 4)), Seq(Map(false -> "mno", true -> "pqr")))
+                Row(Seq(Row(1, 2)), Seq(Map(false -> "def", true -> "jkl")))
               },
               Seq(Map(true -> "xyz"))),
             Row(3,