[SPARK-51119][SQL][FOLLOW-UP] Add fallback to ResolveDefaultColumnsUtil.existenceDefaultValues

Author: szehon-ho

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

@@ -19,14 +19,15 @@ package org.apache.spark.sql.catalyst.util
 
 import scala.collection.mutable.ArrayBuffer
 
-import org.apache.spark.{SparkThrowable, SparkUnsupportedOperationException}
+import org.apache.spark.{SparkException, SparkThrowable, SparkUnsupportedOperationException}
 import org.apache.spark.internal.{Logging, MDC}
 import org.apache.spark.internal.LogKeys._
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.{InternalRow, SQLConfHelper}
 import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, InMemoryCatalog, SessionCatalog}
 import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.{Literal => ExprLiteral}
 import org.apache.spark.sql.catalyst.optimizer.{ConstantFolding, Optimizer}
 import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException}
 import org.apache.spark.sql.catalyst.plans.logical._
@@ -340,12 +341,35 @@ object ResolveDefaultColumns extends QueryErrorsBase
       throw QueryCompilationErrors.defaultValuesMayNotContainSubQueryExpressions(
         "", field.name, defaultSQL)
     }
-    if (!expr.resolved) {
-      throw QueryCompilationErrors.defaultValuesUnresolvedExprError(
-        "", field.name, defaultSQL, null)
+
+    val resolvedExpr = expr match {
+      case _: ExprLiteral | _: Cast => expr
+      case _ =>
+        // In most cases, column existsDefault should already be persisted as resolved sql
+        // statements but because they are fetched from external catalog, it is possible
+        // that this assumption does not hold so we fallback to full analysis if we encounter
+        // an unresolved existsDefault
+        val fallback = fallbackResolveExistenceDefaultValue(field)
+        if (!fallback.resolved) {
+          throw QueryCompilationErrors.defaultValuesUnresolvedExprError(
+            "", field.name, defaultSQL, null)
+        }
+        fallback
     }
 
-    coerceDefaultValue(expr, field.dataType, "", field.name, defaultSQL)
+    coerceDefaultValue(resolvedExpr, field.dataType, "", field.name, defaultSQL)
+  }
+
+  private def fallbackResolveExistenceDefaultValue(field: StructField): Expression = {
+    val defaultValue: Option[String] = field.getExistenceDefaultValue()
+    defaultValue.map { text: String =>
+      val expr = analyze(field, "", EXISTS_DEFAULT_COLUMN_METADATA_KEY)
+      expr match {
+        case _: ExprLiteral | _: Cast => expr
+        case _ => throw SparkException.internalError(s"parse existence default as literal err," +
+          s" field name: ${field.name}, value: $text")
+      }
+    }.orNull
   }
 
   /**

sql/catalyst/src/test/scala/org/apache/spark/sql/types/StructTypeSuite.scala

@@ -831,4 +831,22 @@ class StructTypeSuite extends SparkFunSuite with SQLHelper {
     validateConvertedDefaults("c4", VariantType, "parse_json(null)", "CAST(NULL AS VARIANT)")
 
   }
+
+  test("SPARK-51119: Add fallback to process unresolved EXISTS_DEFAULT") {
+    val source = StructType(
+      Array(
+      StructField("c1", VariantType, true,
+        new MetadataBuilder()
+          .putString(ResolveDefaultColumns.EXISTS_DEFAULT_COLUMN_METADATA_KEY, "parse_json(null)")
+          .putString(ResolveDefaultColumns.CURRENT_DEFAULT_COLUMN_METADATA_KEY, "parse_json(null)")
+          .build()),
+      StructField("c0", StringType, true,
+        new MetadataBuilder()
+          .putString(ResolveDefaultColumns.EXISTS_DEFAULT_COLUMN_METADATA_KEY, "current_catalog()")
+          .putString(ResolveDefaultColumns.CURRENT_DEFAULT_COLUMN_METADATA_KEY, "current_catalog()")
+          .build())))
+    val res = ResolveDefaultColumns.existenceDefaultValues(source)
+    assert(res(0) == null)
+    assert(res(1) == "spark_catalog")
+  }
 }