[SPARK-19861][SS] watermark should not be a negative time. #17202

uncleGen · 2017-03-08T03:30:32Z

What changes were proposed in this pull request?

watermark should not be negative. This behavior is invalid, check it before real run.

How was this patch tested?

add new unit test.

SparkQA · 2017-03-08T05:43:12Z

Test build #74172 has finished for PR 17202 at commit 00bbadc.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

uncleGen · 2017-03-08T06:06:05Z

cc @zsxwing

SparkQA · 2017-03-08T06:07:33Z

Test build #74182 has started for PR 17202 at commit c8905fe.

felixcheung · 2017-03-08T06:30:51Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -576,6 +576,8 @@ class Dataset[T] private[sql](
    val parsedDelay =
      Option(CalendarInterval.fromString("interval " + delayThreshold))
        .getOrElse(throw new AnalysisException(s"Unable to parse time delay '$delayThreshold'"))
+    assert(parsedDelay.microseconds >= 0,


what happens when it is 0?

set 0 means event time should be not less than max event time in last batch.

In other words, 0 means the events always arrive in order

srowen · 2017-03-08T10:54:43Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

+      val millisPerMonth = CalendarInterval.MICROS_PER_DAY / 1000 * 31
+      parsedDelay.milliseconds + parsedDelay.months * millisPerMonth
+    }
+    assert(delayMs >= 0, s"delay threshold should not be a negative time: $delayThreshold")


Why compute all this -- don't you just mean to assert about delayThreshold? this derived value can only be negative if the input is right?

@srowen Thanks for you review!

Why compute all this -- don't you just mean to assert about delayThreshold?

I do mean to check the delayThreshold. delayThreshold is converted from String to CalendarInterval. CalendarInterval divides the delayThreshold into two parts, i.e. month (contain year and month) and microseconds of rest. (https://github.com/apache/spark/blob/master/common/unsafe/src/main/java/org/apache/spark/unsafe/types/CalendarInterval.java#L86)
(https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/EventTimeWatermarkExec.scala#L87)

this derived value can only be negative if the input is right?

Sorry, I dont get what you mean.

Why don't you just assert(delayThreshold >= 0, ...)? Why does that check require computing delayMs?

delayThreshold: String can not be used to assert directly, like:

inputData.withWatermark("value", "1 month -40 days") inputData.withWatermark("value", "-10 seconds")

Ah right. Can you not assert about parsedDelay then? it seems like that's always negative when the result of this conversion, copied from somewhere else? is negative.

Maybe you misunderstand my cases above. Those cases are invalid, i.e. the parsedDelay are negative.

case parsedDelay validity

inputData.withWatermark("value", "1 month -40 days") negative invalid

inputData.withWatermark("value", "-10 seconds") negative invalid

inputData.withWatermark("value", "10 seconds") positive valid

inputData.withWatermark("value", "1 day -10 seconds") positive valid

inputData.withWatermark("value", "0 second") - valid

So can you require(parsedDelay >= 0, ...) to detect invalid cases?

Sorry that last example didn't make sense, it's not a single number. It just doesn't seem like you need to reproduce this conversion to check what you want to, which is that the two fields aren't negative. require(parsedDelay.milliseconds >= 0 && parsedDelay.months >= 0)

In fact, I wonder if this should be some kind of method of CalendarInterval, like isNegative? it's not that trivial to decide if an interval is negative, and maybe other places do this.

@srowen +1 to your comments. We should make it more significative but not just valid.

SparkQA · 2017-03-08T12:42:12Z

Test build #74203 has finished for PR 17202 at commit 9394e0d.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public static class LongWrapper
public static class IntWrapper
trait WatermarkSupport extends UnaryExecNode

SparkQA · 2017-03-08T12:46:25Z

Test build #74204 has finished for PR 17202 at commit c5e2221.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-03-08T17:28:25Z

Test build #74211 has finished for PR 17202 at commit b0c8f12.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

rxin · 2017-03-08T18:12:00Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -563,7 +563,7 @@ class Dataset[T] private[sql](
   * @param eventTime the name of the column that contains the event time of the row.
   * @param delayThreshold the minimum delay to wait to data to arrive late, relative to the latest
   *                       record that has been processed in the form of an interval
-   *                       (e.g. "1 minute" or "5 hours").
+   *                       (e.g. "1 minute" or "5 hours"). NOTE: This should not be a negative time.


I'd just say "This should not be negative."

rxin · 2017-03-08T18:12:16Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

+      val millisPerMonth = CalendarInterval.MICROS_PER_DAY / 1000 * 31
+      parsedDelay.milliseconds + parsedDelay.months * millisPerMonth
+    }
+    require(delayMs >= 0, s"delay threshold should not be a negative time: $delayThreshold")


"delay threshold ($delayThreshold) should not be negative"

zsxwing · 2017-03-08T19:44:38Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -576,6 +576,11 @@ class Dataset[T] private[sql](
    val parsedDelay =
      Option(CalendarInterval.fromString("interval " + delayThreshold))
        .getOrElse(throw new AnalysisException(s"Unable to parse time delay '$delayThreshold'"))
+    val delayMs = {


I guess you copied these codes from EventTimeWatermarkExec. Could you extract them as a new method in object EventTimeWatermark and reuse it?

Since you are touching these files, could you also fix the following two places as well?

spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/EventTimeWatermark.scala

Line 42 in d8830c5

.putLong(EventTimeWatermark.delayKey, delay.milliseconds)

doesn't use the correct delayMs. Right now we don't use this value in the logical plan, but it's better to make it consistent.

I forgot to update EventTimeWatermarkExec in [SPARK-19859][SS]The new watermark should override the old one #17199. Could you help me fix it?

uncleGen · 2017-03-09T02:34:39Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -576,6 +576,8 @@ class Dataset[T] private[sql](
    val parsedDelay =
      Option(CalendarInterval.fromString("interval " + delayThreshold))
        .getOrElse(throw new AnalysisException(s"Unable to parse time delay '$delayThreshold'"))
+    require(parsedDelay.milliseconds >= 0 && parsedDelay.months >= 0,
+      s"delay threshold ($delayThreshold) should not be negative.")


use require(parsedDelay.milliseconds >= 0 && parsedDelay.months >= 0) to make delayThreshold more reasonable and significative.

SparkQA · 2017-03-09T03:28:08Z

Test build #74232 has finished for PR 17202 at commit 10a949d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

uncleGen · 2017-03-09T12:05:46Z

cc @srowen and @zsxwing

zsxwing · 2017-03-09T19:07:00Z

LGTM. Merging to master and 2.1.

## What changes were proposed in this pull request? `watermark` should not be negative. This behavior is invalid, check it before real run. ## How was this patch tested? add new unit test. Author: uncleGen <hustyugm@gmail.com> Author: dylon <hustyugm@gmail.com> Closes #17202 from uncleGen/SPARK-19861. (cherry picked from commit 30b18e6) Signed-off-by: Shixiong Zhu <shixiong@databricks.com>

uncleGen added 2 commits March 8, 2017 11:28

watermark should not be a negative time.

dcc77ed

update

00bbadc

Merge branch 'master' into SPARK-19861

c8905fe

felixcheung reviewed Mar 8, 2017

View reviewed changes

uncleGen added 3 commits March 8, 2017 18:28

fix the wrong delayMs calculation

93a5d6d

Merge branch 'SPARK-19861' of github.com:uncleGen/spark into SPARK-19861

9394e0d

fix

c5e2221

srowen reviewed Mar 8, 2017

View reviewed changes

address comments from @srowen

b0c8f12

rxin reviewed Mar 8, 2017

View reviewed changes

address comments from @srowen and @rxin

10a949d

zsxwing reviewed Mar 9, 2017

View reviewed changes

uncleGen commented Mar 9, 2017

View reviewed changes

srowen approved these changes Mar 9, 2017

View reviewed changes

asfgit closed this in 30b18e6 Mar 9, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-19861][SS] watermark should not be a negative time. #17202

[SPARK-19861][SS] watermark should not be a negative time. #17202

uncleGen commented Mar 8, 2017 •

edited

Loading

SparkQA commented Mar 8, 2017

uncleGen commented Mar 8, 2017

SparkQA commented Mar 8, 2017

felixcheung Mar 8, 2017

uncleGen Mar 8, 2017

zsxwing Mar 8, 2017

srowen Mar 8, 2017

uncleGen Mar 8, 2017

srowen Mar 8, 2017

uncleGen Mar 8, 2017

srowen Mar 8, 2017

uncleGen Mar 8, 2017 •

edited

Loading

srowen Mar 8, 2017 •

edited

Loading

srowen Mar 8, 2017

uncleGen Mar 9, 2017 •

edited

Loading

SparkQA commented Mar 8, 2017

SparkQA commented Mar 8, 2017

SparkQA commented Mar 8, 2017

rxin Mar 8, 2017

rxin Mar 8, 2017

zsxwing Mar 8, 2017

uncleGen Mar 9, 2017

uncleGen Mar 9, 2017

SparkQA commented Mar 9, 2017

uncleGen commented Mar 9, 2017

zsxwing commented Mar 9, 2017

case	parsedDelay	validity
inputData.withWatermark("value", "1 month -40 days")	negative	invalid
inputData.withWatermark("value", "-10 seconds")	negative	invalid
inputData.withWatermark("value", "10 seconds")	positive	valid
inputData.withWatermark("value", "1 day -10 seconds")	positive	valid
inputData.withWatermark("value", "0 second")	-	valid

[SPARK-19861][SS] watermark should not be a negative time. #17202

[SPARK-19861][SS] watermark should not be a negative time. #17202

Conversation

uncleGen commented Mar 8, 2017 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Mar 8, 2017

uncleGen commented Mar 8, 2017

SparkQA commented Mar 8, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

uncleGen Mar 8, 2017 • edited Loading

Choose a reason for hiding this comment

srowen Mar 8, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

uncleGen Mar 9, 2017 • edited Loading

Choose a reason for hiding this comment

SparkQA commented Mar 8, 2017

SparkQA commented Mar 8, 2017

SparkQA commented Mar 8, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Mar 9, 2017

uncleGen commented Mar 9, 2017

zsxwing commented Mar 9, 2017

uncleGen commented Mar 8, 2017 •

edited

Loading

uncleGen Mar 8, 2017 •

edited

Loading

srowen Mar 8, 2017 •

edited

Loading

uncleGen Mar 9, 2017 •

edited

Loading