anuragmantri commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4116772248
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -39,21 +39,30 @@ import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.connector.PartitionPredicateImpl
import org.apache.spark.sql.types._
import org.apache.spark.util.ArrayImplicits._
+import org.apache.spark.util.collection.Utils.sequenceToOption
/**
* A utility class that converts public connector expressions into Catalyst
expressions.
*/
object V2ExpressionUtils extends SQLConfHelper with Logging {
import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
+ /**
+ * Non-throwing counterpart to `resolveRef`. Callers that need to gracefully
degrade when a
+ * reference cannot be resolved (e.g.
`toCatalystOpt`/`toCatalystTransformOpt`, which a reported
+ * `KeyGroupedPartitioning` relies on to fall back to unknown partitioning
instead of failing the
+ * query) must use this, not `resolveRef` -- calling the throwing form there
defeats that
+ * fallback entirely.
+ */
+ def resolveRefOpt[T <: NamedExpression](ref: NamedReference, plan:
LogicalPlan): Option[T] = {
Review Comment:
Done. Used your wording for the scaladoc and made it `private[sql]`.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtilsSuite.scala:
##########
@@ -25,16 +25,46 @@ import org.apache.spark.sql.types.StringType
class V2ExpressionUtilsSuite extends SparkFunSuite {
+ private val plan = LocalRelation.apply(AttributeReference("a", StringType)())
+
test("SPARK-39313: toCatalystOrdering should fail if V2Expression can not be
translated") {
val supportedV2Sort = SortValue(
FieldReference("a"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST)
val unsupportedV2Sort = supportedV2Sort.copy(
expression = ApplyTransform("v2Fun", FieldReference("a") :: Nil))
val exc = intercept[AnalysisException] {
- V2ExpressionUtils.toCatalystOrdering(
- Array(supportedV2Sort, unsupportedV2Sort),
- LocalRelation.apply(AttributeReference("a", StringType)()))
+ V2ExpressionUtils.toCatalystOrdering(Array(supportedV2Sort,
unsupportedV2Sort), plan)
}
assert(exc.message.contains("v2Fun(a) ASC NULLS FIRST is not currently
supported"))
}
+
+ test("toCatalystOpt returns None, not throw, for an unresolvable
FieldReference") {
Review Comment:
Done.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -131,7 +140,7 @@ object V2ExpressionUtils extends SQLConfHelper with Logging
{
SortOrder(catalystChild, toCatalyst(direction),
toCatalyst(nullOrdering), Seq.empty)
}
case ref: FieldReference =>
- Some(resolveRef[NamedExpression](ref, query))
+ resolveRefOpt[NamedExpression](ref, query)
Review Comment:
Good catch, thanks. Fixed as you suggested: `toCatalyst`/`toCatalystOpt` and
`toCatalystTransformOpt` now go through private
`toCatalystCore`/`toCatalystTransformCore` helpers that take a
`failOnUnresolved` flag. `toCatalyst` (and so `toCatalystOrdering`) passes
`true` and keeps throwing `_LEGACY_ERROR_TEMP_1137`, while the `*Opt` variants
pass `false` and return `None`. The flag is threaded through the `SortValue`
child and the `NamedTransform` args, so the recursion you pointed out keeps the
specific error too. `resolveRef` itself is unchanged.
The `toCatalystOrdering` test now asserts `_LEGACY_ERROR_TEMP_1137`, and I
added a direct `toCatalyst` test for an unresolvable `FieldReference`. I also
removed the side-effect paragraph from the PR description since it no longer
applies.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtilsSuite.scala:
##########
@@ -25,16 +25,46 @@ import org.apache.spark.sql.types.StringType
class V2ExpressionUtilsSuite extends SparkFunSuite {
+ private val plan = LocalRelation.apply(AttributeReference("a", StringType)())
+
test("SPARK-39313: toCatalystOrdering should fail if V2Expression can not be
translated") {
val supportedV2Sort = SortValue(
FieldReference("a"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST)
val unsupportedV2Sort = supportedV2Sort.copy(
expression = ApplyTransform("v2Fun", FieldReference("a") :: Nil))
val exc = intercept[AnalysisException] {
- V2ExpressionUtils.toCatalystOrdering(
- Array(supportedV2Sort, unsupportedV2Sort),
- LocalRelation.apply(AttributeReference("a", StringType)()))
+ V2ExpressionUtils.toCatalystOrdering(Array(supportedV2Sort,
unsupportedV2Sort), plan)
}
assert(exc.message.contains("v2Fun(a) ASC NULLS FIRST is not currently
supported"))
}
+
+ test("toCatalystOpt returns None, not throw, for an unresolvable
FieldReference") {
+ assert(V2ExpressionUtils.toCatalystOpt(FieldReference("missing"),
plan).isEmpty)
+ }
+
+ test("toCatalystOpt still resolves an existing FieldReference") {
Review Comment:
Removed
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -144,25 +153,23 @@ object V2ExpressionUtils extends SQLConfHelper with
Logging {
query: LogicalPlan,
funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] =
trans match {
case IdentityTransform(ref) =>
- Some(resolveRef[NamedExpression](ref, query))
+ resolveRefOpt[NamedExpression](ref, query)
case BucketTransform(numBuckets, refs, sorted)
if sorted.isEmpty && refs.length == 1 &&
refs.forall(_.isInstanceOf[NamedReference]) =>
- val resolvedRefs = refs.map(r => resolveRef[NamedExpression](r, query))
// Create a dummy reference for `numBuckets` here and use that, together
with `refs`, to
// look up the V2 function.
val numBucketsRef = AttributeReference("numBuckets", IntegerType,
nullable = false)()
- funCatalogOpt.flatMap { catalog =>
- loadV2FunctionOpt(catalog, "bucket", Seq(numBucketsRef) ++
resolvedRefs).map { bound =>
- TransformExpression(bound, resolvedRefs, Some(numBuckets))
- }
- }
+ for {
+ resolvedRefs <- sequenceToOption(refs.map(r =>
resolveRefOpt[NamedExpression](r, query)))
+ catalog <- funCatalogOpt
+ bound <- loadV2FunctionOpt(catalog, "bucket", Seq(numBucketsRef) ++
resolvedRefs)
+ } yield TransformExpression(bound, resolvedRefs, Some(numBuckets))
case NamedTransform(name, args) =>
- val catalystArgs = args.map(toCatalyst(_, query, funCatalogOpt))
- funCatalogOpt.flatMap { catalog =>
- loadV2FunctionOpt(catalog, name, catalystArgs).map { bound =>
- TransformExpression(bound, catalystArgs)
- }
- }
+ for {
+ catalystArgs <- sequenceToOption(args.map(toCatalystOpt(_, query,
funCatalogOpt)))
Review Comment:
Yes, that's intended. I added both points to the PR description. On the
throwing path, a nested transform whose function can't be loaded still fails
with `_LEGACY_ERROR_TEMP_3054`, but the message now names the outer transform
instead of the inner one.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtilsSuite.scala:
##########
@@ -25,16 +25,46 @@ import org.apache.spark.sql.types.StringType
class V2ExpressionUtilsSuite extends SparkFunSuite {
+ private val plan = LocalRelation.apply(AttributeReference("a", StringType)())
+
test("SPARK-39313: toCatalystOrdering should fail if V2Expression can not be
translated") {
val supportedV2Sort = SortValue(
FieldReference("a"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST)
val unsupportedV2Sort = supportedV2Sort.copy(
expression = ApplyTransform("v2Fun", FieldReference("a") :: Nil))
val exc = intercept[AnalysisException] {
- V2ExpressionUtils.toCatalystOrdering(
- Array(supportedV2Sort, unsupportedV2Sort),
- LocalRelation.apply(AttributeReference("a", StringType)()))
+ V2ExpressionUtils.toCatalystOrdering(Array(supportedV2Sort,
unsupportedV2Sort), plan)
}
assert(exc.message.contains("v2Fun(a) ASC NULLS FIRST is not currently
supported"))
}
+
+ test("toCatalystOpt returns None, not throw, for an unresolvable
FieldReference") {
+ assert(V2ExpressionUtils.toCatalystOpt(FieldReference("missing"),
plan).isEmpty)
+ }
+
+ test("toCatalystOpt still resolves an existing FieldReference") {
+ assert(V2ExpressionUtils.toCatalystOpt(FieldReference("a"),
plan).isDefined)
+ }
+
+ test("toCatalystTransformOpt returns None for an unresolvable
IdentityTransform") {
+
assert(V2ExpressionUtils.toCatalystTransformOpt(Expressions.identity("missing"),
plan).isEmpty)
+ }
+
+ test("toCatalystTransformOpt returns None for an unresolvable
BucketTransform ref") {
+ assert(V2ExpressionUtils.toCatalystTransformOpt(Expressions.bucket(4,
"missing"), plan).isEmpty)
Review Comment:
Good point. The `bucket` and `years` tests now pass a function catalog: a
small `UnboundFunction` registered under both names in an `InMemoryCatalog`.
Each test also asserts that the same transform on the existing column `a`
converts, so the `None` for `missing` can only come from the resolution failure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
