Skip to content

Commit 64b62ca

Browse files
committed
[SPARK-51119][SQL][FOLLOW-UP] Add fallback to ResolveDefaultColumnsUtil.existenceDefaultValues
1 parent e397207 commit 64b62ca

File tree

2 files changed

+55
-5
lines changed

2 files changed

+55
-5
lines changed

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

Lines changed: 37 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -19,14 +19,15 @@ package org.apache.spark.sql.catalyst.util
1919

2020
import scala.collection.mutable.ArrayBuffer
2121

22-
import org.apache.spark.{SparkThrowable, SparkUnsupportedOperationException}
22+
import org.apache.spark.{SparkException, SparkThrowable, SparkUnsupportedOperationException}
2323
import org.apache.spark.internal.{Logging, MDC}
2424
import org.apache.spark.internal.LogKeys._
2525
import org.apache.spark.sql.AnalysisException
2626
import org.apache.spark.sql.catalyst.{InternalRow, SQLConfHelper}
2727
import org.apache.spark.sql.catalyst.analysis._
2828
import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, InMemoryCatalog, SessionCatalog}
2929
import org.apache.spark.sql.catalyst.expressions._
30+
import org.apache.spark.sql.catalyst.expressions.{Literal => ExprLiteral}
3031
import org.apache.spark.sql.catalyst.optimizer.{ConstantFolding, Optimizer}
3132
import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException}
3233
import org.apache.spark.sql.catalyst.plans.logical._
@@ -340,12 +341,43 @@ object ResolveDefaultColumns extends QueryErrorsBase
340341
throw QueryCompilationErrors.defaultValuesMayNotContainSubQueryExpressions(
341342
"", field.name, defaultSQL)
342343
}
343-
if (!expr.resolved) {
344-
throw QueryCompilationErrors.defaultValuesUnresolvedExprError(
345-
"", field.name, defaultSQL, null)
344+
345+
val resolvedExpr = expr match {
346+
case _: ExprLiteral | _: Cast => expr
347+
case _ =>
348+
fallbackResolveExistenceDefaultValue(field, defaultSQL)
346349
}
347350

348-
coerceDefaultValue(expr, field.dataType, "", field.name, defaultSQL)
351+
coerceDefaultValue(resolvedExpr, field.dataType, "", field.name, defaultSQL)
352+
}
353+
354+
// In most cases, column existsDefault should already be persisted as resolved
355+
// and constant-folded literal sql, but because they are fetched from external catalog,
356+
// it is possible that this assumption does not hold, so we fallback to full analysis
357+
// if we encounter an unresolved existsDefault
358+
private def fallbackResolveExistenceDefaultValue(
359+
field: StructField,
360+
defaultSQL: String): Expression = {
361+
logWarning(log"Encountered unresolved exists default value: " +
362+
log"'${MDC(COLUMN_DEFAULT_VALUE, defaultSQL)}' " +
363+
log"for column ${MDC(COLUMN_NAME, field.name)} " +
364+
log"with ${MDC(COLUMN_DATA_TYPE_SOURCE, field.dataType)}, " +
365+
log"falling back to full analysis.")
366+
367+
field.getExistenceDefaultValue().map { text: String =>
368+
val expr = analyze(field, "", EXISTS_DEFAULT_COLUMN_METADATA_KEY)
369+
val literal = expr match {
370+
case _: ExprLiteral | _: Cast => expr
371+
case _ => throw SparkException.internalError(s"parse existence default as literal err," +
372+
s" field name: ${field.name}, value: $text")
373+
}
374+
// sanity check
375+
if (!literal.resolved) {
376+
throw QueryCompilationErrors.defaultValuesUnresolvedExprError(
377+
"", field.name, defaultSQL, null)
378+
}
379+
literal
380+
}.orNull
349381
}
350382

351383
/**

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

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -831,4 +831,22 @@ class StructTypeSuite extends SparkFunSuite with SQLHelper {
831831
validateConvertedDefaults("c4", VariantType, "parse_json(null)", "CAST(NULL AS VARIANT)")
832832

833833
}
834+
835+
test("SPARK-51119: Add fallback to process unresolved EXISTS_DEFAULT") {
836+
val source = StructType(
837+
Array(
838+
StructField("c1", VariantType, true,
839+
new MetadataBuilder()
840+
.putString(ResolveDefaultColumns.EXISTS_DEFAULT_COLUMN_METADATA_KEY, "parse_json(null)")
841+
.putString(ResolveDefaultColumns.CURRENT_DEFAULT_COLUMN_METADATA_KEY, "parse_json(null)")
842+
.build()),
843+
StructField("c0", StringType, true,
844+
new MetadataBuilder()
845+
.putString(ResolveDefaultColumns.EXISTS_DEFAULT_COLUMN_METADATA_KEY, "current_catalog()")
846+
.putString(ResolveDefaultColumns.CURRENT_DEFAULT_COLUMN_METADATA_KEY, "current_catalog()")
847+
.build())))
848+
val res = ResolveDefaultColumns.existenceDefaultValues(source)
849+
assert(res(0) == null)
850+
assert(res(1) == "spark_catalog")
851+
}
834852
}

0 commit comments

Comments
 (0)