[SPARK-39347] [SS] Generate wrong time window when (timestamp-startTime) % slideDuration…

[nyingping]

What changes were proposed in this pull request?

Fix bug that Generate wrong time window when (timestamp-startTime) % slideDuration < 0

The original time window generation rule

 lastStart <- timestamp - (timestamp - startTime + slideDuration) % slideDuration

change like this

 remainder <-  (timestamp - startTime) % slideDuration
 lastStart <-
    if (remainder < 0) timestamp - remainder - slideDuration
    else timestamp - remainder
   

reference: apache/flink#18982

Why are the changes needed?

Since the generation strategy of the sliding window in PR #35362 is changed to the current one, and that leads to a new problem.

A window generation error occurs when the time required to process the recorded data is negative and the modulo value between the time and window length is less than 0. In the current test cases, this bug does not thorw up.

test("negative timestamps")

val df1 = Seq(
  ("1970-01-01 00:00:02", 1),
  ("1970-01-01 00:00:12", 2)).toDF("time", "value")
val df2 = Seq(
  (LocalDateTime.parse("1970-01-01T00:00:02"), 1),
  (LocalDateTime.parse("1970-01-01T00:00:12"), 2)).toDF("time", "value")

Seq(df1, df2).foreach { df =>
  checkAnswer(
    df.select(window($"time", "10 seconds", "10 seconds", "5 seconds"), $"value")
      .orderBy($"window.start".asc)
      .select($"window.start".cast(StringType), $"window.end".cast(StringType), $"value"),
    Seq(
      Row("1969-12-31 23:59:55", "1970-01-01 00:00:05", 1),
      Row("1970-01-01 00:00:05", "1970-01-01 00:00:15", 2))
  )
} 

The timestamp of the above test data is not negative, and the value modulo the window length is not negative, so it can be passes the test case.

An exception occurs when the timestamp becomes something like this.

val df3 = Seq(
      ("1969-12-31 00:00:02", 1),
      ("1969-12-31 00:00:12", 2)).toDF("time", "value")
val df4 = Seq(
      (LocalDateTime.parse("1969-12-31T00:00:02"), 1),
      (LocalDateTime.parse("1969-12-31T00:00:12"), 2)).toDF("time", "value")    Seq(df3, df4).foreach { df =>
      checkAnswer(
        df.select(window($"time", "10 seconds", "10 seconds", "5 seconds"), $"value")
          .orderBy($"window.start".asc)
          .select($"window.start".cast(StringType), $"window.end".cast(StringType), $"value"),
        Seq(
          Row("1969-12-30 23:59:55", "1969-12-31 00:00:05", 1),
          Row("1969-12-31 00:00:05", "1969-12-31 00:00:15", 2))
      )
} 

run and get unexpected result:

== Results ==
!== Correct Answer - 2 ==                      == Spark Answer - 2 ==
!struct<>                                      struct<CAST(window.start AS STRING):string,CAST(window.end AS STRING):string,value:int>
![1969-12-30 23:59:55,1969-12-31 00:00:05,1]   [1969-12-31 00:00:05,1969-12-31 00:00:15,1]
![1969-12-31 00:00:05,1969-12-31 00:00:15,2]   [1969-12-31 00:00:15,1969-12-31 00:00:25,2] 

Does this PR introduce any user-facing change?

No

How was this patch tested?

Add new unit test.

benchmark result

oldlogic#18364  VS 【fix version】

Running benchmark: tumbling windows
Running case: old logic
Stopped after 407 iterations, 10012 ms
Running case: new logic
Stopped after 615 iterations, 10007 ms
Java HotSpot(TM) 64-Bit Server VM 1.8.0_181-b13 on Windows 10 10.0
Intel64 Family 6 Model 158 Stepping 10, GenuineIntel
tumbling windows:                         Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
old logic                                            17             25           9        580.1           1.7       1.0X
new logic                                            15             16           2        680.8           1.5       1.2X

Running benchmark: sliding windows
Running case: old logic
Stopped after 10 iterations, 10296 ms
Running case: new logic
Stopped after 15 iterations, 10391 ms
Java HotSpot(TM) 64-Bit Server VM 1.8.0_181-b13 on Windows 10 10.0
Intel64 Family 6 Model 158 Stepping 10, GenuineIntel
sliding windows:                          Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
old logic                                          1000           1030          19         10.0         100.0       1.0X
new logic                                           668            693          21         15.0          66.8       1.5X

Fixed version than PR #38069 lost a bit of the performance.

[nyingping]

This bug was caused by my previous PR. I'm sorry.
Could you have a look when you have time @HeartSaVioR @viirya,Thanks in advance.

[AmplabJenkins]

Can one of the admins verify this patch?

[HeartSaVioR]

Sorry I'll find a time sooner. I'll also find someone able to review this in prior.

[srowen]

When does processing time become negative -- is this a more fundamental problem to fix?

Also, does not seem to be a reason to use CaseWhen here? just do this in Scala code

[nyingping]

Thank you for reviewing.
My understanding is that whether the event time is negative or not is determined by the upstream data source, not the Spark platform itself. Even if the event time is negative, the platform should give the correct value.

Does in scala code refer to the following code:

val lastStart = if (remainder < 0) {
    timestamp - remainder - window.slideDuration
 } else {
    timestamp - remainder
}

In that case,The remainder is a expressions.predicate, the remainder < 0 can't get a boolean value directly.

[jerrypeng]

Can you explain why this is correct now? Please give an example.

[nyingping]

Thank you for reviewing.

For any event timestamp, the start time of the last window (whether tumbling or sliding) obtained by it should always be less than the current timestamp. When (timestamp - window.starttime)% window.slideduration <0, the obtained laststart will be greater than timestamp. At this time, it is necessary to shift the window to the right, which is the correct laststart value.

For example

code

    val timestamp = -13
    val offset = 0
    val windowSize = 7

    // old code
    val lastStartOld = timestamp - (timestamp - offset - windowSize) % windowSize

   // new code
    val remainder =  (timestamp - offset) % windowSize

    val lastStartNew =
      if (remainder < 0) {
        timestamp - remainder - windowSize
      } else {
        timestamp - remainder
      }

    println(s"lastStartOld = $lastStartOld   lastStartNew = $lastStartNew")

result

lastStartOld = -7 lastStartNew = -14

[srowen]

I think the question is how this arises in the first place. The code is self explanatory, not asking you to explain it

[nyingping]

@srowen sure,As mentioned above, I think this problem is unavoidable because we can't control the behavior of users. All we can do is get the right results even if users enter illegal data.

[jerrypeng]

If the timeColumn is before epoch, what would be the timestamp long val be? Negative?

[nyingping]

before epoch means that B.C or before '1970-01-01 00:00:00'? I understanding that both of negative value.

[dingsl-giser]

Whether has something to do with this PR[SPARK-39176][PYSPARK] Fixed a problem with pyspark serializing pre-1970 datetime in windows.

[HeartSaVioR]

General comment from what I see in review comments:

I see you repeat the explanation of the code you changed; I don't think reviewers asked about the detailed explanation of the code changes. There is no "high-level" explanation why it is broken (I roughly see it's from the language spec of modulo operation), and also "high-level" explanation how you deal with it in this PR. Please look through the description of the reference Flink PR you linked - while it also mentioned about code snippet, it explained with high level first, and then introduced the code change it proposed.

As long as you update the PR description with high-level explanation, I guess it should be straightforward to understand the code change, and you'd easily pass the reviews.

[nyingping]

I get it, and I'll try to update this part as much as possible,thanks a lot. @HeartSaVioR

[github-actions]

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!