[
https://issues.apache.org/jira/browse/SPARK-60040?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Luka Zdravic updated SPARK-60040:
---------------------------------
Description:
SortMergeAsOfJoinExec guesses the AS-OF direction from the shape of
asOfCondition:
{code:scala}
private val asOfDirection: AsOfJoinDirection =
splitConjunctivePredicates(asOfCondition).head match {
case c: BinaryComparison
if c.left.references.subsetOf(left.outputSet) &&
c.right.references.subsetOf(right.outputSet) =>
c match {
case _: GreaterThanOrEqual | _: GreaterThan => Backward
case _: LessThanOrEqual | _: LessThan => Forward
case _ => Nearest
}
case _ => Nearest
}
{code}
The guess works only if the first comparison has the left side first. Any other
shape becomes Nearest.
The plan knows the real direction, but drops it:
* DataFrame path: AsOfJoin.apply uses the direction only to build
asOfCondition and orderExpression.
* SQL path: ResolveAsOfJoin and AsOfJoinResolver set matchOperator = None
after resolution.
Why this is a problem:
* A rewrite that changes the shape silently turns Backward or Forward into
Nearest. For example, moving a constant to the
right turns {{10 >= r.ts}} into {{{}r.ts <= 10{}}}.
* The shape cannot always tell the direction. Nearest with a tolerance is
already right side first: r.ts >= l.ts - tol AND
r.ts <= l.ts + tol. It is correct now only because the side check fails. A
mirrored check would run it as Forward.
* Backward run as Nearest can return a different row when right rows share an
as-of value. Backward keeps the last one.
Nearest keeps the first. Forward run as Nearest returns the same row. Both
compute a distance they do not need, and do not
stop when the as-of condition turns false.
Proposed fix:
* Add direction: AsOfJoinDirection to AsOfJoin and set it in AsOfJoin.apply.
* In the SQL paths, take it from the operator after normalizeMatchOperands
puts the left operand first. So {{r.ts <= l.ts}}
gives Backward.
* Pass it to SortMergeAsOfJoinExec and match on it exhaustively. Remove the
shape-based guess.
Tests:
* Add a test where the as-of condition does not have the left side first.
Check that the planned SortMergeAsOfJoinExec keeps
the requested direction. Check the node, not only the rows, because the rows
differ only when right rows share an as-of
value.
* Tighten the Nearest tests. Today they also pass if the join runs as Forward
or Backward. Example: left 10, right 7 and 9,
tolerance 3. Nearest returns 9. Forward returns 7.
Assigned to : [~lukazdravic]
was:
SortMergeAsOfJoinExec infers the AS-OF direction from the shape of
asOfCondition:
{code:scala}
private val asOfDirection: AsOfJoinDirection =
splitConjunctivePredicates(asOfCondition).head match {
case c: BinaryComparison
if c.left.references.subsetOf(left.outputSet) &&
c.right.references.subsetOf(right.outputSet) =>
c match {
case _: GreaterThanOrEqual | _: GreaterThan => Backward
case _: LessThanOrEqual | _: LessThan => Forward
case _ => Nearest
}
case _ => Nearest
}
{code}
The logical plan knows the direction exactly, but does not keep it:
* DataFrame path: AsOfJoin.apply takes a direction, but uses it only to build
asOfCondition and orderExpression.
* SQL path: ResolveAsOfJoin and AsOfJoinResolver set matchOperator = None
after resolution.
So the exec node relies on the optimized condition keeping the "left-side op
right-side" shape. If a rewrite changes that shape, the direction silently
becomes Nearest. Example: a backward condition l.ts >= r.ts rewritten as r.ts
<= l.ts falls through to case _ => Nearest. A Nearest scan can return a
different row than a Backward or Forward scan, and it computes a distance that
Backward and Forward do not need.
h3. Proposed fix
* Add direction: AsOfJoinDirection to AsOfJoin. Set it in AsOfJoin.apply and
in the SQL resolution paths.
* Pass it to SortMergeAsOfJoinExec and match on it exhaustively. Remove the
shape-based inference.
* Add a test where the as-of condition does not have the "left op right" shape
(for example, right side first), and check that the join keeps the requested
direction.
Assigned to : [~lukazdravic]
> Carry the AS-OF join direction to SortMergeAsOfJoinExec instead of inferring
> it from the as-of condition
> --------------------------------------------------------------------------------------------------------
>
> Key: SPARK-60040
> URL: https://issues.apache.org/jira/browse/SPARK-60040
> Project: Spark
> Issue Type: Improvement
> Components: SQL
> Affects Versions: 4.4.0
> Reporter: Luka Zdravic
> Priority: Minor
>
> SortMergeAsOfJoinExec guesses the AS-OF direction from the shape of
> asOfCondition:
> {code:scala}
> private val asOfDirection: AsOfJoinDirection =
> splitConjunctivePredicates(asOfCondition).head match {
> case c: BinaryComparison
> if c.left.references.subsetOf(left.outputSet) &&
> c.right.references.subsetOf(right.outputSet) =>
> c match {
> case _: GreaterThanOrEqual | _: GreaterThan => Backward
> case _: LessThanOrEqual | _: LessThan => Forward
> case _ => Nearest
> }
> case _ => Nearest
> }
> {code}
> The guess works only if the first comparison has the left side first. Any
> other shape becomes Nearest.
> The plan knows the real direction, but drops it:
> * DataFrame path: AsOfJoin.apply uses the direction only to build
> asOfCondition and orderExpression.
> * SQL path: ResolveAsOfJoin and AsOfJoinResolver set matchOperator = None
> after resolution.
> Why this is a problem:
> * A rewrite that changes the shape silently turns Backward or Forward into
> Nearest. For example, moving a constant to the
> right turns {{10 >= r.ts}} into {{{}r.ts <= 10{}}}.
> * The shape cannot always tell the direction. Nearest with a tolerance is
> already right side first: r.ts >= l.ts - tol AND
> r.ts <= l.ts + tol. It is correct now only because the side check fails. A
> mirrored check would run it as Forward.
> * Backward run as Nearest can return a different row when right rows share
> an as-of value. Backward keeps the last one.
> Nearest keeps the first. Forward run as Nearest returns the same row. Both
> compute a distance they do not need, and do not
> stop when the as-of condition turns false.
> Proposed fix:
> * Add direction: AsOfJoinDirection to AsOfJoin and set it in AsOfJoin.apply.
> * In the SQL paths, take it from the operator after normalizeMatchOperands
> puts the left operand first. So {{r.ts <= l.ts}}
> gives Backward.
> * Pass it to SortMergeAsOfJoinExec and match on it exhaustively. Remove the
> shape-based guess.
> Tests:
> * Add a test where the as-of condition does not have the left side first.
> Check that the planned SortMergeAsOfJoinExec keeps
> the requested direction. Check the node, not only the rows, because the rows
> differ only when right rows share an as-of
> value.
> * Tighten the Nearest tests. Today they also pass if the join runs as
> Forward or Backward. Example: left 10, right 7 and 9,
> tolerance 3. Nearest returns 9. Forward returns 7.
> Assigned to : [~lukazdravic]
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
