This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 2dfc84cd79ef [SPARK-56887][SQL][TEST][FOLLOWUP] Add spill tests for
forward, nearest, and left outer AS-OF joins
2dfc84cd79ef is described below
commit 2dfc84cd79efa5a7bd4177a908b22315a95f1fe0
Author: Kousuke Saruta <[email protected]>
AuthorDate: Fri Jun 26 09:03:21 2026 -0700
[SPARK-56887][SQL][TEST][FOLLOWUP] Add spill tests for forward, nearest,
and left outer AS-OF joins
### What changes were proposed in this pull request?
Add spill-path tests for `SortMergeAsOfJoinExec` directions and join types
not covered by the existing backward+inner spill test added in #56595:
- Forward join with spill
- Nearest join with spill
- Left outer join with spill (null-fill path)
All tests force `ExternalAppendOnlyUnsafeRowArray` to spill by setting
`spark.sql.sortMergeJoinExec.buffer.in.memory.threshold` and
`spark.sql.sortMergeJoinExec.buffer.spill.threshold` to 1.
### Why are the changes needed?
Requested in #56595 review: the spill test added there only covered the
Backward + Inner path (`findBestBackwardForward`). The Forward/Nearest path
(`findBestForwardNearest`) and the LeftOuter null-fill path have their own
`.copy()` over the reused `SpillableArrayIterator` row and were not exercised
under spill conditions.
### Does this PR introduce *any* user-facing change?
No.
### How was this patch tested?
New tests in `SortMergeAsOfJoinSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Kiro CLI / Claude
Closes #56770 from sarutak/asof-join-spill-tests.
Authored-by: Kousuke Saruta <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 2e34b45d1f17e85abbc77af4a36ec86c99fba09e)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../apache/spark/sql/SortMergeAsOfJoinSuite.scala | 103 +++++++++++++++++++++
1 file changed, 103 insertions(+)
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 14b5e7ef9297..547b2fa0dd23 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
@@ -670,4 +670,107 @@ class SortMergeAsOfJoinSuite extends QueryTest
)
}
}
+
+ test("forward join - spill to disk") {
+ withSQLConf(
+ SQLConf.SORT_MERGE_JOIN_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "1",
+ SQLConf.SORT_MERGE_JOIN_EXEC_BUFFER_SPILL_THRESHOLD.key -> "1") {
+ val (df1, df2) = prepareForAsOfJoin()
+ // No equi-key (bufferAllRight path)
+ checkAnswer(
+ df1.joinAsOf(
+ df2, df1.col("a"), df2.col("a"), usingColumns = Seq.empty,
+ joinType = "inner", tolerance = null,
+ allowExactMatches = true, direction = "forward"),
+ Seq(
+ Row(1, "x", "a", 1, "v", 1),
+ Row(5, "y", "b", 6, "y", 6)
+ )
+ )
+ // With equi-key (bufferRightGroup path) - use data with multiple
+ // right rows per group to actually exceed the spill threshold.
+ val schema1 = StructType(
+ StructField("grp", StringType) ::
+ StructField("ts", IntegerType) :: Nil)
+ val schema2 = StructType(
+ StructField("grp", StringType) ::
+ StructField("ts", IntegerType) ::
+ StructField("val", StringType) :: Nil)
+ val left = spark.createDataFrame(
+ List(Row("A", 5), Row("A", 10)).asJava, schema1)
+ val right = spark.createDataFrame(
+ List(Row("A", 6, "a"), Row("A", 8, "b"), Row("A", 12, "c")).asJava,
schema2)
+ checkAnswer(
+ left.joinAsOf(
+ right, left.col("ts"), right.col("ts"), usingColumns = Seq("grp"),
+ joinType = "inner", tolerance = null,
+ allowExactMatches = true, direction = "forward"),
+ Seq(
+ Row("A", 5, "A", 6, "a"),
+ Row("A", 10, "A", 12, "c")
+ )
+ )
+ }
+ }
+
+ test("nearest join - spill to disk") {
+ withSQLConf(
+ SQLConf.SORT_MERGE_JOIN_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "1",
+ SQLConf.SORT_MERGE_JOIN_EXEC_BUFFER_SPILL_THRESHOLD.key -> "1") {
+ val (df1, df2) = prepareForAsOfJoin()
+ // No equi-key (bufferAllRight path)
+ checkAnswer(
+ df1.joinAsOf(
+ df2, df1.col("a"), df2.col("a"), usingColumns = Seq.empty,
+ joinType = "inner", tolerance = null,
+ allowExactMatches = true, direction = "nearest"),
+ Seq(
+ Row(1, "x", "a", 1, "v", 1),
+ Row(5, "y", "b", 6, "y", 6),
+ Row(10, "z", "c", 7, "z", 7)
+ )
+ )
+ // With equi-key (bufferRightGroup path)
+ val schema1 = StructType(
+ StructField("grp", StringType) ::
+ StructField("ts", IntegerType) :: Nil)
+ val schema2 = StructType(
+ StructField("grp", StringType) ::
+ StructField("ts", IntegerType) ::
+ StructField("val", StringType) :: Nil)
+ val left = spark.createDataFrame(
+ List(Row("A", 5), Row("A", 10)).asJava, schema1)
+ val right = spark.createDataFrame(
+ List(Row("A", 3, "a"), Row("A", 7, "b"), Row("A", 12, "c")).asJava,
schema2)
+ checkAnswer(
+ left.joinAsOf(
+ right, left.col("ts"), right.col("ts"), usingColumns = Seq("grp"),
+ joinType = "inner", tolerance = null,
+ allowExactMatches = true, direction = "nearest"),
+ Seq(
+ Row("A", 5, "A", 3, "a"),
+ Row("A", 10, "A", 12, "c")
+ )
+ )
+ }
+ }
+
+ test("left outer join - spill to disk") {
+ withSQLConf(
+ SQLConf.SORT_MERGE_JOIN_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "1",
+ SQLConf.SORT_MERGE_JOIN_EXEC_BUFFER_SPILL_THRESHOLD.key -> "1") {
+ val (df1, df2) = prepareForAsOfJoin()
+ checkAnswer(
+ df1.joinAsOf(
+ df2, df1.col("a"), df2.col("a"), usingColumns = Seq.empty,
+ joinType = "leftouter", tolerance = null,
+ allowExactMatches = true, direction = "backward"),
+ Seq(
+ Row(1, "x", "a", 1, "v", 1),
+ Row(5, "y", "b", 3, "x", 3),
+ Row(10, "z", "c", 7, "z", 7)
+ )
+ )
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
