Fix getMap(offset) scan-sharing bug

add null-input NPE guard
removed bogus MONTHLY interval
changenow() to current_timestamp()
Styling changes

Author: zeotuan

src/main/scala/com/amazon/deequ/analyzers/Analyzer.scala

@@ -443,7 +443,7 @@ object Preconditions {
     }
   }
 
-  /** Specified column has string type */
+  /** Specified column has a date or timestamp type */
   def isDateType(column: String): StructType => Unit = { schema =>
     val columnDataType = structField(column, schema).dataType
     val hasDateType = columnDataType match {

src/main/scala/com/amazon/deequ/analyzers/DateTimeDistribution.scala

@@ -28,7 +28,7 @@ import org.apache.spark.sql.{Column, Row}
 import scala.util.{Failure, Success}
 
 object DistributionInterval extends Enumeration {
-  val QUARTER_HOUR, HOURLY, DAILY, WEEKLY, MONTHLY = Value
+  val QUARTER_HOUR, HOURLY, DAILY, WEEKLY = Value
 }
 
 case class DateTimeDistributionState(distribution: Map[(Instant, Instant), Long])
@@ -90,7 +90,7 @@ case class DateTimeDistribution(
     ifNoNullsIn(result, offset) { _ =>
       DateTimeDistributionState(
         DateTimeDistributionState.computeStateFromResult(
-          Map.empty[Long, Long] ++ result.getMap(0),
+          Map.empty[Long, Long] ++ result.getMap(offset),
           interval
         )
       )

src/main/scala/com/amazon/deequ/analyzers/catalyst/DateTimeAggregation.scala

@@ -28,9 +28,13 @@ private[sql] class DateTimeAggregation(
   override def zero: Map[Long, Long] = Map.empty[Long, Long]
 
   override def reduce(agg: Map[Long, Long], input: Instant): Map[Long, Long] = {
-    val dateTime = input.toEpochMilli
-    val batchTime = dateTime - (dateTime % frequency)
-    agg + (batchTime -> (agg.getOrElse(batchTime, 0L) + 1L))
+    if (input == null) {
+      agg
+    } else {
+      val dateTime = input.toEpochMilli
+      val batchTime = dateTime - (dateTime % frequency)
+      agg + (batchTime -> (agg.getOrElse(batchTime, 0L) + 1L))
+    }
   }
 
   override def merge(b1: Map[Long, Long], b2: Map[Long, Long]): Map[Long, Long] = {
@@ -46,4 +50,4 @@ private[sql] class DateTimeAggregation(
 
   // Define encoder for output
   def outputEncoder: Encoder[Map[Long, Long]] = ExpressionEncoder()
-}
+}

src/main/scala/com/amazon/deequ/checks/Check.scala

@@ -1516,7 +1516,7 @@ case class Check(
     column: String,
     assertion: Double => Boolean = Check.IsOne,
     hint: Option[String] = None)
-  : CheckWithLastConstraintFilterable = satisfies(s"$column < now()",
+  : CheckWithLastConstraintFilterable = satisfies(s"$column < current_timestamp()",
     s"$column has all past dates", assertion,
     hint = hint)
 
@@ -1534,7 +1534,7 @@ case class Check(
     column: String,
     assertion: Double => Boolean = Check.IsOne,
     hint: Option[String] = None)
-  : CheckWithLastConstraintFilterable = satisfies(s"$column > now()",
+  : CheckWithLastConstraintFilterable = satisfies(s"$column > current_timestamp()",
     s"$column has all future dates", assertion,
     hint = hint)
 

src/main/scala/com/amazon/deequ/examples/DateTimeMetricExample.scala

@@ -31,7 +31,7 @@ private[examples] object DateTimeMetricExample extends App {
       Customer(3, "Thomas Yu", Instant.parse("2023-02-11T08:15:00Z")),
       Customer(4, "Steve Powell", Instant.parse("2019-04-11T12:15:00Z")),
       Customer(5, "Andrej Kar", Instant.parse("2020-08-11T12:30:00Z")),
-      Customer(6, "Ji Sung", Instant.parse("2020-08-11T12:30:00Z")),
+      Customer(6, "Ji Sung", Instant.parse("2020-08-11T12:30:00Z"))
     )
 
     val analysisResult: AnalyzerContext = { AnalysisRunner

src/test/scala/com/amazon/deequ/analyzers/AnalyzerTests.scala

@@ -545,7 +545,7 @@ class AnalyzerTests extends AnyWordSpec with Matchers with SparkContextSpec with
 
     "success for DateType column" in withSparkSessionJava8APIEnabled { sparkSession =>
       val df = getDfWithLocalDateAndInstant(sparkSession)
-      assert(DateTimeDistribution("signupDate", DistributionInterval.MONTHLY).calculate(df).value.isSuccess)
+      assert(DateTimeDistribution("signupDate", DistributionInterval.WEEKLY).calculate(df).value.isSuccess)
     }
 
     "success for Timestamp column" in withSparkSessionJava8APIEnabled { sparkSession =>

src/test/scala/com/amazon/deequ/utils/FixtureSupport.scala

@@ -155,7 +155,7 @@ trait FixtureSupport {
       (2, "marry jane", Instant.parse("2021-11-11T08:15:00Z"), LocalDate.of(2017, 10, 14)),
       (3, "Thomas Yu", Instant.parse("2021-11-11T09:15:00Z"), LocalDate.of(2017, 10, 14)),
       (4, "Steve Powell", Instant.parse("2019-04-11T12:15:00Z"), LocalDate.of(2017, 11, 14)),
-      (5, "Andrej Kar", Instant.parse("2019-04-11T13:15:00Z"), LocalDate.of(2017, 11, 14)),
+      (5, "Andrej Kar", Instant.parse("2019-04-11T13:15:00Z"), LocalDate.of(2017, 11, 14))
     ).toDF("id", "name", "dateOfBirth", "signupDate")
   }