[SPARK-38349][SS] No need to filter events when sessionwindow gapDuration greater than 0

[nyingping]

What changes were proposed in this pull request?

Static gapDuration on session Window,No need to filter events when sessionwindow gapDuration greater than 0.

Why are the changes needed?

save calculation resources and improve performance.

Does this PR introduce any user-facing change?

No

How was this patch tested?

add new UT and benchmark.

a simple benchmark on [9dae8a5] . thanks again HeartSaVioR@d532b6f.

---

case 1

spark.range(numOfRow)
      .selectExpr("CAST(id AS timestamp) AS time")
      .select(session_window(col("time"), "10 seconds"))
      .count()

Result:

Java HotSpot(TM) 64-Bit Server VM 1.8.0_291-b10 on Windows 10 10.0
AMD64 Family 23 Model 96 Stepping 1, AuthenticAMD
session windows:                          Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
old logic                                          1688           1730          61          5.9         168.8       1.0X
new logic                                            21             26           5        487.3           2.1      82.3X

[nyingping]

I will check it later.

[AmplabJenkins]

Can one of the admins verify this patch?

[HyukjinKwon]

cc @HeartSaVioR FYI

[HeartSaVioR]

@nyingping
Now this PR has a small conflict; could you please resolve the conflict? Thanks in advance!

[nyingping]

@HeartSaVioR yeah,resolved.thanks for your time!

[HeartSaVioR]

Sorry I have been very busy with handling stuff. I see the rationalization, but it is a bit unclear in the code which cases we want to optimize (the case gapDuration produces > 0 consistently), and the code change does not seem to be intuitive given that the code tries to disassemble with Cast.

[HeartSaVioR]

cc. @viirya since I see your client seems to use session window. For static gap duration this would reduce consistent overhead (filter with comparison between timestamp) per row.

[viirya]

The rationalization makes sense, but the current approach looks problematic.

[HeartSaVioR]

I guess this is being complicated to also handle the case of dynamic gap with "case when" & literals.
What about just narrowing down the scope to the static gap? This is a majority of use case for session window, and this will simplify the logic a lot - probably need to handle Literal only.

[nyingping]

sure,I think it is acceptable that only deal with static gaps will greatly reduce uncertainty.

[nyingping]

I've narrowed it down to Static gap duration.
@HeartSaVioR @viirya Could you have a look when you have a chance?Thanks in advance!

[nyingping]

@viirya @HeartSaVioR
Thanks for suggestion,Thanks a lot!

[HeartSaVioR]

only nits about style

[HeartSaVioR]

+1

[HeartSaVioR]

Thanks! Merging to master.

[HeartSaVioR]

Thanks @nyingping for your contribution! I merged this to master.

[nyingping]

@HeartSaVioR @viirya Thanks a lot!