ryan's comments

Author: marmbrus

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala

@@ -155,6 +155,16 @@ trait CheckAnalysis extends PredicateHelper {
         }
 
         operator match {
+          case etw: EventTimeWatermark =>
+            etw.eventTime.dataType match {
+              case s: StructType
+                if s.find(_.name == "start").map(_.dataType).contains(TimestampType) =>
+              case _: TimestampType =>
+              case _ =>
+                failAnalysis(
+                  s"Event time must be defined on a window or a timestamp, but " +
+                  s"${etw.eventTime.name} is of type ${etw.eventTime.dataType.simpleString}")
+            }
           case f: Filter if f.condition.dataType != BooleanType =>
             failAnalysis(
               s"filter expression '${f.condition.sql}' " +

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/EventTimeWatermarkExec.scala

@@ -26,9 +26,9 @@ import org.apache.spark.sql.types.MetadataBuilder
 import org.apache.spark.unsafe.types.CalendarInterval
 import org.apache.spark.util.AccumulatorV2
 
+/** Tracks the maximum positive long seen. */
 class MaxLong(protected var currentValue: Long = 0)
-  extends AccumulatorV2[Long, Long]
-  with Serializable {
+  extends AccumulatorV2[Long, Long] {
 
   override def isZero: Boolean = value == 0
   override def value: Long = currentValue

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

@@ -450,11 +450,14 @@ class StreamExecution(
     }.headOption.foreach { newWatermark =>
       if (newWatermark > currentEventTimeWatermark) {
         logInfo(s"Updating eventTime watermark to: $newWatermark ms")
-        streamMetrics.reportTriggerDetail(EVENT_TIME_WATERMARK, newWatermark)
         currentEventTimeWatermark = newWatermark
       } else {
         logTrace(s"Event time didn't move: $newWatermark < $currentEventTimeWatermark")
       }
+
+      if (newWatermark != 0) {
+        streamMetrics.reportTriggerDetail(EVENT_TIME_WATERMARK, newWatermark)
+      }
     }
 
     awaitBatchLock.lock()

sql/core/src/test/scala/org/apache/spark/sql/streaming/WatermarkSuite.scala

@@ -40,6 +40,16 @@ class WatermarkSuite extends StreamTest with BeforeAndAfter with Logging {
     assert(e.getMessage contains "badColumn")
   }
 
+  test("error on wrong type") {
+    val inputData = MemoryStream[Int].toDF()
+    val e = intercept[AnalysisException] {
+      inputData.withWatermark("value", "1 minute")
+    }
+    assert(e.getMessage contains "value")
+    assert(e.getMessage contains "int")
+  }
+
+
   test("watermark metric") {
     val inputData = MemoryStream[Int]
 
@@ -129,7 +139,7 @@ class WatermarkSuite extends StreamTest with BeforeAndAfter with Logging {
       CheckAnswer((10, 3)),
       AddData(inputData, 10),     // 10 is later than 15 second watermark
       CheckAnswer((10, 3)),
-      AddData(inputData, 25),     // 10 is later than 15 second watermark
+      AddData(inputData, 25),
       CheckAnswer((10, 3))        // Should not emit an incorrect partial result.
     )
   }