Github user tdas commented on a diff in the pull request: https://github.com/apache/spark/pull/19239#discussion_r139035867 --- Diff: sql/core/src/test/scala/org/apache/spark/sql/streaming/EventTimeWatermarkSuite.scala --- @@ -300,6 +300,67 @@ class EventTimeWatermarkSuite extends StreamTest with BeforeAndAfter with Matche ) } + test("watermark with 2 streams") { + val first = MemoryStream[Int] + + val firstAggregation = first.toDF() + .withColumn("eventTime", $"value".cast("timestamp")) + .withWatermark("eventTime", "10 seconds") + .select('value) + + val second = MemoryStream[Int] + + val secondAggregation = second.toDF() + .withColumn("eventTime", $"value".cast("timestamp")) + .withWatermark("eventTime", "10 seconds") + .select('value) + + val union = firstAggregation.union(secondAggregation) + .writeStream + .format("memory") + .queryName("test") + .start() + + def populateNewWatermarkFromData(stream: MemoryStream[Int], data: Int*): Unit = { + stream.addData(data) + union.processAllAvailable() + // add a dummy batch so lastExecution has the new watermark + stream.addData(0) + union.processAllAvailable() + } + + def assertQueryWatermark(watermark: Int): Unit = { --- End diff -- nit: assertWatermark.
--- --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org