Github user jose-torres commented on a diff in the pull request:
https://github.com/apache/spark/pull/20859#discussion_r175585873
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala
---
@@ -160,6 +160,19 @@ object UnsupportedOperationChecker {
case _: InsertIntoDir =>
throwError("InsertIntoDir is not supported with streaming
DataFrames/Datasets")
+ case e: EventTimeWatermark =>
+ 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 " +
--- End diff --
WDYT of something like "watermarks may not be present...". Talking about
"well-defined" seems a bit confusing to me.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
