cloud-fan commented on code in PR #59957:
URL: https://github.com/apache/spark/pull/59957#discussion_r4446870878
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -434,6 +464,63 @@ private[joins] class SortMergeAsOfJoinScanner(
null
}
+ /**
+ * Finds the best right row for `leftRow` by moving a cursor over the right
input, or null.
+ *
+ * Left rows of a group come in ascending as-of key order. So a right row
that passes for one
+ * left row also passes for the next (backward), and a row that fails stays
failed (forward).
+ * The cursor never steps back: backward stops at the first failing row and
keeps it for the
+ * next left row, and forward drops each failing row, since no later left
row can match it.
+ */
+ private def findWithCursor(leftRow: InternalRow, leftKey: UnsafeRow):
InternalRow = {
+ equiKeyOrdering match {
+ case Some(ordering)
+ if cursorGroupKey == null || ordering.compare(leftKey,
cursorGroupKey) != 0 =>
+ cursorGroupKey = leftKey
+ cursorBest = null
+ case _ =>
+ }
+ val (leftAsOfKey, rightAsOfKey) = cursorKeys
+ // A NULL left as-of key matches nothing. It must not move the cursor, or
forward would drop
+ // right rows that later left rows need.
+ if (leftAsOfKey.eval(leftRow) == null) return null
Review Comment:
**Blocking (P1):** `findWithCursor` evaluates the left as-of key here (the
NULL check) for every left row, before it knows whether the right side has any
row with that equi-key. The buffered scan evaluates the as-of condition, and
with it the left operand, only while it iterates a non-empty right group. The
left sort evaluates the as-of key only when two rows tie on the equi-keys. So
under ANSI, a MATCH_CONDITION operand that fails at runtime, such as
`CAST('bad' AS BIGINT)` from STRING-vs-INT coercion or an overflowing `t.k *
1000`, used to be harmless on a left row whose key has no right rows and no
other left row in its partition. Now it fails the query.
I reproduced this at the PR head with ANSI on:
```sql
SELECT * FROM VALUES ('A', 'bad') AS t(sym, s)
LEFT ASOF JOIN VALUES ('B', 1) AS q(sym, n)
MATCH_CONDITION (t.s >= q.n) ON t.sym = q.sym
```
With `spark.sql.join.sortMergeAsOfJoin.cursorScan.enabled=false`, this
returns `[A, bad, null]`, and the INNER form returns no rows. With the default
cursor scan, both fail with `CAST_INVALID_INPUT`. The forward form `t.s <= q.n`
fails the same way, and `t.k * 1000 >= q.k` with `t.k = Long.MaxValue` fails
with `ARITHMETIC_OVERFLOW`. Adding a second left row `('C', '3')` that matches
`('C', 2)` doesn't help: the cursor run still fails the whole query. With ANSI
off, or with a castable value, the two runs agree.
Suggested fix: before evaluating anything on the left row, check that the
current equi-key group has right rows. Track a per-group flag for whether the
cursor has seen a right row with this key, counting rows whose as-of key is
NULL, and skip earlier-group rows without evaluating either sort key. If the
group is empty, return no match without evaluating the left key. Otherwise,
keep the current NULL check, the forward reuse check and the cursor loop.
Please also add ANSI tests that run a failing operand with the cursor scan off
and on.
**Verification:**
- **Regression:** Under ANSI, with the cursor scan off and then on, backward
and forward LEFT and INNER ASOF JOINs with an equi-key, where a left row is the
only one with its key, has no matching right rows, and has an operand that
fails at runtime (a STRING operand that cannot be cast for a STRING-versus-INT
comparison, and separately an arithmetic operand that overflows), return the
same rows in both runs, including when other left rows in the same query do
have matches, and the cursor run reports that it used the cursor scan.
- **Behavior:** Under ANSI, when a left row with a failing operand has right
rows with its equi-key, the query fails with the same error condition with the
cursor scan off and on.
- **Regression:** The existing cursor-scan checks still pass: equal results
with the scan off and on, no spill at a one-row spill threshold, and the
per-left-row bound on condition evaluations, including joins without equi-keys
and group transitions in which a key has no right rows.
##########
sql/core/src/test/scala/org/apache/spark/sql/SortMergeAsOfJoinSuite.scala:
##########
@@ -78,6 +78,8 @@ class SortMergeAsOfJoinSuite extends QueryTest
assert(op.get.metrics("spillSize").value > 0,
s"Expected the right-side buffer to spill (spillSize > 0), " +
s"got ${op.get.metrics("spillSize").value}")
+ assert(op.get.metrics("numAsOfCursorTasks").value == 0,
Review Comment:
**Nit (P3):** This is the only assertion on the new metric in this suite,
and it checks `== 0` for the cursor-off spill tests. The PR description says
the existing DataFrame `joinAsOf` tests now run with the cursor scan on, and
they do today: a same-typed DataFrame join reports cursor scan tasks when I run
it locally. But nothing would fail if DataFrame-built joins stopped qualifying.
The DataFrame path builds its condition from the raw operands in
`makeAsOfCond`, separately from the sort keys. So the operand checks in
`supportsCursorScan` are a separate eligibility point that the SQL-only
`checkCursorScan` tests don't cover. Results would stay correct on the buffered
scan, so such a regression would go unnoticed. Consider asserting
`numAsOfCursorTasks > 0` (and no spill) in a backward and a forward default-on
`joinAsOf` test, and in the test that puts the right side first.
**Verification:**
- **Regression:** With the cursor scan on, eligible DataFrame joinAsOf
queries (a backward join, a forward join, and one whose comparison is rewritten
to put the right side first) return the expected rows and report that they used
the cursor scan with no spill, and these checks fail if DataFrame-built
conditions stop qualifying.
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala:
##########
@@ -1393,4 +1430,283 @@ class AsOfJoinSortMergeSQLSuite extends QueryTest
|""".stripMargin),
Row(Seq(Row(1, 2))) :: Nil)
}
+
+ test("cursor scan: backward match moves forward across trades and keys") {
+ // B has no quotes, C has quotes but no trades, and NULL keys never match.
A best row kept
+ // from key A would give B a match.
+ checkCursorScan(
+ sql(
+ """
+ |SELECT t.sym, t.ts, q.tag
+ |FROM VALUES ('A', 2), ('A', 6), ('A', 7), ('B', 4), ('D', 5),
(NULL, 3) AS t(sym, ts)
+ |LEFT ASOF JOIN VALUES ('A', 1, 'a1'), ('A', 3, 'a3'), ('A', 5,
'a5'), ('A', 8, 'a8'),
+ | ('C', 1, 'c1'), ('D', 2, 'd2'), ('D', 6, 'd6'), (NULL, 1, 'n1')
AS q(sym, ts, tag)
+ | MATCH_CONDITION (t.ts >= q.ts)
+ | ON t.sym = q.sym
+ |""".stripMargin),
+ Row("A", 2, "a1") :: Row("A", 6, "a5") :: Row("A", 7, "a5") :: Row("B",
4, null) ::
+ Row("D", 5, "d2") :: Row(null, 3, null) :: Nil)
+ }
+
+ test("cursor scan: forward match is kept while later trades still pass it") {
+ // A3 reuses q4, A6 must drop it, and B1 must not reuse q7 from key A.
+ checkCursorScan(
+ sql(
+ """
+ |SELECT t.sym, t.ts, q.tag
+ |FROM VALUES ('A', 2), ('A', 3), ('A', 6), ('B', 1), ('C', 9) AS
t(sym, ts)
+ |LEFT ASOF JOIN VALUES ('A', 1, 'q1'), ('A', 4, 'q4'), ('A', 7,
'q7'),
+ | ('B', 0, 'b0'), ('B', 2, 'b2'), ('C', 5, 'c5') AS q(sym, ts, tag)
+ | MATCH_CONDITION (t.ts <= q.ts)
+ | ON t.sym = q.sym
+ |""".stripMargin),
+ Row("A", 2, "q4") :: Row("A", 3, "q4") :: Row("A", 6, "q7") :: Row("B",
1, "b2") ::
+ Row("C", 9, null) :: Nil)
+ }
+
+ test("cursor scan: strict and non-strict MATCH_CONDITION ties") {
+ // The second of two equal trades must see the same right rows as the
first.
+ Seq(">=" -> "q5", ">" -> "q3", "<=" -> "q5", "<" -> "q7").foreach { case
(op, tag) =>
+ withClue(s"$op: ") {
+ checkCursorScan(
+ sql(
+ s"""
+ |SELECT t.id, q.tag
+ |FROM VALUES (1, 5), (2, 5) AS t(id, ts)
+ |ASOF JOIN VALUES (3, 'q3'), (5, 'q5'), (7, 'q7') AS q(ts, tag)
+ | MATCH_CONDITION (t.ts $op q.ts)
+ |""".stripMargin),
+ Row(1, tag) :: Row(2, tag) :: Nil)
+ }
+ }
+ }
+
+ test("cursor scan: NULL as-of keys never match and do not move the cursor") {
+ // The NULL trade sorts first. If it moved a forward cursor, trade 2 would
lose q3.
+ checkCursorScan(
+ sql(
+ """
+ |SELECT t.id, q.tag
+ |FROM VALUES (1, NULL), (2, 2) AS t(id, ts)
+ |LEFT ASOF JOIN VALUES (NULL, 'qn'), (3, 'q3') AS q(ts, tag)
+ | MATCH_CONDITION (t.ts <= q.ts)
+ |""".stripMargin),
+ Row(1, null) :: Row(2, "q3") :: Nil)
+ // The NULL quote sorts first. If it stopped a backward cursor, trade 2
would get no match.
+ checkCursorScan(
+ sql(
+ """
+ |SELECT t.id, q.tag
+ |FROM VALUES (1, NULL), (2, 2) AS t(id, ts)
+ |LEFT ASOF JOIN VALUES (NULL, 'qn'), (1, 'q1') AS q(ts, tag)
+ | MATCH_CONDITION (t.ts >= q.ts)
+ |""".stripMargin),
+ Row(1, null) :: Row(2, "q1") :: Nil)
+ }
+
+ test("cursor scan: scalar MATCH_CONDITION types") {
+ val numbers = Seq("TINYINT", "SMALLINT", "INT", "BIGINT", "FLOAT",
"DOUBLE", "DECIMAL(38, 18)")
+ .map(t => (t, s"CAST(1 AS $t)", s"CAST(2 AS $t)", s"CAST(3 AS $t)"))
+ def lcase(value: String): String = s"'$value' COLLATE UTF8_LCASE"
+ val others = Seq(
+ ("STRING", "'a'", "'b'", "'c'"),
+ // Binary order is 'B' < 'a' < 'c', but UTF8_LCASE order is 'a' < 'B' <
'c'.
+ ("STRING COLLATE UTF8_LCASE", lcase("a"), lcase("B"), lcase("c")),
+ ("BINARY", "X'01'", "X'02'", "X'03'"),
+ ("DATE", "DATE'2026-01-01'", "DATE'2026-01-02'", "DATE'2026-01-03'"),
+ ("TIMESTAMP", "TIMESTAMP'2026-01-01 00:00:01'", "TIMESTAMP'2026-01-01
00:00:02'",
+ "TIMESTAMP'2026-01-01 00:00:03'"),
+ ("TIMESTAMP_NTZ", "TIMESTAMP_NTZ'2026-01-01 00:00:01'",
+ "TIMESTAMP_NTZ'2026-01-01 00:00:02'", "TIMESTAMP_NTZ'2026-01-01
00:00:03'"),
+ ("INTERVAL YEAR TO MONTH", "INTERVAL '1-1' YEAR TO MONTH", "INTERVAL
'1-2' YEAR TO MONTH",
+ "INTERVAL '1-3' YEAR TO MONTH"),
+ ("INTERVAL DAY TO SECOND", "INTERVAL '1 00:00:01' DAY TO SECOND",
+ "INTERVAL '1 00:00:02' DAY TO SECOND", "INTERVAL '1 00:00:03' DAY TO
SECOND"))
+ for {
+ (typeName, before, at, after) <- numbers ++ others
+ (op, tag) <- Seq(">=" -> "before", "<=" -> "after")
+ } {
+ withClue(s"$typeName $op: ") {
+ checkCursorScan(
+ sql(
+ s"""
+ |SELECT q.tag
+ |FROM VALUES ($at) AS t(k)
+ |ASOF JOIN VALUES ($before, 'before'), ($after, 'after') AS
q(k, tag)
+ | MATCH_CONDITION (t.k $op q.k)
+ |""".stripMargin),
+ Row(tag) :: Nil)
+ }
+ }
+ }
+
+ test("cursor scan: DOUBLE NaN and negative zero MATCH_CONDITION") {
+ // Spark orders -0.0 equal to 0.0, and NaN above every other value.
+ Seq(">=", "<=").foreach { op =>
+ withClue(s"$op: ") {
+ checkCursorScan(
+ sql(
+ s"""
+ |SELECT t.id, q.tag
+ |FROM VALUES (1, 0.0D), (2, CAST('NaN' AS DOUBLE)) AS t(id, k)
+ |ASOF JOIN VALUES (-0.0D, 'negative zero'), (1.0D, 'one'),
+ | (CAST('NaN' AS DOUBLE), 'nan') AS q(k, tag)
+ | MATCH_CONDITION (t.k $op q.k)
+ |""".stripMargin),
+ Row(1, "negative zero") :: Row(2, "nan") :: Nil)
+ }
+ }
+ }
+
+ test("cursor scan: MATCH_CONDITION sides that are expressions of a column") {
+ // Each side sorts on its whole operand, so the cursor walks the operand
values in order,
+ // even when the operand reverses the column order.
+ Seq(
+ // (match condition, backward rows, forward rows)
+ ("t.k + 10 OP q.k", Seq(Row(1, 8), Row(5, 12)), Seq(Row(1, 12), Row(5,
20))),
+ ("-t.k OP q.k", Seq(Row(1, -3), Row(5, -6)), Seq(Row(1, 0), Row(5, -3))),
+ ("t.k OP -q.k", Seq(Row(1, 0), Row(5, -3)), Seq(Row(1, -3), Row(5, -6)))
+ ).foreach { case (condition, backwardRows, forwardRows) =>
+ Seq(">=" -> backwardRows, "<=" -> forwardRows).foreach { case (op,
expected) =>
+ val matchCondition = condition.replace("OP", op)
+ withClue(s"$matchCondition: ") {
+ checkCursorScan(
+ sql(
+ s"""
+ |SELECT t.k, q.k
+ |FROM VALUES (1), (5) AS t(k)
+ |LEFT ASOF JOIN VALUES (-6), (-3), (0), (3), (6), (8), (12),
(20) AS q(k)
+ | MATCH_CONDITION ($matchCondition)
+ |""".stripMargin),
+ expected)
+ }
+ }
+ }
+ }
+
+ test("cursor scan: STRING against INT MATCH_CONDITION casts both sides") {
Review Comment:
**Nit (P3):** With ANSI off, the common type for STRING vs INT is INT
(`TypeCoercion.findCommonTypeForBinaryComparison`), so only `t.s` is cast and
`q.n` stays a bare INT. Only the ANSI-on run casts both sides, to BIGINT. The
assertions are fine, but the name misstates what half of the runs check.
Something like "casts the STRING side to the comparison type" would hold in
both modes.
**Verification:**
- **Inspection:** The renamed test still runs both ANSI modes and both
comparison operators with unchanged assertions, and its name is true for each
mode.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -518,3 +605,54 @@ private[joins] class SortMergeAsOfJoinScanner(
if (lower.isNaN || upper.isNaN) 2 else if (lower.isInfinite ||
upper.isInfinite) 1 else 0
}
}
+
+private[execution] object SortMergeAsOfJoinScanner {
+
+ /**
+ * Whether the scan can read the right side once with a cursor, instead of
buffering each
+ * right-side group and scanning it again for every left row.
+ *
+ * This needs a backward or forward join with no residual condition, whose
as-of condition
+ * only compares the two sort keys. Both sides are sorted on exactly those
keys, so the right
+ * rows that pass the condition move forward as the left key grows. Anything
else, such as a
+ * tolerance, a tuple, or an operand cast that the sort key does not have,
keeps the buffer.
+ */
+ def supportsCursorScan(
+ direction: AsOfJoinDirection,
+ asOfCondition: Expression,
+ residualCondition: Option[Expression],
+ leftSortExprs: Seq[Expression],
+ rightSortExprs: Seq[Expression]): Boolean = {
+ // The operands as (lesser, greater).
+ val ordered = asOfCondition match {
+ case LessThan(lesser, greater) => Some((lesser, greater))
+ case LessThanOrEqual(lesser, greater) => Some((lesser, greater))
+ case GreaterThan(greater, lesser) => Some((lesser, greater))
+ case GreaterThanOrEqual(greater, lesser) => Some((lesser, greater))
+ case _ => None
+ }
+ // The operands as (left, right): backward needs left >= right, forward
left <= right.
+ val operands = (direction, ordered) match {
+ case (Backward, Some((lesser, greater))) => Some((greater, lesser))
+ case (Forward, Some((lesser, greater))) => Some((lesser, greater))
+ case _ => None
+ }
+ (leftSortExprs, rightSortExprs, operands) match {
+ case (Seq(leftKey), Seq(rightKey), Some((leftOperand, rightOperand))) =>
+ residualCondition.isEmpty &&
+ leftOperand.semanticEquals(leftKey) &&
+ rightOperand.semanticEquals(rightKey) &&
+ leftKey.dataType == rightKey.dataType &&
+ supportsCursorKey(leftKey.dataType)
+ case _ => false
+ }
+ }
+
+ /** Scalar key types only for now. STRUCT and ARRAY keys keep the buffered
scan. */
Review Comment:
**Nit (P3):** This comment, and the new config doc ("compares the two scalar
sort keys"), describe the rule as any scalar key except STRUCT and ARRAY. But
the allowlist below falls through to `false` for other orderable scalar types:
TIME, and the nanosecond `TIMESTAMP_LTZ(p)` / `TIMESTAMP_NTZ(p)` types. Those
are valid MATCH_CONDITION operands, and the nearest path in this file supports
nanosecond keys. `SortMergeAsOfJoinScannerSuite` even asserts that the
nanosecond types are rejected, in a test named "cursor scan accepts scalar sort
key types only". Results are unaffected, but anyone reading the docs would
expect those joins to use the cursor scan. Please name the supported types (or
the excluded scalar types, and why they're excluded) in this comment, the
config doc and that test name, or widen the allowlist to match.
**Verification:**
- **Inspection:** The eligibility comment, the config doc and the test name
each name TIME and nanosecond-precision timestamp keys among the types that
keep the buffered scan, and they agree with the allowlist in the code.
- **Behavior:** The key-type acceptance checks still accept every listed
type and reject STRUCT, ARRAY and nanosecond-precision timestamp keys unchanged.
--
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]
