This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 825b77d9f2cd [SPARK-59829][SQL] Make the forward ASOF scan keep the
first match like backwards scan
825b77d9f2cd is described below
commit 825b77d9f2cd90798e84e4a43be837a46e6ddc02
Author: Luka Zdravić <[email protected]>
AuthorDate: Thu Oct 1 08:42:42 2026 +0000
[SPARK-59829][SQL] Make the forward ASOF scan keep the first match like
backwards scan
### What changes were proposed in this pull request?
Forward ASOF joins (`t.k <= r.k` or `t.k < r.k`) in `SortMergeAsOfJoinExec`
now keep the first right row that matches. Before, they reused the Nearest
loop, which computes `right - left` for every candidate and keeps the smallest.
The right rows are sorted ascending, so the first match is always the
closest one. This mirrors Backward, which keeps the last match. The operator
now reads the direction (Backward, Forward, Nearest) from the as-of condition.
Nearest, and any condition shape it does not recognize, keep the old distance
loop.
Forward also stops once the as-of condition turns false after a match (past
the tolerance bound). The SQL reference now says how NULL parts of ARRAY and
STRUCT operands sort.
### Why are the changes needed?
The subtraction overflows when values are far apart, even though the answer
never needs it:
```sql
SET spark.sql.join.asofJoin.enabled=true;
SELECT r.tag
FROM VALUES (-100) AS t(k)
ASOF JOIN VALUES (-50, 'near'), (2147483647, 'far') AS r(k, tag)
MATCH_CONDITION (t.k <= r.k);
```
- ANSI on: fails with `ARITHMETIC_OVERFLOW`.
- ANSI off: `2147483647 - (-100)` wraps to a negative number, so it returns
`far`.
- Expected, and the result after this PR: `near`.
The same bug hits BIGINT, DECIMAL(38, 0), and both interval types, as an
error or a wrong result.
The same distance bug also picked the wrong row for NULL array elements,
NULL struct fields, and a leading STRING struct field, because the distance
ranks those values wrongly. For example, left `ARRAY(NULL)` with right rows
`ARRAY(NULL, NULL)` and `ARRAY(5)` returned `ARRAY(5)`, but `ARRAY(NULL, NULL)`
sorts first.
With `spark.sql.legacy.interval.enabled=true`, `TIMESTAMP - TIMESTAMP` is a
CalendarInterval, which has no ordering. The operator built that ordering for
every direction, so these joins failed with `DATATYPE_CANNOT_ORDER`. Now only
Nearest builds it.
### Does this PR introduce _any_ user-facing change?
Yes. A forward ASOF join whose `right - left` overflows used to fail with
ARITHMETIC_OVERFLOW (ANSI on) or return a farther row (ANSI off). It now
returns the closest row, for example `near` in the example above.
### How was this patch tested?
New tests in `AsOfJoinSortMergeSQLSuite`:
- One overflow test per type (INT, BIGINT, DECIMAL(38, 0), INTERVAL YEAR TO
MONTH, INTERVAL DAY TO SECOND). Each runs `<=` and `<`, with ANSI on and off,
and expects `near`.
- A forward test where the closest row fails an extra `ON` filter, so the
join must skip it and return the next row.
- Forward tests with NULL array elements, a NULL struct field, and a
leading STRING struct field, for `<=` and `<`.
- TIMESTAMP with legacy intervals, for all four operators.
- In `SortMergeAsOfJoinSuite`, a forward join with tolerance where the ON
condition rejects every row within the tolerance.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #59102 from LukaZdravic/asof-forward-first-match.
Authored-by: Luka Zdravić <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
(cherry picked from commit 5b4a2cc4b5e084069d7d6bd9e199c04650f9353d)
Signed-off-by: Uros Bojanic <[email protected]>
---
docs/sql-ref-syntax-qry-select-asof-join.md | 6 +-
.../execution/joins/SortMergeAsOfJoinExec.scala | 173 +++++++++++----------
.../spark/sql/AsOfJoinSortMergeSQLSuite.scala | 130 ++++++++++++++++
.../apache/spark/sql/SortMergeAsOfJoinSuite.scala | 43 +++++
4 files changed, 264 insertions(+), 88 deletions(-)
diff --git a/docs/sql-ref-syntax-qry-select-asof-join.md
b/docs/sql-ref-syntax-qry-select-asof-join.md
index ebbffd1a9005..f4e11a25a485 100644
--- a/docs/sql-ref-syntax-qry-select-asof-join.md
+++ b/docs/sql-ref-syntax-qry-select-asof-join.md
@@ -119,9 +119,11 @@ comparison_operator
* *L* `<=` *R*: the smallest *R* not less than *L* (first-following).
* *L* `<` *R*: the smallest *R* strictly greater than *L* (first
strictly-following).
-* **NULL.** A `NULL` in either operand never satisfies the `MATCH_CONDITION`
+* **NULL.** A `NULL` operand on either side never satisfies the
`MATCH_CONDITION`
comparison. Left rows whose operand is `NULL` are dropped under `INNER ASOF`
and
- retained with `NULL` right-side columns under `LEFT ASOF`.
+ retained with `NULL` right-side columns under `LEFT ASOF`. A `NULL` element
of an
+ `ARRAY` operand, or a `NULL` field of a `STRUCT` operand, does not block a
match. It
+ sorts before any non-`NULL` value, so `ARRAY(NULL) <= ARRAY(5)` is true.
* **Ties.** If multiple right rows share the same *R* value that is closest to
*L*
after applying `ON`/`USING`, the choice among tied rows is not
deterministic. Add a
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala
index 22e4f5580cf5..4050da08793d 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala
@@ -53,7 +53,7 @@ case class SortMergeAsOfJoinExec(
condition: Option[Expression],
left: SparkPlan,
right: SparkPlan,
- isSkewJoin: Boolean = false) extends ShuffledJoin {
+ isSkewJoin: Boolean = false) extends ShuffledJoin with PredicateHelper {
require(leftSortExprs.nonEmpty && rightSortExprs.nonEmpty &&
leftSortExprs.length == rightSortExprs.length,
@@ -92,34 +92,24 @@ 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
+ // The first `left op right` comparison gives the direction. Nearest has
none.
+ 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
}
- isBackwardPredicate(condition)
- }
protected override def doExecute(): RDD[InternalRow] = {
val numOutputRows = longMetric("numOutputRows")
val spillSize = longMetric("spillSize")
- val isBackward = isBackwardJoin
+ val direction = asOfDirection
val inMemoryThreshold = conf.sortMergeJoinExecBufferInMemoryThreshold
val sizeInBytesSpillThreshold =
conf.sortMergeJoinExecBufferSpillSizeThreshold
val spillThreshold = conf.sortMergeJoinExecBufferSpillThreshold
@@ -131,7 +121,7 @@ case class SortMergeAsOfJoinExec(
leftKeys, rightKeys,
asOfCondition, orderExpression,
joinType, condition,
- numOutputRows, spillSize, isBackward,
+ numOutputRows, spillSize, direction,
inMemoryThreshold, sizeInBytesSpillThreshold, spillThreshold
)
TaskContext.get().addTaskCompletionListener[Unit](_ => scanner.close())
@@ -150,14 +140,10 @@ case class SortMergeAsOfJoinExec(
*
* Both inputs are sorted by (equi-keys, as-of key) ascending. For each
* left row within an equi-key group, we scan the right-side buffer
- * forward to find the best match.
- *
- * For Backward joins (left.t >= right.t), the forward scan keeps the
- * 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
- * early termination (stop when distance starts increasing).
+ * forward to find the best match:
+ * - Backward (left.t >= right.t): the last satisfying row is the closest.
+ * - Forward (left.t <= right.t): the first satisfying row is the closest.
+ * - Nearest: the satisfying row with the smallest distance.
*/
private[joins] class SortMergeAsOfJoinScanner(
leftIter: Iterator[InternalRow],
@@ -172,7 +158,7 @@ private[joins] class SortMergeAsOfJoinScanner(
residualCondition: Option[Expression],
numOutputRows: SQLMetric,
spillSize: SQLMetric,
- isBackwardJoin: Boolean,
+ direction: AsOfJoinDirection,
inMemoryThreshold: Int,
sizeInBytesSpillThreshold: Long,
spillThreshold: Int) {
@@ -189,7 +175,10 @@ private[joins] class SortMergeAsOfJoinScanner(
}
private val boundAsOfCond = bindReference(asOfCondition, joinedOutput)
- private val boundOrderExpr = bindReference(orderExpression, joinedOutput)
+ // Lazy, since only Nearest reads the distance, and its type may have no
ordering.
+ private lazy val boundOrderExpr = bindReference(orderExpression,
joinedOutput)
+ private lazy val distanceOrdering =
+ TypeUtils.getInterpretedOrdering(orderExpression.dataType)
private val boundResidualCond =
residualCondition.map(bindReference(_, joinedOutput))
@@ -207,9 +196,6 @@ private[joins] class SortMergeAsOfJoinScanner(
private val leftKeyProj = UnsafeProjection.create(leftKeys, leftOutput)
private val rightKeyProj = UnsafeProjection.create(rightKeys, rightOutput)
- private val distanceOrdering =
- TypeUtils.getInterpretedOrdering(orderExpression.dataType)
-
// Materialize an all-null right row as UnsafeRow. GenericInternalRow cannot
be
// passed through identity UnsafeProjection when right columns are NOT NULL.
private val nullRightRow: InternalRow = {
@@ -357,23 +343,11 @@ private[joins] class SortMergeAsOfJoinScanner(
rightDone = true
}
- /**
- * Find the best matching right row using forward-only scan.
- *
- * For Backward joins: keeps the last as-of-satisfying row (since the
- * buffer is sorted ascending, the last satisfying row has the largest
- * 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
- * (stop when distance starts increasing past the minimum).
- */
- private def findBestInGroup(leftRow: InternalRow): InternalRow = {
- if (isBackwardJoin) {
- findBestBackwardForward(leftRow)
- } else {
- findBestForwardNearest(leftRow)
- }
+ /** Finds the best right row in the current group for `leftRow`, or null if
none passes. */
+ private def findBestInGroup(leftRow: InternalRow): InternalRow = direction
match {
+ case Backward => findLastBackward(leftRow)
+ case Forward => findFirstForward(leftRow)
+ case Nearest => findNearest(leftRow)
}
/**
@@ -388,35 +362,39 @@ private[joins] class SortMergeAsOfJoinScanner(
if (needsCopy) rightRow.copy() else rightRow
}
+ /** Whether `cond` is TRUE on the current joined row. FALSE and NULL both
fail. */
+ private def holds(cond: Expression): Boolean = {
+ val result = cond.eval(joinedRow)
+ result != null && result.asInstanceOf[Boolean]
+ }
+
+ /** Whether the residual ON condition, if any, holds on the current joined
row. */
+ private def residualHolds: Boolean = boundResidualCond.forall(holds)
+
/**
- * Forward scan for Backward joins: last-match-wins.
- * Buffer is sorted ascending by as-of key. For left.t >= right.t,
- * as-of condition is monotone: true for right.t <= left.t, then false.
- * The last satisfying row is the closest match.
+ * Backward joins keep the last row that passes the as-of and residual
conditions.
+ * Once the as-of condition has been true, the scan stops at the first row
that fails
+ * it, as later sorted rows cannot satisfy it.
*/
- private def findBestBackwardForward(leftRow: InternalRow): InternalRow = {
+ private def findLastBackward(leftRow: InternalRow): InternalRow = {
var bestMatch: InternalRow = null
val iter = rightGroupBuffer.generateIterator()
val needsCopy = rightGroupBuffer.isSpillBacked
+ var asOfSeen = false
joinedRow.withLeft(leftRow)
while (iter.hasNext) {
val rightRow = iter.next()
joinedRow.withRight(rightRow)
- val asOfSatisfied = boundAsOfCond.eval(joinedRow)
- if (asOfSatisfied != null && asOfSatisfied.asInstanceOf[Boolean]) {
- val residualSatisfied = boundResidualCond.forall { cond =>
- val result = cond.eval(joinedRow)
- result != null && result.asInstanceOf[Boolean]
- }
- if (residualSatisfied) {
+ if (holds(boundAsOfCond)) {
+ asOfSeen = true
+ if (residualHolds) {
// Last match wins (closest right.t to left.t)
bestMatch = retainMatch(rightRow, needsCopy)
}
- } else if (bestMatch != null) {
- // as-of condition transitioned true -> false (monotone for Backward).
- // No further rows can satisfy it.
+ } else if (asOfSeen) {
+ // At or past left.t, so no later row passes the as-of condition.
return bestMatch
}
}
@@ -424,37 +402,60 @@ 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 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.
*/
- private def findBestForwardNearest(leftRow: InternalRow): InternalRow = {
+ private def findFirstForward(leftRow: InternalRow): InternalRow = {
+ val iter = rightGroupBuffer.generateIterator()
+ val needsCopy = rightGroupBuffer.isSpillBacked
+ var asOfSeen = false
+
+ joinedRow.withLeft(leftRow)
+ while (iter.hasNext) {
+ val rightRow = iter.next()
+ joinedRow.withRight(rightRow)
+
+ if (holds(boundAsOfCond)) {
+ asOfSeen = true
+ if (residualHolds) {
+ return retainMatch(rightRow, needsCopy)
+ }
+ } 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 stops decreasing.
+ */
+ private def findNearest(leftRow: InternalRow): InternalRow = {
var bestMatch: InternalRow = null
var bestDistance: Any = null
val iter = rightGroupBuffer.generateIterator()
val needsCopy = rightGroupBuffer.isSpillBacked
+ // Read the lazy vals once, not once per row.
+ val orderExpr = boundOrderExpr
+ val ordering = distanceOrdering
joinedRow.withLeft(leftRow)
while (iter.hasNext) {
val rightRow = iter.next()
joinedRow.withRight(rightRow)
- val asOfSatisfied = boundAsOfCond.eval(joinedRow)
- if (asOfSatisfied != null && asOfSatisfied.asInstanceOf[Boolean]) {
- val residualSatisfied = boundResidualCond.forall { cond =>
- val result = cond.eval(joinedRow)
- result != null && result.asInstanceOf[Boolean]
- }
- if (residualSatisfied) {
- val distance = boundOrderExpr.eval(joinedRow)
+ if (holds(boundAsOfCond)) {
+ if (residualHolds) {
+ val distance = orderExpr.eval(joinedRow)
if (distance != null) {
- if (bestMatch == null || distanceOrdering.lt(distance,
bestDistance)) {
+ if (bestMatch == null || ordering.lt(distance, bestDistance)) {
bestMatch = retainMatch(rightRow, needsCopy)
bestDistance = distance
} else {
- // Distance is increasing past the minimum. For Forward,
- // the as-of condition guarantees no closer row exists
- // further right. For Nearest, distance is V-shaped so
- // once past the minimum no later row can beat it.
+ // The distance stopped decreasing. It is V-shaped, so no later
row is closer.
return bestMatch
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala
index 200800f08ab3..66a3786b2ecf 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala
@@ -204,6 +204,28 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
Row(180.15) :: Nil)
}
+ test("TIMESTAMP MATCH_CONDITION with legacy intervals") {
+ // The distance is then a CalendarInterval, which has no ordering. Only
Nearest reads it.
+ withSQLConf(SQLConf.LEGACY_INTERVAL_ENABLED.key -> "true") {
+ for ((op, tag) <- Seq(">=" -> "r10", ">" -> "r9", "<=" -> "r10", "<" ->
"r11")) {
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES (TIMESTAMP '2026-06-29 10:00:00') AS t(ts)
+ |ASOF JOIN VALUES
+ | (TIMESTAMP '2026-06-29 08:00:00', 'r8'),
+ | (TIMESTAMP '2026-06-29 09:00:00', 'r9'),
+ | (TIMESTAMP '2026-06-29 10:00:00', 'r10'),
+ | (TIMESTAMP '2026-06-29 11:00:00', 'r11'),
+ | (TIMESTAMP '2026-06-29 12:00:00', 'r12') AS r(ts, tag)
+ | MATCH_CONDITION (t.ts $op r.ts)
+ |""".stripMargin),
+ Row(tag) :: Nil)
+ }
+ }
+ }
+
test("DATE scalar MATCH_CONDITION") {
checkSortMergeAsOf(
sql(
@@ -807,6 +829,114 @@ 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]. [null] equals
the left row.
+ val nullInt = "CAST(NULL AS INT)"
+ Seq("<=" -> "n", "<" -> "nn").foreach { case (op, tag) =>
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES (ARRAY($nullInt)) AS t(a)
+ |ASOF JOIN VALUES
+ | (ARRAY($nullInt), 'n'),
+ | (ARRAY($nullInt, $nullInt), 'nn'),
+ | (ARRAY(5), 'five') AS r(a, tag)
+ | MATCH_CONDITION (t.a $op r.a)
+ |""".stripMargin),
+ Row(tag) :: Nil)
+ }
+ }
+
+ test("forward MATCH_CONDITION with a NULL struct field picks the smallest
right row") {
+ // A NULL field sorts first, so {null, 5} < {null, 7} < {0, 1}. {null, 5}
equals the left row.
+ val nullInt = "CAST(NULL AS INT)"
+ for {
+ (left, right) <- Seq("t.k" -> "r.k", "(t.k.f, t.k.seq)" -> "(r.k.f,
r.k.seq)")
+ (op, tag) <- Seq("<=" -> "s5", "<" -> "s7")
+ } {
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES (named_struct('f', $nullInt, 'seq', 5)) AS t(k)
+ |ASOF JOIN VALUES
+ | (named_struct('f', $nullInt, 'seq', 5), 's5'),
+ | (named_struct('f', $nullInt, 'seq', 7), 's7'),
+ | (named_struct('f', 0, 'seq', 1), 's1') AS r(k, tag)
+ | MATCH_CONDITION ($left $op $right)
+ |""".stripMargin),
+ Row(tag) :: Nil)
+ }
+ }
+
+ test("forward MATCH_CONDITION with a leading STRING field picks the smallest
right row") {
+ // {b, 5} equals the left row. {c, 1} < {d, 0}, but 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, tag) <- Seq("<=" -> "b5", "<" -> "c1")
+ } {
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES (named_struct('s', 'b', 'n', 5)) AS t(k)
+ |ASOF JOIN VALUES
+ | (named_struct('s', 'b', 'n', 5), 'b5'),
+ | (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(tag) :: 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)
+ }
+
+ // (right - left) overflows the type for the far row, and for DECIMAL for
both rows. So a
+ // distance-based scan throws or gives a wrong answer.
+ Seq(
+ ("INT", "-100", "-50", "2147483647"),
+ ("BIGINT", "-100", "-50", "9223372036854775807"),
+ ("DECIMAL(38, 0)", "-99999999999999999999999999999999999999", "1", "2"),
+ ("INTERVAL YEAR TO MONTH",
+ "INTERVAL '-100' MONTH", "INTERVAL '-50' MONTH", "INTERVAL '178956970-7'
YEAR TO MONTH"),
+ ("INTERVAL DAY TO SECOND",
+ "INTERVAL '-100' DAY", "INTERVAL '-50' DAY", "INTERVAL '106751991' DAY")
+ ).foreach { case (dataType, left, near, far) =>
+ test(s"forward $dataType MATCH_CONDITION picks the closest row when right
- left overflows") {
+ Seq(true, false).foreach { ansiEnabled =>
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString) {
+ Seq("<=", "<").foreach { op =>
+ checkSortMergeAsOf(
+ sql(
+ s"""
+ |SELECT r.tag
+ |FROM VALUES (CAST($left AS $dataType)) AS t(k)
+ |ASOF JOIN VALUES
+ | (CAST($near AS $dataType), 'near'),
+ | (CAST($far AS $dataType), 'far') AS r(k, tag)
+ | MATCH_CONDITION (t.k $op r.k)
+ |""".stripMargin),
+ Row("near") :: Nil)
+ }
+ }
+ }
+ }
+ }
+
test("ARRAY<STRUCT> MATCH_CONDITION with fields of different types") {
// Each element has an INT distance and an INTERVAL distance.
def element(first: String, time: String): String =
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/SortMergeAsOfJoinSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/SortMergeAsOfJoinSuite.scala
index 77cad8690d23..b7e13e848e19 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SortMergeAsOfJoinSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SortMergeAsOfJoinSuite.scala
@@ -685,6 +685,49 @@ class SortMergeAsOfJoinSuite extends QueryTest
)
}
+ test("forward join - residual rejects every row within the tolerance") {
+ // For left (5, 10), ts 6 is within the tolerance but fails amount >
threshold. ts 9 passes
+ // the residual but is past the tolerance, so it gets no match. Left (4,
30) matches ts 6.
+ val schema1 = StructType(
+ StructField("ts", IntegerType) ::
+ StructField("amount", IntegerType) :: Nil)
+ val schema2 = StructType(
+ StructField("ts", IntegerType) ::
+ StructField("threshold", IntegerType) :: Nil)
+ val df1 = spark.createDataFrame(List(Row(4, 30), Row(5, 10)).asJava,
schema1)
+ val df2 = spark.createDataFrame(List(Row(6, 20), Row(9, 1)).asJava,
schema2)
+ checkAnswer(
+ df1.joinAsOf(
+ df2, df1.col("ts"), df2.col("ts"),
+ joinExprs = df1.col("amount") > df2.col("threshold"),
+ joinType = "leftouter", tolerance = functions.lit(2),
+ allowExactMatches = true, direction = "forward"),
+ Seq(Row(4, 30, 6, 20), Row(5, 10, null, null)))
+ }
+
+ test("forward join - picks the closest row when right - left overflows") {
+ // Int.MaxValue - (-100) overflows INT, so a distance-based scan throws or
picks far.
+ val schema1 = StructType(StructField("ts", IntegerType) :: Nil)
+ val schema2 = StructType(
+ StructField("ts", IntegerType) ::
+ StructField("tag", StringType) :: Nil)
+ val df1 = spark.createDataFrame(List(Row(-100)).asJava, schema1)
+ val df2 = spark.createDataFrame(
+ List(Row(-50, "near"), Row(Int.MaxValue, "far")).asJava, schema2)
+ Seq(true, false).foreach { ansiEnabled =>
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString) {
+ Seq(true, false).foreach { allowExactMatches =>
+ checkAnswer(
+ df1.joinAsOf(
+ df2, df1.col("ts"), df2.col("ts"), usingColumns = Seq.empty,
+ joinType = "inner", tolerance = null,
+ allowExactMatches = allowExactMatches, direction = "forward"),
+ Seq(Row(-100, -50, "near")))
+ }
+ }
+ }
+ }
+
test("backward join - spill to disk") {
// Force spill by setting in-memory threshold to 1 row.
// Verifies that ExternalAppendOnlyUnsafeRowArray's spill path
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
