[SPARK-23702][SS] Forbid watermarks on both sides of stateful streaming operators. #20859
Conversation
| val statefulChildren = e.collect { | ||
| case a: Aggregate if a.isStreaming => a | ||
| case d: Deduplicate if d.isStreaming => d | ||
| case f: FlatMapGroupsWithState if f.isStreaming => f |
| } | ||
| statefulChildren.foreach { statefulNode => | ||
| if (statefulNode.collectFirst{ case e: EventTimeWatermark => e }.isDefined) { | ||
| throwError("Watermarks both before and after a stateful operator in a streaming " + |
There was a problem hiding this comment.
This gives the impression that it makes sense but we dont support it. In fact, its just ill-defined. May change this to something like ... Multiple watermarks before and after stateful operators is not well-defined in a streaming query.
There was a problem hiding this comment.
WDYT of something like "watermarks may not be present...". Talking about "well-defined" seems a bit confusing to me.
| CalendarInterval.fromString("interval 2 seconds"), | ||
| streamRelation))), | ||
| outputMode = Append, | ||
| expectedMsgs = Seq("both before and after")) |
There was a problem hiding this comment.
Add for other stateful operators as well.
|
Test build #88383 has finished for PR 20859 at commit
|
|
Test build #88424 has finished for PR 20859 at commit
|
|
Test build #88432 has finished for PR 20859 at commit
|
|
How would we like to handle this patch? I guess we add feature on handling multiple watermarks in #21701 so based on the direction this patch might be going to be abandoned. IMHO I'm not 100% sure we have clear use cases for defining multiple watermarks for one source, so this patch might help on simplify handling watermark. |
|
Test build #97493 has finished for PR 20859 at commit
|
4 participants
What changes were proposed in this pull request?
Forbid watermarks on both sides of stateful streaming operators.
Multiple sequential watermarks are in general not supported by the execution engine; support is only in parallel, e.g. on both sides of a join. We can normally resolve this by simply picking the topmost watermark operator and ignoring the rest, but this is not semantically valid when there's a stateful operator in between.
How was this patch tested?
new unit test