HeartSaVioR commented on code in PR #59629:
URL: https://github.com/apache/spark/pull/59629#discussion_r4333746848
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala:
##########
@@ -1274,6 +1274,42 @@ abstract class StreamingOuterJoinBase extends
StreamingJoinSuite {
import testImplicits._
import org.apache.spark.sql.functions._
+ for {
+ joinType <- Seq("left_outer", "right_outer")
+ history <- Seq("continuous", "restart")
+ } {
+ test(s"SPARK-60378: $joinType eviction includes old unwatermarked keys,
$history") {
+ val leftInput = MemoryStream[(Int, Int)]
+ val rightInput = MemoryStream[(Int, Int)]
+ val left = leftInput.toDF().toDF("leftKey", "time")
+ .select($"leftKey", timestamp_seconds($"time").as("leftTime"))
+ val right = rightInput.toDF().toDF("rightKey", "time")
+ .select($"rightKey", timestamp_seconds($"time").as("rightTime"))
+ val leftOuter = joinType == "left_outer"
+ val joined = (if (leftOuter) left else left.withWatermark("leftTime",
"10 seconds"))
+ .join(if (leftOuter) right.withWatermark("rightTime", "10 seconds")
else right,
+ expr("leftKey = rightKey AND leftTime = rightTime"), joinType)
+ .select($"leftKey", $"rightKey", $"leftTime".cast("int"),
$"rightTime".cast("int"))
+ val stateInput = if (leftOuter) leftInput else rightInput
+ val watermarkInput = if (leftOuter) rightInput else leftInput
+ val expected = if (leftOuter) Row(2, null, 5, null) else Row(null, 2,
null, 5)
+
+ withSQLConf(SQLConf.STREAMING_NO_DATA_MICRO_BATCHES_ENABLED.key ->
"false") {
+ withTempDir { checkpointDir =>
+ val start = StartStream(checkpointLocation =
checkpointDir.getCanonicalPath)
+ val setup = Seq(start,
+ AddData(watermarkInput, (9, 20)), CheckNewAnswer(),
+ AddData(watermarkInput, (9, 30)), CheckNewAnswer()) ++
+ (if (history == "restart") Seq(StopStream, start) else Nil)
+ testStream(joined)((setup ++ Seq(
+ // This input has no late filter, so it can add keys below the
previous scan bound.
Review Comment:
self-review: I'd clarify the watermark for late record filtering here rather
than forcing reviewers to reason about this
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala:
##########
@@ -1274,6 +1274,42 @@ abstract class StreamingOuterJoinBase extends
StreamingJoinSuite {
import testImplicits._
import org.apache.spark.sql.functions._
+ for {
+ joinType <- Seq("left_outer", "right_outer")
Review Comment:
self-review: define enum for this and below
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala:
##########
@@ -1274,6 +1274,42 @@ abstract class StreamingOuterJoinBase extends
StreamingJoinSuite {
import testImplicits._
import org.apache.spark.sql.functions._
+ for {
+ joinType <- Seq("left_outer", "right_outer")
+ history <- Seq("continuous", "restart")
+ } {
+ test(s"SPARK-60378: $joinType eviction includes old unwatermarked keys,
$history") {
+ val leftInput = MemoryStream[(Int, Int)]
+ val rightInput = MemoryStream[(Int, Int)]
+ val left = leftInput.toDF().toDF("leftKey", "time")
+ .select($"leftKey", timestamp_seconds($"time").as("leftTime"))
+ val right = rightInput.toDF().toDF("rightKey", "time")
+ .select($"rightKey", timestamp_seconds($"time").as("rightTime"))
+ val leftOuter = joinType == "left_outer"
+ val joined = (if (leftOuter) left else left.withWatermark("leftTime",
"10 seconds"))
+ .join(if (leftOuter) right.withWatermark("rightTime", "10 seconds")
else right,
+ expr("leftKey = rightKey AND leftTime = rightTime"), joinType)
+ .select($"leftKey", $"rightKey", $"leftTime".cast("int"),
$"rightTime".cast("int"))
+ val stateInput = if (leftOuter) leftInput else rightInput
Review Comment:
self-review: it's hard to reason about the name; why this is named as
"state"Input and why the below is named as "watermark"Input? Either clarify it
in code comment or look for better name on better clarification.
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingSymmetricHashJoinHelperSuite.scala:
##########
@@ -127,4 +129,23 @@ class StreamingSymmetricHashJoinHelperSuite extends
StreamTest {
leftAttributeA === rightAttributeC && randAttribute > Literal(0)))
assert(split.full.contains(predicate))
}
+
+ for (watermarkColumn <- Seq("none", "key", "other")) {
Review Comment:
self-review: define enum
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala:
##########
@@ -1274,6 +1274,42 @@ abstract class StreamingOuterJoinBase extends
StreamingJoinSuite {
import testImplicits._
import org.apache.spark.sql.functions._
+ for {
+ joinType <- Seq("left_outer", "right_outer")
+ history <- Seq("continuous", "restart")
+ } {
+ test(s"SPARK-60378: $joinType eviction includes old unwatermarked keys,
$history") {
+ val leftInput = MemoryStream[(Int, Int)]
+ val rightInput = MemoryStream[(Int, Int)]
+ val left = leftInput.toDF().toDF("leftKey", "time")
+ .select($"leftKey", timestamp_seconds($"time").as("leftTime"))
+ val right = rightInput.toDF().toDF("rightKey", "time")
+ .select($"rightKey", timestamp_seconds($"time").as("rightTime"))
+ val leftOuter = joinType == "left_outer"
+ val joined = (if (leftOuter) left else left.withWatermark("leftTime",
"10 seconds"))
Review Comment:
self-review: this hurts readability - don't do this and below line in here;
do this in the definition of left and right (in above)
--
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]
