LukaZdravic commented on code in PR #59957:
URL: https://github.com/apache/spark/pull/59957#discussion_r4455905280
##########
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:
Fixed the naming!
--
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]
