vranes commented on code in PR #59016:
URL: https://github.com/apache/spark/pull/59016#discussion_r4116301564
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -94,32 +94,36 @@ case class SortMergeAsOfJoinExec(
// Backward joins scan the right-side buffer forward and keep the last match.
// Detect from asOfCondition so composite MATCH_CONDITION sort keys still
work.
- private val isBackwardJoin: Boolean =
- isBackwardAsOfCondition(asOfCondition, left.output, right.output)
-
- private def isBackwardAsOfCondition(
- condition: Expression,
- leftOutput: Seq[Attribute],
- rightOutput: Seq[Attribute]): Boolean = {
- val leftAttrs = AttributeSet(leftOutput)
- val rightAttrs = AttributeSet(rightOutput)
- def isBackwardPredicate(expr: Expression): Boolean = expr match {
- case GreaterThanOrEqual(l, r)
- if l.references.subsetOf(leftAttrs) &&
r.references.subsetOf(rightAttrs) =>
- true
- case GreaterThan(l, r)
- if l.references.subsetOf(leftAttrs) &&
r.references.subsetOf(rightAttrs) =>
- true
- case And(c, _) => isBackwardPredicate(c)
- case _ => false
+ private val isBackwardJoin: Boolean = leadingAsOfComparison.exists {
Review Comment:
Could we derive one AsOfJoinDirection here and pass it to the scanner? That
would avoid duplicating the check and make `nearest` explicit instead of
representing it as two false values
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala:
##########
@@ -3123,16 +3115,11 @@ object AsOfJoin {
}
}
- private def wrapCompositeOrderExpression(
- diffs: Seq[Expression],
- compositeType: DataType): Expression = {
+ /** A struct, not an array, since field distances can differ in type (INT vs
INTERVAL). */
Review Comment:
```suggestion
/** Groups multiple field distances in a struct because their types may
differ. */
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala:
##########
@@ -2772,8 +2772,8 @@ object AsOfJoin {
*/
private[catalyst] object MatchConditionTypes {
- def isValidOperandType(dataType: DataType): Boolean =
- RowOrdering.isOrderable(dataType) && !containsEmptyStructType(dataType)
+ /** Any orderable type, including an empty struct, which equals any other
empty struct. */
Review Comment:
nit: unnecessary comment
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -424,10 +436,40 @@ private[joins] class SortMergeAsOfJoinScanner(
}
/**
- * Forward scan for Forward/Nearest joins: distance-based termination.
Review Comment:
nit: This repeats nearby documentation and could be shorter
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala:
##########
@@ -435,6 +435,67 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
Row(Row(5, Seq(1, 1))) :: Nil)
}
Review Comment:
Add a forward test that makes the closest right row fail the ON condition
and a later row pass
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -156,7 +160,10 @@ case class SortMergeAsOfJoinExec(
* last as-of-satisfying row as the best match (since the buffer is
* sorted ascending, the last satisfying row is the closest).
*
- * For Forward/Nearest joins, the forward scan uses distance-based
+ * For Forward joins (left.t <= right.t), the first as-of-satisfying row
+ * is the match (the buffer is sorted ascending, so it is the smallest).
+ *
+ * For Nearest joins, the forward scan uses distance-based
Review Comment:
nit: If this comment describes Forward and Nearest, why does it leave out
Backward?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -424,10 +436,40 @@ private[joins] class SortMergeAsOfJoinScanner(
}
/**
- * Forward scan for Forward/Nearest joins: distance-based termination.
Review Comment:
Also, `findFirstForward` and `findLastBackward` comment style should be
aligned.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -365,14 +373,18 @@ private[joins] class SortMergeAsOfJoinScanner(
* right.t <= left.t, which is the closest match). Early-terminates
* when as-of condition transitions from true to false (monotone).
*
- * For Forward/Nearest joins: uses distance-based early termination
+ * For Forward joins: returns the first as-of-satisfying row.
+ *
+ * For Nearest joins: uses distance-based early termination
* (stop when distance starts increasing past the minimum).
*/
private def findBestInGroup(leftRow: InternalRow): InternalRow = {
if (isBackwardJoin) {
findBestBackwardForward(leftRow)
Review Comment:
Rename to `findLastBackward` to match `findFirstForward`. The third one
should be `findNearest`.
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala:
##########
@@ -728,6 +807,74 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
Row(Seq(1, 6)) :: Nil)
}
+ test("forward MATCH_CONDITION with NULL array elements picks the smallest
right row") {
+ // NULL elements sort first: [null] < [null, null] < [5]. A distance-based
pick chose [5].
Review Comment:
nit: Historical / iterative comment. This belongs more to the PR
description, not in the code.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -156,7 +160,10 @@ case class SortMergeAsOfJoinExec(
* last as-of-satisfying row as the best match (since the buffer is
* sorted ascending, the last satisfying row is the closest).
*
- * For Forward/Nearest joins, the forward scan uses distance-based
+ * For Forward joins (left.t <= right.t), the first as-of-satisfying row
+ * is the match (the buffer is sorted ascending, so it is the smallest).
+ *
+ * For Nearest joins, the forward scan uses distance-based
Review Comment:
Same below, plus these 2 comments are repeating themselves
--
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]
