uros-b commented on code in PR #56944:
URL: https://github.com/apache/spark/pull/56944#discussion_r3524893433
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/EventTimeWatermarkSuite.scala:
##########
@@ -1051,4 +1052,92 @@ class EventTimeWatermarkSuite extends StreamTest with
BeforeAndAfter with Matche
private def awaitTermination(): AssertOnQuery = Execute("AwaitTermination")
{ q =>
q.awaitTermination()
}
+
+ test("SPARK-57830: withWatermark accepts TimestampLTZNanosType column") {
+ val inputData = MemoryStream[Long]
+ // Use Complete mode so we don't need window-based eviction.
+ // This verifies the type-check fix and nanos->ms conversion in
EventTimeWatermarkExec.
+ val aggWithWatermark = inputData.toDF()
+ .withColumn("eventTime", timestamp_nanos($"value"))
+ .withWatermark("eventTime", "10 seconds")
+ .groupBy($"eventTime")
+ .agg(count("*") as Symbol("count"))
+ .select($"count".as[Long])
+
+ testStream(aggWithWatermark, outputMode = Complete)(
+ AddData(inputData, 15L * 1000000000L),
+ CheckAnswer(1L),
+ assertEventStats(min = 15, max = 15, avg = 15, wtrmark = 0),
+ AddData(inputData, 10L * 1000000000L, 12L * 1000000000L, 14L *
1000000000L),
+ CheckAnswer(1L, 1L, 1L, 1L),
+ assertEventStats(min = 10, max = 14, avg = 12, wtrmark = 5),
+ AddData(inputData, 25L * 1000000000L),
+ CheckAnswer(1L, 1L, 1L, 1L, 1L),
+ assertEventStats(min = 25, max = 25, avg = 25, wtrmark = 5)
+ )
+ }
+
+ test("SPARK-57830: withWatermark accepts TimestampNTZNanosType column") {
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ val inputData = MemoryStream[Long]
+ val aggWithWatermark = inputData.toDF()
+ .withColumn("eventTime",
+ timestamp_nanos($"value").cast(TimestampNTZNanosType(9)))
+ .withWatermark("eventTime", "10 seconds")
+ .groupBy($"eventTime")
+ .agg(count("*") as Symbol("count"))
+ .select($"count".as[Long])
+
+ testStream(aggWithWatermark, outputMode = Complete)(
+ AddData(inputData, 15L * 1000000000L),
+ CheckAnswer(1L),
+ assertEventStats(min = 15, max = 15, avg = 15, wtrmark = 0),
+ AddData(inputData, 10L * 1000000000L, 12L * 1000000000L, 14L *
1000000000L),
+ CheckAnswer(1L, 1L, 1L, 1L),
+ assertEventStats(min = 10, max = 14, avg = 12, wtrmark = 5),
+ AddData(inputData, 25L * 1000000000L),
+ CheckAnswer(1L, 1L, 1L, 1L, 1L),
+ assertEventStats(min = 25, max = 25, avg = 25, wtrmark = 5)
+ )
+ }
+ }
+
+ test("SPARK-57830: nanos watermark matches micros watermark for equivalent
instants") {
+ // Compare that the watermark advancement is identical between a micros
and nanos column
+ // representing the same instant.
+ val inputDataMicros = MemoryStream[Int]
+ val microsDf = inputDataMicros.toDF()
+ .withColumn("eventTime", timestamp_seconds($"value"))
+ .withWatermark("eventTime", "10 seconds")
+ .groupBy($"eventTime")
+ .agg(count("*") as Symbol("count"))
+ .select($"count".as[Long])
+
+ val inputDataNanos = MemoryStream[Long]
+ val nanosDf = inputDataNanos.toDF()
+ .withColumn("eventTime", timestamp_nanos($"value"))
+ .withWatermark("eventTime", "10 seconds")
+ .groupBy($"eventTime")
+ .agg(count("*") as Symbol("count"))
+ .select($"count".as[Long])
+
+ // Run both streams with the same logical timestamps and verify same
watermark advancement
+ testStream(microsDf, outputMode = Complete)(
+ AddData(inputDataMicros, 15),
+ CheckAnswer(1L),
+ assertEventStats(min = 15, max = 15, avg = 15, wtrmark = 0),
+ AddData(inputDataMicros, 25),
+ CheckAnswer(1L, 1L),
+ assertEventStats(min = 25, max = 25, avg = 25, wtrmark = 5)
+ )
+
+ testStream(nanosDf, outputMode = Complete)(
+ AddData(inputDataNanos, 15L * 1000000000L),
+ CheckAnswer(1L),
+ assertEventStats(min = 15, max = 15, avg = 15, wtrmark = 0),
+ AddData(inputDataNanos, 25L * 1000000000L),
+ CheckAnswer(1L, 1L),
+ assertEventStats(min = 25, max = 25, avg = 25, wtrmark = 5)
+ )
+ }
Review Comment:
All three new tests run in Complete output mode, which skips state eviction
(StateStoreSaveExec only evicts in Append/Update). The new watermarkLiteral
eviction path (both the non-struct branch and the struct endFieldType branch,
incl. LessThanOrEqual over TimestampNanosVal ordering) is therefore never
exercised at runtime, asserted only by construction. Please consider adding an
Append-mode test that forces eviction of a nano-event-time row.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
