ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[SPARK-35748][SS][SQL] Fix StreamingJoinHelper to be able to handle day-time interval

Browse source 
### What changes were proposed in this pull request?

This PR fixes `StreamingJoinHelper` to be able to handle day-time interval.

### Why are the changes needed?

In the current master, `StreamingJoinHelper.getStateValueWatermark` can't handle conditions which contain day-time interval literals.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

New assertions added to `StreamingJoinHlelperSuite`.

Closes #32896 from sarutak/streamingjoinhelper-daytime.

Authored-by: Kousuke Saruta <sarutak@oss.nttdata.com>
Signed-off-by: Max Gekk <max.gekk@gmail.com>
This commit is contained in:
Kousuke Saruta 2021-06-14 15:45:36 +03:00 committed by Max Gekk
parent 439e94c171
commit 82af318c31
2 changed files with 15 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelper.scala
View file

@ -266,6 +266,9 @@ object StreamingJoinHelper extends PredicateHelper with Logging {
Literal(calendarInterval.days * MICROS_PER_DAY.toDouble +
calendarInterval.microseconds.toDouble)
}
case _: DayTimeIntervalType =>
// Unbox and then cast
Literal(lit.value.asInstanceOf[Long].toDouble)
case DoubleType =>
Multiply(lit, Literal(1000000.0))
case _: NumericType =>

12
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelperSuite.scala
View file

@ -86,6 +86,18 @@ class StreamingJoinHelperSuite extends AnalysisTest {
assert(watermarkFrom("rightTime - interval 1 second < leftTime - interval 3 second")
=== Some(12000))
assert(watermarkFrom("leftTime > rightTime + interval '0 00:00:01' day to second")
=== Some(11000))
assert(watermarkFrom("leftTime + interval '00:00:02' hour to second > rightTime ")
=== Some(8000))
assert(watermarkFrom("leftTime > rightTime - interval '00:03' minute to second")
=== Some(7000))
assert(watermarkFrom("rightTime < leftTime - interval '1 20:30:40' day to second")
=== Some(160250000))
assert(watermarkFrom(
"rightTime - interval 1 second < leftTime - interval '20:15:32' hour to second")
=== Some(72941000))
// Test with casted long type + constants on either side of equation
// Note: long type and constants commute, so more combinations to test.
// -- Constants on the right