yadavay-amzn commented on code in PR #56944:
URL: https://github.com/apache/spark/pull/56944#discussion_r3525849313
##########
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:
You are right - the three existing tests run in Complete mode, which never
evicts. Added a dedicated Update-mode test (groupBy on a nanosecond event-time
column) that advances the watermark and asserts state eviction
(assertNumStateRows) plus a late-row drop (assertNumRowsDroppedByWatermark), so
the scalar watermarkLiteral eviction path (LessThanOrEqual over
TimestampNanosVal) is now exercised at runtime.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/statefulOperators.scala:
##########
@@ -671,17 +672,35 @@ object WatermarkSupport {
// use the attribute itself.
val evictionExpression =
if (watermarkAttribute.dataType.isInstanceOf[StructType]) {
Review Comment:
Correct - it was inert here, since window()/session_window() over nanos is
rejected until SPARK-57829, so the struct end was always micros. I removed the
endFieldType struct-branch handling from this PR and consolidated it into
#56951 (SPARK-57829), where window-over-nanos is enabled so the struct path is
reachable and now e2e-tested. This PR stays scoped to the reachable scalar
nanos watermark.
--
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]
