peter-toth commented on code in PR #58979:
URL: https://github.com/apache/spark/pull/58979#discussion_r4132086733
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -113,25 +128,35 @@ object V2ExpressionUtils extends SQLConfHelper with
Logging {
expr: V2Expression,
query: LogicalPlan,
funCatalogOpt: Option[FunctionCatalog] = None): Expression =
- toCatalystOpt(expr, query, funCatalogOpt)
+ toCatalystCore(expr, query, funCatalogOpt, failOnUnresolved = true)
.getOrElse(throw new AnalysisException(
errorClass = "_LEGACY_ERROR_TEMP_3054", messageParameters =
Map("expr" -> expr.toString)))
def toCatalystOpt(
expr: V2Expression,
query: LogicalPlan,
- funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] = {
+ funCatalogOpt: Option[FunctionCatalog] = None): Option[Expression] =
+ toCatalystCore(expr, query, funCatalogOpt, failOnUnresolved = false)
Review Comment:
**Finding 1.** The precision fix from
[r4117400462](https://github.com/apache/spark/pull/58979#discussion_r4117400462)
landed on `resolveRefOpt`, which is `private[sql]`. `toCatalystOpt` and
`toCatalystTransformOpt` are the public entry points, and they carry no
scaladoc, so the split contract is only discoverable three levels down.
Measured on this head:
```
toCatalystOpt(FieldReference("missing"), plan) -> None
toCatalystOpt(FieldReference("s.missing"), plan) -> throws
[FIELD_NOT_FOUND]
resolveRefOpt(FieldReference("a"), relationWithTwoAs) -> throws
[AMBIGUOUS_REFERENCE]
toCatalystOpt(GeneralScalarExpression("+", [id, 1]), plan) -> throws "id +
1 is not currently supported"
```
This is not hypothetical. The rule in this PR had to wrap `resolveRefOpt` in
`try { ... } catch { case _: AnalysisException => true }` for exactly that
reason, and the next caller will read `toCatalystOpt` the way its name invites.
A couple of lines each:
```scala
/**
* Variant of `toCatalyst` that returns `None` when a reference cannot be
resolved. It still
* throws for a missing nested field or an ambiguous reference (see
`resolveRefOpt`), and for an
* expression type this conversion does not handle.
*/
def toCatalystOpt(
```
and the same on `toCatalystTransformOpt`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala:
##########
@@ -75,8 +87,34 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
private def ordering(plan: LogicalPlan) = plan.transformDownWithPruning(
_.containsPattern(DATA_SOURCE_V2_SCAN_RELATION)) {
case d @ ExtractV2ScanInfo(relation, scan: SupportsReportOrdering, _) =>
- val ordering =
- V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation,
relation.funCatalog)
- d.copy(ordering = Some(ordering))
+ val unresolvedColumns =
+ unresolvableColumns(scan.outputOrdering().toImmutableArraySeq,
relation)
+ if (unresolvedColumns.nonEmpty) {
+ logWarning(
+ log"Spark ignores the reported ordering because the ordering columns
cannot be " +
+ log"resolved: ${MDC(COLUMN_NAMES, unresolvedColumns.mkString(",
"))}.")
+ d.copy(ordering = None)
+ } else {
+ val ordering =
+ V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(),
relation, relation.funCatalog)
+ d.copy(ordering = Some(ordering))
+ }
+ }
+
+ private def unresolvableColumns(
+ exprs: Seq[V2Expression],
+ relation: LogicalPlan): Seq[String] = {
+ def references(expr: V2Expression): Seq[NamedReference] = expr match {
+ case ref: NamedReference => Seq(ref)
+ case other => other.children().toImmutableArraySeq.flatMap(references)
+ }
+ exprs.flatMap(references).distinct.filter { ref =>
+ try {
+ V2ExpressionUtils.resolveRefOpt[NamedExpression](ref, relation).isEmpty
+ } catch {
+ // A missing nested field or an ambiguous name throws instead of
returning None.
+ case _: AnalysisException => true
Review Comment:
**Finding 2.** This `catch` absorbs two distinct throws from
`resolveRefOpt`, and only one of them is exercised. The end-to-end test covers
`s.missing` (`FIELD_NOT_FOUND`); the ambiguous-reference arm has no test in
this suite or in `V2ExpressionUtilsSuite`, although the `resolveRefOpt`
scaladoc and the PR description both name it.
Building a v2 relation with two same-named columns to reach it end to end
looks like more trouble than it is worth, so the cheap place is
`V2ExpressionUtilsSuite`, pinning the precondition this `catch` rests on:
```scala
test("SPARK-59721: resolveRefOpt throws for an ambiguous reference") {
val dup = LocalRelation.apply(
AttributeReference("a", StringType)(), AttributeReference("a",
StringType)())
intercept[AnalysisException] {
V2ExpressionUtils.resolveRefOpt[NamedExpression](FieldReference("a"),
dup)
}
}
```
I ran that relation on this head and it throws ``[AMBIGUOUS_REFERENCE]
Reference `a` is ambiguous, could be: [`a`, `a`]``. `checkError` with
`condition = "AMBIGUOUS_REFERENCE"` would be the better assertion; I did not
run it, so the parameter map would need checking.
--
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]
