dongjoon-hyun commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4077137405
##########
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:
nit. This scaladoc describes the caller-side details at length. A one or two
line description would be enough, e.g., `Non-throwing variant of resolveRef.
Returns None if the reference cannot be resolved.` In addition, can we make
this `private[sql]`?
##########
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:
This switches `toCatalyst` to `toCatalystOpt` for the arguments, so not only
unresolvable references but also nested transforms whose function cannot be
loaded now return `None` instead of throwing. It seems intended, but could you
mention it in the PR description? Also, unsupported expression types still
throw from `case _ =>` in `toCatalystOpt`, so the graceful degradation is
partial.
##########
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:
nit. This test case looks trivial. Do we need it?
##########
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:
Please add the JIRA ID prefix to the new test names, e.g.,
`test("SPARK-59721: ...")`.
##########
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:
This changes the error of `toCatalyst`/`toCatalystOrdering` too, because
`toCatalyst` is `toCatalystOpt(...).getOrElse(throw _LEGACY_ERROR_TEMP_3054)`.
The impact is broader than `toCatalystOrdering` mentioned in the PR description:
- `DistributionAndOrderingUtils` (write path): the clustering of
`requiredDistribution` via `toCatalyst` and `requiredOrdering` via
`toCatalystOrdering`
- `V2ScanPartitioningAndOrdering.ordering()`: `scan.outputOrdering()`
For writes, when a connector requires a non-existent column, the error
changes from `_LEGACY_ERROR_TEMP_1137` (`Unable to resolve x given [a,b]`) to
`_LEGACY_ERROR_TEMP_3054` (`x is not currently supported`), which is misleading.
Could you keep the specific error for the throwing APIs? For example, a
private helper with a `failOnUnresolved` flag which
`toCatalyst`/`toCatalystOrdering` use, while only the `*Opt` variants return
`None`. Note that adding a `FieldReference` case to `toCatalyst` alone is not
enough because `SortValue` recurses into `toCatalystOpt` for its child.
##########
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:
Since `funCatalogOpt` is `None` here (and in the `years` test below), the
result would be `None` even if the reference were resolvable. It distinguishes
the old behavior (exception), but could you pass a function catalog so that the
test proves `None` comes 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]
