[SPARK-27340][SS] Alias on TimeWindow expression may cause watermark metadata lost #24457

LiangchangZ · 2019-04-25T11:19:55Z

What changes were proposed in this pull request?

window($"fooTime", "2 seconds").alias("fooWindow") can generate an expression tree Alias(fooWindow) <- TimeWindow. The tree will become Alias(fooWindow) <- Alias(window) <- Window(start, end) after analyzed by TimeWindowing rule. The Alias(window) got metadata of watermark when created:

val windowStruct = Alias(getWindow(0, 1), WINDOW_COL_NAME)(
exprId = windowAttr.exprId, explicitMetadata = Some(metadata))

but the Alias(fooWindow) is created before TimeWindowing rule effected. Its code path is:

...
case ne: NamedExpression => Alias(expr, alias)(explicitMetadata = Some(ne.metadata))
...

before TimeWindowing rule effected, the ne.metadata is None and cause the watermark metadata lost

We make the def name(alias: String) return a Alias which get metadata from its child automatically, when not specifying metadata explicitly.

Thank @LinhongLiu for helping analyzing this problem!

How was this patch tested?

Add a UT and do the integration tests by run the example in jira successfully and do not throw org.apache.spark.sql.AnalysisException anymore

…ne explicitMetadata, so the metadata of Alias object will directly got from its child Change-Id: Ia2246b05688461ad907f1e16c96e7282f655d5a6

Change-Id: Ie04be37fd6a4859b3b0da0c7c07d54558cb23758

Change-Id: If8f4ba6a93f4a37ffb096b95c62dce8f87db3233

LiangchangZ · 2019-04-26T06:41:00Z

cc @xuanyuanking

xuanyuanking · 2019-04-26T12:20:22Z

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

-      case ne: NamedExpression => Alias(expr, alias)(explicitMetadata = Some(ne.metadata))
-      case other => Alias(other, alias)()
-    }
+    Alias(expr, alias)()


This function introduced in #11908, the change here will also influence def alias(alias: String): Column and def as(alias: String): Column, do you check all the test cases related?

Yes, I have successfully run all relevant UT test.
(I‘m sorry that we have a long holiday recently, so the reply is late.

Related tests maybe not enough, as your change, as function has a different behavior, metadata no longer pass to explicitMetadata. My suggestion:

For safety, run all test, not only the related.

Do the fix just for SS scenario.

dongjoon-hyun · 2019-07-28T22:56:12Z

ok to test

SparkQA · 2019-07-29T02:18:37Z

Test build #108285 has finished for PR 24457 at commit a0508c9.

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

github-actions · 2019-12-30T00:07:09Z

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!

dongjoon-hyun · 2020-01-06T19:43:32Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/EventTimeWatermarkSuite.scala

+      .select(window($"eventTime", "5 seconds") as 'aliasWindow)
+
+    assert(aliasWindow.logicalPlan.output.exists(
+        _.metadata.contains(EventTimeWatermark.delayKey)))


Since this test case seems to fail on the master branch (as of today), the issue seems to exist still.

dongjoon-hyun · 2020-01-06T20:09:05Z

While preparing at 2.4.5 release, I just noticed that this was closed recently and we might need to fix the underlying issue. The test case failed in both master and branch-2.4.

If watermarks are ignored, the internal state grows indefinitely. How do you think about the reported issue, @tdas , @zsxwing , @cloud-fan , @HeartSaVioR , @gatorsmile?

HeartSaVioR · 2020-01-07T00:15:05Z

Looking at the test code, the issue seems to be valid and PR fixes the issue correctly. But I'm not sure about the side effect, as @xuanyuanking commented.

Btw I think this has been known issue and underlying issue may not just be missing copying metadata. I'm not sure Spark can ensure metadata is propagated correctly during any multiple transformations, including typed -> untyped, and vice versa. It doesn't seem to be a thing we can rely on.

I think the root issue is that the event time column and value are open to modify. Other streaming frameworks provide the way to specify the event time per row, and the value is treated as special column which cannot be modified (both column and value) during transformation.

I've had a long discussion with @echauchot (working with Spark runner in Beam) regarding this. Please follow the link : #23576 (comment)

HeartSaVioR · 2020-01-07T00:16:38Z

Oh bot closed the PR again... Looks like we should also remove Stale tag as well when reopening.

dongjoon-hyun · 2020-01-07T00:58:15Z

Thank you for feedback, @HeartSaVioR . I also agree with you and was surprised with this. With watermark bugs, Apache Spark structured streaming is not usable at all in case of the state operations.

For auto-close, @nchammas , what is the correct reopening process?
(also cc @srowen )

AmplabJenkins · 2020-01-07T01:07:59Z

Can one of the admins verify this patch?

nchammas · 2020-01-07T01:39:43Z

I think removing the Stale tag should do the trick. If not, I can investigate. We should perhaps update the close message to direct people accordingly.

Can PR authors remove the tag themselves, by the way, or does that require a committer?

LinhongLiu · 2020-01-07T02:39:43Z

@dongjoon-hyun @HeartSaVioR @xuanyuanking
Hi, the author of this patch @LiangchangZ currently has no time to continue it. If it needs to be fixed, I can carry on the following works.

dongjoon-hyun · 2020-01-07T03:09:58Z

Thank you, @nchammas . Tagging requires committership.

Thank you, @LinhongLiu . Yes. We can do that while keeping @LiangchangZ 's authorship. The current status of this PR is a stage where we are discussing the validity and impact of this bug. I believe this should be considered as Critical because this is disabling the core feature's availability. Let's wait and see the other people's comments.

xuanyuanking · 2020-01-07T03:35:57Z

@LinhongLiu Thanks Linhong, agree to continue this bugfix, please go ahead.

echauchot · 2020-01-07T09:35:33Z

With watermark bugs, Apache Spark structured streaming is not usable at all in case of the state operations.

Indeed, we (Beam) are stuck in the streaming mode implementation of the translation layer (Beam runner) using StructuredStreaming framework. What about re-opening #23576 as well ? CC @arunmahadevan

srowen · 2020-01-07T14:18:13Z

This was already open. I think the stale tag just has to be removed, done yesterday.

Follow-on to #26877. ### What changes were proposed in this pull request? This PR tweaks the stale PR message to [clarify](#24457 (comment)) the procedure for reopening a PR after it has been marked as stale. ### Why are the changes needed? This change should clarify the reopening process for contributors. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? N/A Closes #27114 from nchammas/SPARK-30173-stale-tweaks. Authored-by: Nicholas Chammas <nicholas.chammas@gmail.com> Signed-off-by: Sean Owen <srowen@gmail.com>

github-actions · 2020-04-17T00:10:39Z

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!

xuanyuanking · 2020-04-17T02:01:09Z

@LinhongLiu Are you still working on this? If not, I will take over and ping you for review.

dongjoon-hyun · 2020-04-18T00:39:26Z

Please take over this, @xuanyuanking . Thanks~

xuanyuanking · 2020-04-24T01:30:15Z

Thanks, I will submit a new PR today.

…data lost Credit to LiangchangZ, this PR reuses the UT as well as integrate test in #24457. Thanks Liangchang for your solid work. ### What changes were proposed in this pull request? Make metadata propagatable between Aliases. ### Why are the changes needed? In Structured Streaming, we added an Alias for TimeWindow by default. https://github.com/apache/spark/blob/590b9a0132b68d9523e663997def957b2e46dfb1/sql/core/src/main/scala/org/apache/spark/sql/functions.scala#L3272-L3273 For some cases like stream join with watermark and window, users need to add an alias for convenience(we also added one in StreamingJoinSuite). The current metadata handling logic for `as` will lose the watermark metadata https://github.com/apache/spark/blob/590b9a0132b68d9523e663997def957b2e46dfb1/sql/core/src/main/scala/org/apache/spark/sql/Column.scala#L1049-L1054 and finally cause the AnalysisException: ``` Stream-stream outer join between two streaming DataFrame/Datasets is not supported without a watermark in the join keys, or a watermark on the nullable side and an appropriate range condition ``` ### Does this PR introduce any user-facing change? Bugfix for an alias on time window with watermark. ### How was this patch tested? New UTs added. One for the functionality and one for explaining the common scenario. Closes #28326 from xuanyuanking/SPARK-27340. Authored-by: Yuanjian Li <xyliyuanjian@gmail.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

…data lost Credit to LiangchangZ, this PR reuses the UT as well as integrate test in #24457. Thanks Liangchang for your solid work. ### What changes were proposed in this pull request? Make metadata propagatable between Aliases. ### Why are the changes needed? In Structured Streaming, we added an Alias for TimeWindow by default. https://github.com/apache/spark/blob/590b9a0132b68d9523e663997def957b2e46dfb1/sql/core/src/main/scala/org/apache/spark/sql/functions.scala#L3272-L3273 For some cases like stream join with watermark and window, users need to add an alias for convenience(we also added one in StreamingJoinSuite). The current metadata handling logic for `as` will lose the watermark metadata https://github.com/apache/spark/blob/590b9a0132b68d9523e663997def957b2e46dfb1/sql/core/src/main/scala/org/apache/spark/sql/Column.scala#L1049-L1054 and finally cause the AnalysisException: ``` Stream-stream outer join between two streaming DataFrame/Datasets is not supported without a watermark in the join keys, or a watermark on the nullable side and an appropriate range condition ``` ### Does this PR introduce any user-facing change? Bugfix for an alias on time window with watermark. ### How was this patch tested? New UTs added. One for the functionality and one for explaining the common scenario. Closes #28326 from xuanyuanking/SPARK-27340. Authored-by: Yuanjian Li <xyliyuanjian@gmail.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org> (cherry picked from commit ba7adc4) Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

…data lost Credit to LiangchangZ, this PR reuses the UT as well as integrate test in apache#24457. Thanks Liangchang for your solid work. Make metadata propagatable between Aliases. In Structured Streaming, we added an Alias for TimeWindow by default. https://github.com/apache/spark/blob/590b9a0132b68d9523e663997def957b2e46dfb1/sql/core/src/main/scala/org/apache/spark/sql/functions.scala#L3272-L3273 For some cases like stream join with watermark and window, users need to add an alias for convenience(we also added one in StreamingJoinSuite). The current metadata handling logic for `as` will lose the watermark metadata https://github.com/apache/spark/blob/590b9a0132b68d9523e663997def957b2e46dfb1/sql/core/src/main/scala/org/apache/spark/sql/Column.scala#L1049-L1054 and finally cause the AnalysisException: ``` Stream-stream outer join between two streaming DataFrame/Datasets is not supported without a watermark in the join keys, or a watermark on the nullable side and an appropriate range condition ``` Bugfix for an alias on time window with watermark. New UTs added. One for the functionality and one for explaining the common scenario. Closes apache#28326 from xuanyuanking/SPARK-27340. Authored-by: Yuanjian Li <xyliyuanjian@gmail.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

LiangchangZ added 2 commits April 25, 2019 16:15

when use Column.name(alias: String), we create Alias directly with No…

c5b08d9

…ne explicitMetadata, so the metadata of Alias object will directly got from its child Change-Id: Ia2246b05688461ad907f1e16c96e7282f655d5a6

modify comments

4ba4120

Change-Id: Ie04be37fd6a4859b3b0da0c7c07d54558cb23758

LiangchangZ changed the title ~~[SPARK-27340] Alias on TimeWindow expression may cause watermark metadata lost~~ [SPARK-27340][Streaming] Alias on TimeWindow expression may cause watermark metadata lost Apr 25, 2019

add ut

f291d54

Change-Id: If8f4ba6a93f4a37ffb096b95c62dce8f87db3233

xuanyuanking reviewed Apr 26, 2019

View reviewed changes

modify code style

a0508c9

dongjoon-hyun added SQL STRUCTURED STREAMING labels Jun 14, 2019

dongjoon-hyun changed the title ~~[SPARK-27340][Streaming] Alias on TimeWindow expression may cause watermark metadata lost~~ [SPARK-27340][SS] Alias on TimeWindow expression may cause watermark metadata lost Jul 28, 2019

github-actions bot added the Stale label Dec 30, 2019

github-actions bot closed this Dec 31, 2019

dongjoon-hyun reopened this Jan 6, 2020

dongjoon-hyun reviewed Jan 6, 2020

View reviewed changes

github-actions bot closed this Jan 7, 2020

dongjoon-hyun reopened this Jan 7, 2020

dongjoon-hyun removed the Stale label Jan 7, 2020

nchammas mentioned this pull request Jan 7, 2020

[SPARK-30173] Tweak stale PR message #27114

Closed

github-actions bot added the Stale label Apr 17, 2020

github-actions bot closed this Apr 18, 2020

xuanyuanking mentioned this pull request Apr 24, 2020

[SPARK-27340][SS] Alias on TimeWindow expression cause watermark metadata lost #28326

Closed

dongjoon-hyun mentioned this pull request Apr 27, 2020

[SPARK-27340][SS][2.4] Alias on TimeWindow expression may cause watermark metadata lost #28377

Closed

[SPARK-27340][SS] Alias on TimeWindow expression may cause watermark metadata lost #24457

[SPARK-27340][SS] Alias on TimeWindow expression may cause watermark metadata lost #24457

Uh oh!

Conversation

LiangchangZ commented Apr 25, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

LiangchangZ commented Apr 26, 2019

Uh oh!

xuanyuanking Apr 26, 2019

Choose a reason for hiding this comment

Uh oh!

LiangchangZ May 5, 2019

Choose a reason for hiding this comment

Uh oh!

xuanyuanking May 7, 2019

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun commented Jul 28, 2019

Uh oh!

SparkQA commented Jul 29, 2019

Uh oh!

github-actions bot commented Dec 30, 2019

Uh oh!

dongjoon-hyun Jan 6, 2020

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun commented Jan 6, 2020

Uh oh!

HeartSaVioR commented Jan 7, 2020

Uh oh!

HeartSaVioR commented Jan 7, 2020

Uh oh!

dongjoon-hyun commented Jan 7, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

AmplabJenkins commented Jan 7, 2020

Uh oh!

nchammas commented Jan 7, 2020

Uh oh!

LinhongLiu commented Jan 7, 2020

Uh oh!

dongjoon-hyun commented Jan 7, 2020

Uh oh!

xuanyuanking commented Jan 7, 2020

Uh oh!

echauchot commented Jan 7, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

srowen commented Jan 7, 2020

Uh oh!

github-actions bot commented Apr 17, 2020

Uh oh!

xuanyuanking commented Apr 17, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

dongjoon-hyun commented Apr 18, 2020

Uh oh!

xuanyuanking commented Apr 24, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

10 participants

LiangchangZ commented Apr 25, 2019 •

edited

Loading

dongjoon-hyun commented Jan 7, 2020 •

edited

Loading

echauchot commented Jan 7, 2020 •

edited

Loading

xuanyuanking commented Apr 17, 2020 •

edited

Loading