cloud-fan commented on code in PR #59957:
URL: https://github.com/apache/spark/pull/59957#discussion_r4474035322
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1200,6 +1200,19 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val SORT_MERGE_AS_OF_JOIN_CURSOR_SCAN_ENABLED =
+ buildConf("spark.sql.join.sortMergeAsOfJoin.cursorScan.enabled")
+ .internal()
+ .doc("When true, the sort-merge AS-OF join reads the right side once
with a cursor if " +
+ "the join is backward or forward, has no residual join condition, and
its as-of " +
+ "condition compares the two sort keys, of the same type. Other joins,
and joins on " +
+ "STRUCT, ARRAY, TIME or nanosecond timestamp keys, buffer each
right-side group and " +
Review Comment:
**Nit (P3):** This list, and the matching one in the `supportsCursorKey`
comment, now names TIME and nanosecond timestamps, but the allowlist's `case _
=> false` also rejects user-defined types. A UDT whose `sqlType` is orderable
is a valid MATCH_CONDITION operand (`OrderUtils.isOrderable` accepts it), so a
backward or forward join on the same UDT on both sides meets every condition
this doc lists and still keeps the buffered scan. Results don't change, but the
doc reads as if such joins use the cursor scan. Please add user-defined types
to both lists, or describe the supported key types positively in both places.
As far as I can tell, UDT is the only orderable key type that is neither
accepted nor listed: CHAR and VARCHAR extend StringType, and VARIANT, GEOMETRY,
GEOGRAPHY and MAP aren't orderable.
This extends an [existing
discussion](https://github.com/apache/spark/pull/59957#discussion_r4446870890).
**Verification:**
- **Inspection:** The cursor-scan config doc and the eligibility comment
name exactly the orderable key types the allowlist rejects, including
user-defined types, or exactly the types it accepts.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/AsOfJoinBenchmark.scala:
##########
@@ -75,9 +77,19 @@ object AsOfJoinBenchmark extends SqlBasedBenchmark {
}
}
- benchmark.addCase("Sort-merge AS-OF join") { _ =>
+ benchmark.addCase("Sort-merge AS-OF join, buffered scan") { _ =>
Review Comment:
**Nit (P3):** This renames the case and adds a cursor-scan case, here and in
the no-equi-key benchmark, and the right `ts` is now BIGINT, but
`AsOfJoinBenchmark-results.txt` and `AsOfJoinBenchmark-jdk25-results.txt`
weren't regenerated. They still list only the old 'Sort-merge AS-OF join' rows,
measured on DOUBLE right keys. So the cursor scan, which is on by default, has
no committed measurement, and the numbers in the PR description can't be
checked against the repository. Could you regenerate both files, for example
with the benchmark GitHub Actions workflow on your fork, and commit them?
This extends an [existing
discussion](https://github.com/apache/spark/pull/59957#issuecomment-6613192265).
**Verification:**
- **Inspection:** The committed AS-OF join benchmark results contain exactly
the cases the benchmark now defines, under their current names, and were
generated from the current benchmark code on the standard benchmark environment
for each JDK.
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala:
##########
@@ -1393,4 +1434,407 @@ 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 the STRING side
to the " +
+ "comparison type") {
+ Seq(true, false).foreach { ansi =>
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansi.toString) {
+ Seq(">=" -> 3, "<=" -> 7).foreach { case (op, expected) =>
+ withClue(s"ansi=$ansi, $op: ") {
+ checkCursorScan(
+ sql(
+ s"""
+ |SELECT q.n
+ |FROM VALUES ('5') AS t(s)
+ |ASOF JOIN VALUES (3), (7) AS q(n)
+ | MATCH_CONDITION (t.s $op q.n)
+ |""".stripMargin),
+ Row(expected) :: Nil)
+ }
+ }
+ }
+ }
+ }
+
+ test("cursor scan: a left row whose key has no right rows never evaluates
its as-of side") {
+ // Under ANSI, 'bad' cannot be cast to a number and Long.MaxValue * 1000
overflows. The
+ // buffered scan evaluates the as-of condition only on right rows with the
same key, so key
+ // B never fails the query. The cursor scan must not evaluate the left
side of key B either.
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ Seq(
+ ("('A', '1'), ('B', 'bad'), ('C', '3')",
+ "('A', 0, 'a0'), ('A', 2, 'a2'), ('C', 2, 'c2'), ('C', 4, 'c4')",
"t.v"),
+ ("('A', 1L), ('B', 9223372036854775807L), ('C', 3L)",
+ "('A', 0L, 'a0'), ('A', 2000L, 'a2'), ('C', 2000L, 'c2'), ('C',
4000L, 'c4')",
+ "t.v * 1000")
+ ).foreach { case (leftRows, rightRows, leftSide) =>
+ for {
+ (op, tagA, tagC) <- Seq((">=", "a0", "c2"), ("<=", "a2", "c4"))
+ (joinType, unmatched) <- Seq("LEFT" -> Seq(Row("B", null)), "INNER"
-> Nil)
+ } {
+ withClue(s"$leftSide $op q.v, $joinType: ") {
+ checkCursorScan(
+ sql(
+ s"""
+ |SELECT t.k, q.tag
+ |FROM VALUES $leftRows AS t(k, v)
+ |$joinType ASOF JOIN VALUES $rightRows AS q(k, v, tag)
+ | MATCH_CONDITION ($leftSide $op q.v) ON t.k = q.k
+ |""".stripMargin),
+ Seq(Row("A", tagA), Row("C", tagC)) ++ unmatched)
+ }
+ }
+ }
+ }
+ }
+
+ test("cursor scan: a failing as-of side fails the query when its key has
right rows") {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ Seq(
+ ("('A', 'bad')", "('A', 1)", "t.v", "CAST_INVALID_INPUT"),
Review Comment:
**Nit (P3):** This test only uses right rows with non-NULL as-of keys, and
the no-right-rows test above covers keys with no right rows at all. Neither
covers a key whose only right rows have a NULL as-of key, which
`startCursorGroup` deliberately counts as a right row so that the cursor
evaluates the left side just as the buffered scan does. If that rule regressed,
for example by checking the right as-of key before counting the row, then under
ANSI `('A', 'bad')` against `('A', CAST(NULL AS INT))` would come back
null-padded from the cursor scan (or be dropped for INNER) while the buffered
scan throws `CAST_INVALID_INPUT`, and no test would fail. Could you add cases
here whose right rows for the key have a typed NULL as-of key, such as `('A',
CAST(NULL AS INT))` and `('A', CAST(NULL AS BIGINT))`, for both operators?
This extends an [existing
discussion](https://github.com/apache/spark/pull/59957#discussion_r4446870878).
**Verification:**
- **Regression:** Under ANSI, with the cursor scan off and then on, backward
and forward joins of a left row whose as-of side fails at runtime, against
right rows with the same key whose as-of keys are all NULL, both fail with the
same error condition, for an invalid cast and for an overflow.
##########
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:
Thanks, this fixes it. The cursor now returns before it evaluates the left
as-of side when the key has no right rows, and the new ANSI tests compare both
scans for an invalid cast and an overflow. This review raises one more test
case: a key whose right rows all have NULL as-of keys.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4446870878","thread_id":"inline:4446870878","verdict_sha256":"15d852c5f334851eeedc66e635c5985339075ac76c136f07b93dc4aa44c7de6e"}
-->
##########
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:
Thanks, the comment, the config doc and the test name now name TIME and
nanosecond timestamps. This review raises one remaining gap in those lists:
user-defined-type keys.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4446870890","thread_id":"inline:4446870890","verdict_sha256":"15d852c5f334851eeedc66e635c5985339075ac76c136f07b93dc4aa44c7de6e"}
-->
--
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]
