vranes commented on code in PR #59102:
URL: https://github.com/apache/spark/pull/59102#discussion_r4133628651
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala:
##########
@@ -204,6 +204,26 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
Row(180.15) :: Nil)
}
+ test("TIMESTAMP MATCH_CONDITION with legacy intervals") {
+ // TIMESTAMP - TIMESTAMP is then a CalendarInterval, which has no
ordering. Backward and
Review Comment:
nit: This comment should be simplified or removed.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -207,7 +197,8 @@ private[joins] class SortMergeAsOfJoinScanner(
private val leftKeyProj = UnsafeProjection.create(leftKeys, leftOutput)
private val rightKeyProj = UnsafeProjection.create(rightKeys, rightOutput)
- private val distanceOrdering =
+ // Lazy for the same reason as boundOrderExpr.
+ private lazy val distanceOrdering =
Review Comment:
nit: Can `distanceOrdering` be next to `boundOrderExpr`? That way we can
have just one comment for both of them.
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala:
##########
@@ -204,6 +204,26 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
Row(180.15) :: Nil)
}
+ test("TIMESTAMP MATCH_CONDITION with legacy intervals") {
+ // TIMESTAMP - TIMESTAMP is then a CalendarInterval, which has no
ordering. Backward and
+ // Forward never read that distance, so they must still run.
+ withSQLConf(SQLConf.LEGACY_INTERVAL_ENABLED.key -> "true") {
+ for ((op, tag) <- Seq(">=" -> "r9", ">" -> "r9", "<=" -> "r11", "<" ->
"r11")) {
Review Comment:
This test has only one qualifying row per direction, so its coverage is
limited. Could it have multiple qualifying rows, including an exact match?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -92,34 +92,26 @@ case class SortMergeAsOfJoinExec(
override def outputOrdering: Seq[SortOrder] = left.outputOrdering
- // 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
+ // Detect from asOfCondition so composite MATCH_CONDITION sort keys still
work. Only a first
+ // `left op right` conjunct counts. Nearest has none: its condition is TRUE,
NOT(l = r), or
+ // starts with `right >= left - tolerance`.
Review Comment:
nit: This comment should be simplified or removed
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -424,10 +406,38 @@ private[joins] class SortMergeAsOfJoinScanner(
}
/**
- * Forward scan for Forward/Nearest joins: distance-based termination.
- * Stop when distance starts increasing past the minimum found so far.
+ * Forward joins: returns the first row that passes both conditions. The
buffer is sorted
+ * ascending, so that row is the closest and no distance is needed. The scan
stops once the
+ * as-of condition turns false after being true, since it stays false after
that.
*/
- private def findBestForwardNearest(leftRow: InternalRow): InternalRow = {
+ private def findFirstForward(leftRow: InternalRow): InternalRow = {
+ val iter = rightGroupBuffer.generateIterator()
+ var asOfSeen = false
+
+ joinedRow.withLeft(leftRow)
+ while (iter.hasNext) {
+ val rightRow = iter.next()
+ joinedRow.withRight(rightRow)
+
+ if (holds(boundAsOfCond)) {
+ asOfSeen = true
+ if (residualHolds) {
+ // No copy: the caller projects this row before the buffer iterator
moves again.
+ return rightRow
+ }
+ } else if (asOfSeen) {
+ // Past the tolerance bound, so no later row passes the as-of
condition.
+ return null
+ }
+ }
+ null
+ }
+
+ /**
+ * Nearest joins: keeps the row with the smallest distance. The scan stops
once the
+ * distance starts to grow past the minimum found so far.
Review Comment:
nit: This should say "once the distance stops decreasing", since the code
also stops on a tie.
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala:
##########
@@ -728,6 +748,111 @@ 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].
+ val nullInt = "CAST(NULL AS INT)"
+ for {
+ (left, rights) <- Seq(
Review Comment:
nit: Could one-item Seq generators in these tests be plain values? The
operator loops can stay.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -424,10 +406,38 @@ private[joins] class SortMergeAsOfJoinScanner(
}
/**
- * Forward scan for Forward/Nearest joins: distance-based termination.
- * Stop when distance starts increasing past the minimum found so far.
+ * Forward joins: returns the first row that passes both conditions. The
buffer is sorted
Review Comment:
nit: No need to mention the distance approach here.
Proposed comment:
```suggestion
* Forward joins return the first row that passes the as-of and residual
conditions.
* With a tolerance, if the scan enters the allowed range but finds no
match, it
* stops at the first row beyond that range, as later sorted rows cannot
qualify.
```
Backward coment:
```suggestion
/**
* Backward joins keep the last row that passes the as-of and residual
conditions.
* After saving a match, the scan stops at the first row that fails the
as-of
* condition, as later sorted rows cannot satisfy it.
*/
```
##########
sql/core/src/test/scala/org/apache/spark/sql/SortMergeAsOfJoinSuite.scala:
##########
@@ -685,6 +685,26 @@ class SortMergeAsOfJoinSuite extends QueryTest
)
}
+ test("forward join - residual rejects every row within the tolerance") {
Review Comment:
Could we add a left row (4, 30) to this test? It should match the existing
right row (6, 20), and the current (5, 10) row remains without a match.
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala:
##########
@@ -728,6 +748,111 @@ 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].
+ val nullInt = "CAST(NULL AS INT)"
+ for {
+ (left, rights) <- Seq(
+ s"ARRAY($nullInt)" -> s"(ARRAY($nullInt, $nullInt), 'nn'), (ARRAY(5),
'five')")
+ op <- Seq("<=", "<")
+ } {
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES ($left) AS t(a)
+ |ASOF JOIN VALUES $rights AS r(a, tag)
+ | MATCH_CONDITION (t.a $op r.a)
+ |""".stripMargin),
+ Row("nn") :: Nil)
+ }
+ }
+
+ test("forward MATCH_CONDITION with a NULL struct field picks the smallest
right row") {
+ // A NULL field sorts first, so {null, 7} < {x, 1}.
+ for {
+ (nullField, value) <- Seq("CAST(NULL AS INT)" -> "0")
+ (left, right) <- Seq("t.k" -> "r.k", "(t.k.f, t.k.seq)" -> "(r.k.f,
r.k.seq)")
+ op <- Seq("<=", "<")
+ } {
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES (named_struct('f', $nullField, 'seq', 5)) AS t(k)
+ |ASOF JOIN VALUES
+ | (named_struct('f', $nullField, 'seq', 7), 's7'),
+ | (named_struct('f', $value, 'seq', 1), 's1') AS r(k, tag)
+ | MATCH_CONDITION ($left $op $right)
+ |""".stripMargin),
+ Row("s7") :: Nil)
+ }
+ }
+
+ test("forward MATCH_CONDITION with a leading STRING field picks the smallest
right row") {
+ // {c, 1} < {d, 0}. The STRING field distance is 1 for both, so a distance
picks {d, 0}.
+ for {
+ (left, right) <- Seq("t.k" -> "r.k", "(t.k.s, t.k.n)" -> "(r.k.s,
r.k.n)")
+ op <- Seq("<=", "<")
+ } {
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES (named_struct('s', 'b', 'n', 5)) AS t(k)
+ |ASOF JOIN VALUES
+ | (named_struct('s', 'c', 'n', 1), 'c1'),
+ | (named_struct('s', 'd', 'n', 0), 'd0') AS r(k, tag)
+ | MATCH_CONDITION ($left $op $right)
+ |""".stripMargin),
+ Row("c1") :: Nil)
+ }
+ }
+
+ test("forward MATCH_CONDITION skips a closer row that fails a non-equi ON
predicate") {
+ checkSortMergeAsOf(
+ sql(
+ """
+ |SELECT r.k
+ |FROM VALUES (10, 5) AS t(k, lim)
+ |ASOF JOIN VALUES (12, 9), (15, 1), (20, 0) AS r(k, v)
+ | MATCH_CONDITION (t.k <= r.k)
+ | ON r.v < t.lim
+ |""".stripMargin),
+ Row(15) :: Nil)
+ }
+
+ // The far row makes (right - left) too big for the type, so a
distance-based scan would fail.
+ Seq(
+ ("INT", "-100", "-50", "2147483647"),
+ ("BIGINT", "-100", "-50", "9223372036854775807"),
+ ("DECIMAL(38, 0)", "-99999999999999999999999999999999999999", "1", "2"),
Review Comment:
nit: This test fails on the near row first, not the far one like the comment
says.
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala:
##########
@@ -728,6 +748,111 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
Row(Seq(1, 6)) :: Nil)
}
+ test("forward MATCH_CONDITION with NULL array elements picks the smallest
right row") {
Review Comment:
For this and 2 cases below: could we include a right key equal to the left
key to distinguish <= from < properly?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -189,7 +177,9 @@ private[joins] class SortMergeAsOfJoinScanner(
}
private val boundAsOfCond = bindReference(asOfCondition, joinedOutput)
- private val boundOrderExpr = bindReference(orderExpression, joinedOutput)
+ // Lazy, since only Nearest reads the distance. Its type can have no
ordering, for example
+ // CalendarInterval from TIMESTAMP - TIMESTAMP with
spark.sql.legacy.interval.enabled.
Review Comment:
nit: Remove "Its type can have no ordering, for example CalendarInterval
from TIMESTAMP - TIMESTAMP with spark.sql.legacy.interval.enabled."
--
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]
