From 8924c9239ecc65e9c1c29a27aa90b7fd880e075a Mon Sep 17 00:00:00 2001
From: Yash0215 <55651356+Yash0215@users.noreply.github.com>
Date: Wed, 23 Sep 2020 12:46:22 +0530
Subject: [PATCH 01/14] Entity Enums name correction

---
 src/main/scala/com/amazon/deequ/metrics/Metric.scala | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/main/scala/com/amazon/deequ/metrics/Metric.scala b/src/main/scala/com/amazon/deequ/metrics/Metric.scala
index 0964d160a..77f4366b8 100644
--- a/src/main/scala/com/amazon/deequ/metrics/Metric.scala
+++ b/src/main/scala/com/amazon/deequ/metrics/Metric.scala
@@ -19,7 +19,7 @@ package com.amazon.deequ.metrics
 import scala.util.{Failure, Success, Try}
 
 object Entity extends Enumeration {
-  val Dataset, Column, Mutlicolumn = Value
+  val Dataset, Column, MultiColumn = Value
 }
 
 /** Common trait for all data quality metrics */

From 0efb7a81199a1648e79f71c2efd8c5622842f1f5 Mon Sep 17 00:00:00 2001
From: Yash0215 <yash15.chawada@gmail.com>
Date: Thu, 24 Sep 2020 11:03:08 +0530
Subject: [PATCH 02/14] Timestamp analysis support - Features: > Support
 Spark's DateType and TimestampType > min and max analysis on date time fields

---
 .../com/amazon/deequ/analyzers/Analyzer.scala | 48 ++++++++++++++-
 .../deequ/analyzers/MaximumDateTime.scala     | 57 ++++++++++++++++++
 .../deequ/analyzers/MinimumDateTime.scala     | 60 +++++++++++++++++++
 .../deequ/examples/AnalyzerExample.scala      | 36 +++++++++++
 .../amazon/deequ/examples/ExampleUtils.scala  |  5 ++
 .../com/amazon/deequ/examples/entities.scala  |  9 +++
 .../com/amazon/deequ/metrics/Metric.scala     | 20 ++++++-
 7 files changed, 232 insertions(+), 3 deletions(-)
 create mode 100644 src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala
 create mode 100644 src/main/scala/com/amazon/deequ/analyzers/MinimumDateTime.scala
 create mode 100644 src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala

diff --git a/src/main/scala/com/amazon/deequ/analyzers/Analyzer.scala b/src/main/scala/com/amazon/deequ/analyzers/Analyzer.scala
index 528003da4..d626dc427 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/Analyzer.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/Analyzer.scala
@@ -17,12 +17,11 @@
 package com.amazon.deequ.analyzers
 
 import com.amazon.deequ.analyzers.Analyzers._
-import com.amazon.deequ.metrics.{DoubleMetric, Entity, Metric}
+import com.amazon.deequ.metrics.{DoubleMetric, Entity, Metric, TimestampMetric}
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.{Column, DataFrame, Row, SparkSession}
 import com.amazon.deequ.analyzers.runners._
-
 import scala.language.existentials
 import scala.util.{Failure, Success}
 
@@ -225,6 +224,36 @@ abstract class StandardScanShareableAnalyzer[S <: DoubleValuedState[_]](
   }
 }
 
+abstract class TimestampScanShareableAnalyzer[S <: TimestampValuedState[_]](
+    name: String,
+    instance: String,
+    entity: Entity.Value = Entity.Column)
+  extends ScanShareableAnalyzer[S, TimestampMetric] {
+
+  override def computeMetricFrom(state: Option[S]): TimestampMetric = {
+    state match {
+      case Some(theState) =>
+        TimestampMetric(entity, name, instance, Success(theState.metricValue()))
+      case _ =>
+        TimestampMetric(entity, name, instance, Failure(
+          MetricCalculationException.wrapIfNecessary(emptyStateException(this))))
+    }
+  }
+
+  override private[deequ] def toFailureMetric(exception: Exception): TimestampMetric = {
+    TimestampMetric(entity, name, instance, Failure(
+      MetricCalculationException.wrapIfNecessary(exception)))
+  }
+
+  override def preconditions: Seq[StructType => Unit] = {
+    additionalPreconditions() ++ super.preconditions
+  }
+
+  protected def additionalPreconditions(): Seq[StructType => Unit] = {
+    Seq.empty
+  }
+}
+
 /** A state for computing ratio-based metrics,
   * contains #rows that match a predicate and overall #rows */
 case class NumMatchesAndCount(numMatches: Long, count: Long)
@@ -287,6 +316,9 @@ object Preconditions {
   private[this] val numericDataTypes =
     Set(ByteType, ShortType, IntegerType, LongType, FloatType, DoubleType, DecimalType)
 
+  private[this] val dateTypes =
+    Set(TimestampType, DateType)
+
   private[this] val nestedDataTypes = Set(StructType, MapType, ArrayType)
 
   private[this] val caseSensitive = {
@@ -380,6 +412,18 @@ object Preconditions {
     }
   }
 
+  def isDateType(column: String): StructType => Unit = { schema =>
+    val columnDataType = structField(column, schema).dataType
+    val hasDateType = columnDataType match {
+      case TimestampType | DateType => true
+      case _ => false
+    }
+    if (!hasDateType) {
+      throw new WrongColumnTypeException(s"Expected type of column $column to be one of " +
+        s"(${dateTypes.mkString(",")}), but found $columnDataType instead!")
+    }
+  }
+
   /** Specified column has string type */
   def isString(column: String): StructType => Unit = { schema =>
     val columnDataType = structField(column, schema).dataType
diff --git a/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala b/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala
new file mode 100644
index 000000000..6b644dc18
--- /dev/null
+++ b/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala
@@ -0,0 +1,57 @@
+/**
+ * Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ * use this file except in compliance with the License. A copy of the License
+ * is located at
+ *
+ *     http://aws.amazon.com/apache2.0/
+ *
+ * or in the "license" file accompanying this file. This file is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ *
+ */
+
+package com.amazon.deequ.analyzers
+
+import com.amazon.deequ.analyzers.Preconditions.{hasColumn, isDateType}
+import org.apache.spark.sql.{Column, Row}
+import org.apache.spark.sql.functions.max
+import org.apache.spark.sql.types.{TimestampType, StructType}
+import Analyzers._
+import java.sql.Timestamp
+
+case class MaxTimestampState(maxValue: Timestamp) extends TimestampValuedState[MaxTimestampState] {
+
+  override def sum(other: MaxTimestampState): MaxTimestampState = {
+    MaxTimestampState(if(maxValue.compareTo(other.maxValue) > 0) maxValue else other.maxValue)
+  }
+
+  override def metricValue(): Timestamp = {
+    maxValue
+  }
+}
+
+case class MaximumDateTime(column: String, where: Option[String] = None)
+  extends TimestampScanShareableAnalyzer[MaxTimestampState]("Maximum Date Time", column)
+  with FilterableAnalyzer {
+
+  override def aggregationFunctions(): Seq[Column] = {
+    max(conditionalSelection(column, where)).cast(TimestampType) :: Nil
+  }
+
+  override def fromAggregationResult(result: Row, offset: Int): Option[MaxTimestampState] = {
+    ifNoNullsIn(result, offset) { _ =>
+        println("getting " + result.getTimestamp(offset))
+        MaxTimestampState(result.getTimestamp(offset))
+    }
+  }
+
+  override protected def additionalPreconditions(): Seq[StructType => Unit] = {
+    hasColumn(column) :: isDateType(column) :: Nil
+  }
+
+  override def filterCondition: Option[String] = where
+}
\ No newline at end of file
diff --git a/src/main/scala/com/amazon/deequ/analyzers/MinimumDateTime.scala b/src/main/scala/com/amazon/deequ/analyzers/MinimumDateTime.scala
new file mode 100644
index 000000000..d5f7228fe
--- /dev/null
+++ b/src/main/scala/com/amazon/deequ/analyzers/MinimumDateTime.scala
@@ -0,0 +1,60 @@
+/**
+ * Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ * use this file except in compliance with the License. A copy of the License
+ * is located at
+ *
+ *     http://aws.amazon.com/apache2.0/
+ *
+ * or in the "license" file accompanying this file. This file is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ *
+ */
+
+package com.amazon.deequ.analyzers
+
+import com.amazon.deequ.analyzers.Preconditions.{hasColumn, isDateType}
+import org.apache.spark.sql.{Column, Row}
+import org.apache.spark.sql.functions.min
+import org.apache.spark.sql.types.{TimestampType, StructType}
+import Analyzers._
+import java.sql.Timestamp
+
+trait TimestampValuedState[S <: TimestampValuedState[S]] extends State[S] {
+  def metricValue(): Timestamp
+}
+
+case class MinTimestampState(minValue: Timestamp) extends TimestampValuedState[MinTimestampState] {
+
+  override def sum(other: MinTimestampState): MinTimestampState = {
+    MinTimestampState(if(minValue.compareTo(other.minValue) < 0) minValue else other.minValue)
+  }
+
+  override def metricValue(): Timestamp = {
+    minValue
+  }
+}
+
+case class MinimumDateTime(column: String, where: Option[String] = None)
+  extends TimestampScanShareableAnalyzer[MinTimestampState]("Minimum Date Time", column)
+  with FilterableAnalyzer {
+
+  override def aggregationFunctions(): Seq[Column] = {
+    min(conditionalSelection(column, where)).cast(TimestampType) :: Nil
+  }
+
+  override def fromAggregationResult(result: Row, offset: Int): Option[MinTimestampState] = {
+    ifNoNullsIn(result, offset) { _ =>
+        MinTimestampState(result.getTimestamp(offset))
+    }
+  }
+
+  override protected def additionalPreconditions(): Seq[StructType => Unit] = {
+    hasColumn(column) :: isDateType(column) :: Nil
+  }
+
+  override def filterCondition: Option[String] = where
+}
diff --git a/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
new file mode 100644
index 000000000..9651b6135
--- /dev/null
+++ b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
@@ -0,0 +1,36 @@
+package com.amazon.deequ.examples
+
+import java.math.BigDecimal
+import java.sql.Timestamp
+
+import com.amazon.deequ.analyzers.runners.{AnalysisRunner, AnalyzerContext}
+import com.amazon.deequ.analyzers.{MaximumDateTime, MinimumDateTime}
+import com.amazon.deequ.examples.ExampleUtils.{ordersAsDataframe, withSpark}
+import com.amazon.deequ.analyzers.runners.AnalyzerContext.successMetricsAsDataFrame
+
+private[examples] object AnalyzerExample extends App {
+  withSpark { session =>
+
+    val data = ordersAsDataframe(session,
+      Order(1, new BigDecimal(213.2132), Timestamp.valueOf("2020-02-15 07:15:00")),
+      Order(2, new BigDecimal(43.21324432876), Timestamp.valueOf("2020-02-15 07:45:00.999")),
+      Order(3, new BigDecimal(56.8881238823888), Timestamp.valueOf("2020-02-15 08:15:49.786")),
+      Order(4, new BigDecimal(101.2324434978788), Timestamp.valueOf("2020-02-15 12:15:00")),
+      Order(5, new BigDecimal(723.234324234324324324), Timestamp.valueOf("2020-02-15 15:14:23.678"))
+    )
+
+    val analysisResult: AnalyzerContext = { AnalysisRunner
+      .onData(data)
+      .addAnalyzer(MinimumDateTime("orderDate"))
+      .addAnalyzer(MaximumDateTime("orderDate"))
+      .run()
+    }
+
+    successMetricsAsDataFrame(session, analysisResult).show(false)
+
+    analysisResult.metricMap.foreach( x =>
+      println(s"column '${x._2.instance}' has ${x._2.name} : ${x._2.value.get}")
+    )
+
+  }
+}
diff --git a/src/main/scala/com/amazon/deequ/examples/ExampleUtils.scala b/src/main/scala/com/amazon/deequ/examples/ExampleUtils.scala
index 699711a5d..44f305be7 100644
--- a/src/main/scala/com/amazon/deequ/examples/ExampleUtils.scala
+++ b/src/main/scala/com/amazon/deequ/examples/ExampleUtils.scala
@@ -45,4 +45,9 @@ private[deequ] object ExampleUtils {
     val rdd = session.sparkContext.parallelize(manufacturers)
     session.createDataFrame(rdd)
   }
+
+  def ordersAsDataframe(session: SparkSession, orders: Order*): DataFrame = {
+    val rdd = session.sparkContext.parallelize(orders)
+    session.createDataFrame(rdd)
+  }
 }
diff --git a/src/main/scala/com/amazon/deequ/examples/entities.scala b/src/main/scala/com/amazon/deequ/examples/entities.scala
index f2750ecfe..ebd81eb64 100644
--- a/src/main/scala/com/amazon/deequ/examples/entities.scala
+++ b/src/main/scala/com/amazon/deequ/examples/entities.scala
@@ -16,6 +16,9 @@
 
 package com.amazon.deequ.examples
 
+import java.sql.Timestamp
+import java.math.BigDecimal
+
 private[deequ] case class Item(
     id: Long,
     productName: String,
@@ -24,6 +27,12 @@ private[deequ] case class Item(
     numViews: Long
 )
 
+private[deequ] case class Order(
+    id: Long,
+    amount: BigDecimal,
+    orderDate: Timestamp
+)
+
 private[deequ] case class Manufacturer(
     id: Long,
     manufacturerName: String,
diff --git a/src/main/scala/com/amazon/deequ/metrics/Metric.scala b/src/main/scala/com/amazon/deequ/metrics/Metric.scala
index 77f4366b8..97e9665d1 100644
--- a/src/main/scala/com/amazon/deequ/metrics/Metric.scala
+++ b/src/main/scala/com/amazon/deequ/metrics/Metric.scala
@@ -16,10 +16,12 @@
 
 package com.amazon.deequ.metrics
 
+import java.sql.Timestamp
+
 import scala.util.{Failure, Success, Try}
 
 object Entity extends Enumeration {
-  val Dataset, Column, MultiColumn = Value
+  val Dataset, Column, Mutlicolumn = Value
 }
 
 /** Common trait for all data quality metrics */
@@ -66,3 +68,19 @@ case class KeyedDoubleMetric(
     }
   }
 }
+
+case class TimestampMetric(
+    entity: Entity.Value,
+    name: String,
+    instance: String,
+    value: Try[Timestamp])
+  extends Metric[Timestamp] {
+
+  override def flatten(): Seq[DoubleMetric] =  {
+    if(value.isSuccess){
+      Seq(DoubleMetric(entity, "Timestamp milliseconds", instance, Success(value.get.getTime.toDouble)))
+    } else {
+      Seq(DoubleMetric(entity, "Timestamp milliseconds", instance, Failure(value.failed.get)))
+    }
+  }
+}
\ No newline at end of file

From 3f6e27e465c7b2fe9a05039bed2cd08692544b87 Mon Sep 17 00:00:00 2001
From: Yash0215 <yash15.chawada@gmail.com>
Date: Thu, 24 Sep 2020 14:39:25 +0530
Subject: [PATCH 03/14] Feature Added: > New constraints for Spark's DateType
 and Timestamp Types

---
 .../com/amazon/deequ/analyzers/Analyzer.scala |   2 +
 .../deequ/analyzers/MaximumDateTime.scala     |   1 -
 .../scala/com/amazon/deequ/checks/Check.scala | 126 +++++++++++++++++-
 .../amazon/deequ/constraints/Constraint.scala |  33 ++++-
 .../deequ/examples/AnalyzerExample.scala      |   3 +
 5 files changed, 160 insertions(+), 5 deletions(-)

diff --git a/src/main/scala/com/amazon/deequ/analyzers/Analyzer.scala b/src/main/scala/com/amazon/deequ/analyzers/Analyzer.scala
index d626dc427..7e5fea1e4 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/Analyzer.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/Analyzer.scala
@@ -336,6 +336,8 @@ object Preconditions {
     }
   }
 
+
+
   def hasColumn(column: String, schema: StructType): Boolean = {
     if (caseSensitive) {
       schema.fieldNames.contains(column)
diff --git a/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala b/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala
index 6b644dc18..20c2b7ef5 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala
@@ -44,7 +44,6 @@ case class MaximumDateTime(column: String, where: Option[String] = None)
 
   override def fromAggregationResult(result: Row, offset: Int): Option[MaxTimestampState] = {
     ifNoNullsIn(result, offset) { _ =>
-        println("getting " + result.getTimestamp(offset))
         MaxTimestampState(result.getTimestamp(offset))
     }
   }
diff --git a/src/main/scala/com/amazon/deequ/checks/Check.scala b/src/main/scala/com/amazon/deequ/checks/Check.scala
index c71652f14..5fb194449 100644
--- a/src/main/scala/com/amazon/deequ/checks/Check.scala
+++ b/src/main/scala/com/amazon/deequ/checks/Check.scala
@@ -16,16 +16,18 @@
 
 package com.amazon.deequ.checks
 
+import java.sql.Timestamp
+
 import com.amazon.deequ.anomalydetection.{AnomalyDetectionStrategy, AnomalyDetector, DataPoint}
 import com.amazon.deequ.analyzers.runners.AnalyzerContext
-import com.amazon.deequ.analyzers.{Analyzer, Histogram, Patterns, State, KLLParameters}
-import com.amazon.deequ.constraints.Constraint._
+import com.amazon.deequ.analyzers.{Analyzer, Histogram, KLLParameters, Patterns, State}
+import com.amazon.deequ.constraints.Constraint.{maxTimestampConstraint, minTimestampConstraint, _}
 import com.amazon.deequ.constraints._
 import com.amazon.deequ.metrics.{BucketDistribution, Distribution, Metric}
 import com.amazon.deequ.repository.MetricsRepository
 import org.apache.spark.sql.expressions.UserDefinedFunction
 import com.amazon.deequ.anomalydetection.HistoryUtils
-import com.amazon.deequ.checks.ColumnCondition.{isEachNotNull, isAnyNotNull}
+import com.amazon.deequ.checks.ColumnCondition.{isAnyNotNull, isEachNotNull}
 
 import scala.util.matching.Regex
 
@@ -1016,6 +1018,124 @@ case class Check(
     satisfies(predicate, s"$column between $lowerBound and $upperBound", hint = hint)
   }
 
+  /**
+   * Creates a constraint that asserts a minimum java.sql.Timestamp
+   *
+   * @param column – Column to run the assertion on
+   * @param assertion – Function that receives a java.sql.Timestamp input parameter and returns a boolean
+   * @param hint – A hint to provide additional context why a constraint could have failed
+   * @return
+   */
+  def hasMinTimestamp(
+      column: String,
+      assertion: Timestamp => Boolean,
+      hint: Option[String] = None)
+  : CheckWithLastConstraintFilterable = {
+
+    addFilterableConstraint { filter => minTimestampConstraint(column, assertion, filter, hint) }
+  }
+
+  /**
+   * Creates a constraint that asserts a maximum java.sql.Timestamp
+   *
+   * @param column – Column to run the assertion on
+   * @param assertion – Function that receives a java.sql.Timestamp input parameter and returns a boolean
+   * @param hint – A hint to provide additional context why a constraint could have failed
+   * @return
+   */
+  def hasMaxTimestamp(
+      column: String,
+      assertion: Timestamp => Boolean,
+      hint: Option[String] = None)
+  : CheckWithLastConstraintFilterable = {
+
+    addFilterableConstraint { filter => maxTimestampConstraint(column, assertion, filter, hint) }
+  }
+
+  /**
+   *
+   * Asserts that, in each row, the value of column (DateType or TimestampType) is less than the given datetime (Timestamp)
+   *
+   * @param column Column to run the assertion on
+   * @param datetime value of Timestamp to run assert
+   * @param assertion Function that receives a Timestamp input parameter and returns a boolean
+   * @param hint A hint to provide additional context why a constraint could have failed
+   * @return
+   */
+  def isDateTimeLessThan(
+      column: String,
+      datetime: Timestamp,
+      assertion: Double => Boolean = Check.IsOne,
+      hint: Option[String] = None)
+  : CheckWithLastConstraintFilterable = {
+
+    satisfies(s"$column < to_timestamp('${datetime.toString}')",
+      s"$column is less than '${datetime.toString}'", assertion,
+      hint = hint)
+  }
+
+  /**
+   *
+   * Asserts that, in each row, the value of column (DateType or TimestampType) is greater than the given datetime (Timestamp)
+   *
+   * @param column Column to run the assertion on
+   * @param datetime value of Timestamp to run assert
+   * @param assertion Function that receives a Timestamp input parameter and returns a boolean
+   * @param hint A hint to provide additional context why a constraint could have failed
+   * @return
+   */
+  def isDateTimeGreaterThan(
+      column: String,
+      datetime: Timestamp,
+      assertion: Double => Boolean = Check.IsOne,
+      hint: Option[String] = None)
+  : CheckWithLastConstraintFilterable = {
+
+    satisfies(s"$column > to_timestamp('${datetime.toString}')",
+      s"$column is greater than '${datetime.toString}'", assertion,
+      hint = hint)
+  }
+
+  /**
+   *
+   * Asserts that, in each row, the value of column (DateType or TimestampType) contains a past date
+   *
+   * @param column Column to run the assertion on
+   * @param assertion Function that receives a Timestamp input parameter and returns a boolean
+   * @param hint A hint to provide additional context why a constraint could have failed
+   * @return
+   */
+  def hasPastDates(
+      column: String,
+      assertion: Double => Boolean = Check.IsOne,
+      hint: Option[String] = None)
+  : CheckWithLastConstraintFilterable = {
+
+    satisfies(s"$column < now()",
+      s"$column has all past dates", assertion,
+      hint = hint)
+  }
+
+  /**
+   *
+   * Asserts that, in each row, the value of column (DateType or TimestampType) contains a future date
+   *
+   * @param column Column to run the assertion on
+   * @param assertion Function that receives a Timestamp input parameter and returns a boolean
+   * @param hint A hint to provide additional context why a constraint could have failed
+   * @return
+   */
+  def hasFutureDates(
+       column: String,
+       assertion: Double => Boolean = Check.IsOne,
+       hint: Option[String] = None)
+  : CheckWithLastConstraintFilterable = {
+
+    satisfies(s"$column > now()",
+      s"$column has all future dates", assertion,
+      hint = hint)
+  }
+
   /**
     * Evaluate this check on computed metrics
     * @param context result of the metrics computation
diff --git a/src/main/scala/com/amazon/deequ/constraints/Constraint.scala b/src/main/scala/com/amazon/deequ/constraints/Constraint.scala
index c7963ce41..495226cd1 100644
--- a/src/main/scala/com/amazon/deequ/constraints/Constraint.scala
+++ b/src/main/scala/com/amazon/deequ/constraints/Constraint.scala
@@ -16,7 +16,9 @@
 
 package com.amazon.deequ.constraints
 
-import com.amazon.deequ.analyzers._
+import java.sql.Timestamp
+
+import com.amazon.deequ.analyzers.{MaximumDateTime, _}
 import com.amazon.deequ.metrics.{BucketDistribution, Distribution, Metric}
 import org.apache.spark.sql.expressions.UserDefinedFunction
 
@@ -484,6 +486,35 @@ object Constraint {
     new NamedConstraint(constraint, s"MaximumConstraint($maximum)")
   }
 
+  def minTimestampConstraint(
+      column: String,
+      assertion: Timestamp => Boolean,
+      where: Option[String] = None,
+      hint: Option[String] = None)
+  : Constraint = {
+
+    val minimum = MinimumDateTime(column, where)
+
+    val constraint = AnalysisBasedConstraint[MinTimestampState, Timestamp, Timestamp](minimum, assertion,
+      hint = hint)
+
+    new NamedConstraint(constraint, s"MinimumTimestampConstraint($minimum)")
+  }
+
+  def maxTimestampConstraint(
+      column: String,
+      assertion: Timestamp => Boolean,
+      where: Option[String] = None,
+      hint: Option[String] = None)
+  : Constraint = {
+
+    val maximum = MaximumDateTime(column, where)
+
+    val constraint = AnalysisBasedConstraint[MaxTimestampState, Timestamp, Timestamp](maximum, assertion,
+      hint = hint)
+
+    new NamedConstraint(constraint, s"MaximumTimestampConstraint($maximum)")
+  }
   /**
     * Runs mean analysis on the given column and executes the assertion
     *
diff --git a/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
index 9651b6135..abec67ae3 100644
--- a/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
+++ b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
@@ -3,10 +3,13 @@ package com.amazon.deequ.examples
 import java.math.BigDecimal
 import java.sql.Timestamp
 
+import com.amazon.deequ.VerificationSuite
 import com.amazon.deequ.analyzers.runners.{AnalysisRunner, AnalyzerContext}
 import com.amazon.deequ.analyzers.{MaximumDateTime, MinimumDateTime}
 import com.amazon.deequ.examples.ExampleUtils.{ordersAsDataframe, withSpark}
 import com.amazon.deequ.analyzers.runners.AnalyzerContext.successMetricsAsDataFrame
+import com.amazon.deequ.checks.{Check, CheckLevel, CheckStatus}
+import com.amazon.deequ.constraints.ConstraintStatus
 
 private[examples] object AnalyzerExample extends App {
   withSpark { session =>

From 927ed981c03b77d585eea94d8c37927c9263770f Mon Sep 17 00:00:00 2001
From: Yash0215 <yash15.chawada@gmail.com>
Date: Thu, 24 Sep 2020 20:24:47 +0530
Subject: [PATCH 04/14] Feature Added: > BigDecimal State and Metric Support in
 Min, Max, Sum and Mean Analyzers

---
 .../com/amazon/deequ/analyzers/Analyzer.scala | 77 +++++++++++++++++--
 .../com/amazon/deequ/analyzers/Maximum.scala  | 39 +++++++++-
 .../deequ/analyzers/MaximumDateTime.scala     | 12 +--
 .../com/amazon/deequ/analyzers/Mean.scala     | 40 +++++++++-
 .../com/amazon/deequ/analyzers/Minimum.scala  | 37 ++++++++-
 .../deequ/analyzers/MinimumDateTime.scala     | 14 ++--
 .../com/amazon/deequ/analyzers/Sum.scala      | 36 ++++++++-
 .../amazon/deequ/constraints/Constraint.scala |  4 +-
 .../deequ/examples/AnalyzerExample.scala      |  6 +-
 .../com/amazon/deequ/metrics/Metric.scala     | 18 ++++-
 10 files changed, 252 insertions(+), 31 deletions(-)

diff --git a/src/main/scala/com/amazon/deequ/analyzers/Analyzer.scala b/src/main/scala/com/amazon/deequ/analyzers/Analyzer.scala
index 7e5fea1e4..519230a28 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/Analyzer.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/Analyzer.scala
@@ -16,12 +16,15 @@
 
 package com.amazon.deequ.analyzers
 
+import java.math.BigDecimal
+
 import com.amazon.deequ.analyzers.Analyzers._
-import com.amazon.deequ.metrics.{DoubleMetric, Entity, Metric, TimestampMetric}
+import com.amazon.deequ.metrics.{BigDecimalMetric, DoubleMetric, Entity, Metric, DateTimeMetric}
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.{Column, DataFrame, Row, SparkSession}
 import com.amazon.deequ.analyzers.runners._
+
 import scala.language.existentials
 import scala.util.{Failure, Success}
 
@@ -51,6 +54,11 @@ trait DoubleValuedState[S <: DoubleValuedState[S]] extends State[S] {
   def metricValue(): Double
 }
 
+/** A state which produces a BigDecimalValued metric  */
+trait BigDecimalValuedState[S <: BigDecimalValuedState[S]] extends State[S] {
+  def metricValue(): BigDecimal
+}
+
 /** Common trait for all analyzers which generates metrics from states computed on data frames */
 trait Analyzer[S <: State[_], +M <: Metric[_]] {
 
@@ -224,24 +232,56 @@ abstract class StandardScanShareableAnalyzer[S <: DoubleValuedState[_]](
   }
 }
 
-abstract class TimestampScanShareableAnalyzer[S <: TimestampValuedState[_]](
+/** A scan-shareable analyzer that produces a DateTimeMetric */
+abstract class TimestampScanShareableAnalyzer[S <: DateTimeValuedState[_]](
+    name: String,
+    instance: String,
+    entity: Entity.Value = Entity.Column)
+  extends ScanShareableAnalyzer[S, DateTimeMetric] {
+
+  override def computeMetricFrom(state: Option[S]): DateTimeMetric = {
+    state match {
+      case Some(theState) =>
+        DateTimeMetric(entity, name, instance, Success(theState.metricValue()))
+      case _ =>
+        DateTimeMetric(entity, name, instance, Failure(
+          MetricCalculationException.wrapIfNecessary(emptyStateException(this))))
+    }
+  }
+
+  override private[deequ] def toFailureMetric(exception: Exception): DateTimeMetric = {
+    DateTimeMetric(entity, name, instance, Failure(
+      MetricCalculationException.wrapIfNecessary(exception)))
+  }
+
+  override def preconditions: Seq[StructType => Unit] = {
+    additionalPreconditions() ++ super.preconditions
+  }
+
+  protected def additionalPreconditions(): Seq[StructType => Unit] = {
+    Seq.empty
+  }
+}
+
+/** A scan-shareable analyzer that produces a BigDecimalMetric */
+abstract class BigDecimalScanShareableAnalyzer[S <: BigDecimalValuedState[_]](
     name: String,
     instance: String,
     entity: Entity.Value = Entity.Column)
-  extends ScanShareableAnalyzer[S, TimestampMetric] {
+  extends ScanShareableAnalyzer[S, BigDecimalMetric] {
 
-  override def computeMetricFrom(state: Option[S]): TimestampMetric = {
+  override def computeMetricFrom(state: Option[S]): BigDecimalMetric = {
     state match {
       case Some(theState) =>
-        TimestampMetric(entity, name, instance, Success(theState.metricValue()))
+        BigDecimalMetric(entity, name, instance, Success(theState.metricValue()))
       case _ =>
-        TimestampMetric(entity, name, instance, Failure(
+        BigDecimalMetric(entity, name, instance, Failure(
           MetricCalculationException.wrapIfNecessary(emptyStateException(this))))
     }
   }
 
-  override private[deequ] def toFailureMetric(exception: Exception): TimestampMetric = {
-    TimestampMetric(entity, name, instance, Failure(
+  override private[deequ] def toFailureMetric(exception: Exception): BigDecimalMetric = {
+    BigDecimalMetric(entity, name, instance, Failure(
       MetricCalculationException.wrapIfNecessary(exception)))
   }
 
@@ -414,6 +454,10 @@ object Preconditions {
     }
   }
 
+  /** Asserts if Specified column is a DateType or TimestampType type throw Exception if not
+   * @param column for which assertion is performed
+   * @return
+   * */
   def isDateType(column: String): StructType => Unit = { schema =>
     val columnDataType = structField(column, schema).dataType
     val hasDateType = columnDataType match {
@@ -426,6 +470,23 @@ object Preconditions {
     }
   }
 
+  /** Asserts if Specified column is a Decimal type throw Exception if not
+   * @param column for which assertion is performed
+   * @return
+   * */
+  def isDecimalType(column: String): StructType => Unit = { schema =>
+    val columnDataType = structField(column, schema).dataType
+    val hasNumericType = columnDataType match {
+      case _ : DecimalType => true
+      case _ => false
+    }
+
+    if (!hasNumericType) {
+      throw new WrongColumnTypeException(s"Expected type of column $column to be one of " +
+        s"(${numericDataTypes.mkString(",")}), but found $columnDataType instead!")
+    }
+  }
+
   /** Specified column has string type */
   def isString(column: String): StructType => Unit = { schema =>
     val columnDataType = structField(column, schema).dataType
diff --git a/src/main/scala/com/amazon/deequ/analyzers/Maximum.scala b/src/main/scala/com/amazon/deequ/analyzers/Maximum.scala
index cb4532e38..cdf086031 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/Maximum.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/Maximum.scala
@@ -16,9 +16,11 @@
 
 package com.amazon.deequ.analyzers
 
-import com.amazon.deequ.analyzers.Preconditions.{hasColumn, isNumeric}
+import java.math.BigDecimal
+
+import com.amazon.deequ.analyzers.Preconditions.{hasColumn, isDecimalType, isNumeric}
 import org.apache.spark.sql.{Column, Row}
-import org.apache.spark.sql.functions.max
+import org.apache.spark.sql.functions.{max, min}
 import org.apache.spark.sql.types.{DoubleType, StructType}
 import Analyzers._
 
@@ -54,3 +56,36 @@ case class Maximum(column: String, where: Option[String] = None)
 
   override def filterCondition: Option[String] = where
 }
+
+case class MaxBigDecimalState(minValue: BigDecimal) extends BigDecimalValuedState[MaxBigDecimalState] {
+
+  override def sum(other: MaxBigDecimalState): MaxBigDecimalState = {
+    MaxBigDecimalState(minValue.max(other.minValue))
+  }
+
+  override def metricValue(): BigDecimal = {
+    minValue
+  }
+}
+
+case class MaximumBigDecimal(column: String, where: Option[String] = None)
+  extends BigDecimalScanShareableAnalyzer[MaxBigDecimalState]("Maximum BigDecimal", column)
+    with FilterableAnalyzer {
+
+  override def aggregationFunctions(): Seq[Column] = {
+    max(conditionalSelection(column, where)) :: Nil
+  }
+
+
+  override def fromAggregationResult(result: Row, offset: Int): Option[MaxBigDecimalState] = {
+    ifNoNullsIn(result, offset) { _ =>
+      MaxBigDecimalState(result.getDecimal(offset))
+    }
+  }
+
+  override protected def additionalPreconditions(): Seq[StructType => Unit] = {
+    hasColumn(column) :: isDecimalType(column) :: Nil
+  }
+
+  override def filterCondition: Option[String] = where
+}
\ No newline at end of file
diff --git a/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala b/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala
index 20c2b7ef5..eb26bbc74 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala
@@ -23,10 +23,10 @@ import org.apache.spark.sql.types.{TimestampType, StructType}
 import Analyzers._
 import java.sql.Timestamp
 
-case class MaxTimestampState(maxValue: Timestamp) extends TimestampValuedState[MaxTimestampState] {
+case class MaxDateTimeState(maxValue: Timestamp) extends DateTimeValuedState[MaxDateTimeState] {
 
-  override def sum(other: MaxTimestampState): MaxTimestampState = {
-    MaxTimestampState(if(maxValue.compareTo(other.maxValue) > 0) maxValue else other.maxValue)
+  override def sum(other: MaxDateTimeState): MaxDateTimeState = {
+    MaxDateTimeState(if(maxValue.compareTo(other.maxValue) > 0) maxValue else other.maxValue)
   }
 
   override def metricValue(): Timestamp = {
@@ -35,16 +35,16 @@ case class MaxTimestampState(maxValue: Timestamp) extends TimestampValuedState[M
 }
 
 case class MaximumDateTime(column: String, where: Option[String] = None)
-  extends TimestampScanShareableAnalyzer[MaxTimestampState]("Maximum Date Time", column)
+  extends TimestampScanShareableAnalyzer[MaxDateTimeState]("Maximum Date Time", column)
   with FilterableAnalyzer {
 
   override def aggregationFunctions(): Seq[Column] = {
     max(conditionalSelection(column, where)).cast(TimestampType) :: Nil
   }
 
-  override def fromAggregationResult(result: Row, offset: Int): Option[MaxTimestampState] = {
+  override def fromAggregationResult(result: Row, offset: Int): Option[MaxDateTimeState] = {
     ifNoNullsIn(result, offset) { _ =>
-        MaxTimestampState(result.getTimestamp(offset))
+        MaxDateTimeState(result.getTimestamp(offset))
     }
   }
 
diff --git a/src/main/scala/com/amazon/deequ/analyzers/Mean.scala b/src/main/scala/com/amazon/deequ/analyzers/Mean.scala
index d9afbd227..44792b92d 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/Mean.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/Mean.scala
@@ -16,10 +16,12 @@
 
 package com.amazon.deequ.analyzers
 
-import com.amazon.deequ.analyzers.Preconditions.{hasColumn, isNumeric}
+import java.math.BigDecimal
+
+import com.amazon.deequ.analyzers.Preconditions.{hasColumn, isDecimalType, isNumeric}
 import org.apache.spark.sql.{Column, Row}
 import org.apache.spark.sql.functions.{count, sum}
-import org.apache.spark.sql.types.{DoubleType, StructType, LongType}
+import org.apache.spark.sql.types.{DoubleType, LongType, StructType}
 import Analyzers._
 
 case class MeanState(sum: Double, count: Long) extends DoubleValuedState[MeanState] {
@@ -55,3 +57,37 @@ case class Mean(column: String, where: Option[String] = None)
 
   override def filterCondition: Option[String] = where
 }
+
+case class BigDecimalMeanState(sum: BigDecimal, count: Long) extends BigDecimalValuedState[BigDecimalMeanState] {
+
+  override def sum(other: BigDecimalMeanState): BigDecimalMeanState = {
+    BigDecimalMeanState(sum.add(other.sum), count + other.count)
+  }
+
+  override def metricValue(): BigDecimal = {
+    if (count == 0L) null else sum.divide(new BigDecimal(count))
+  }
+}
+
+case class BigDecimalMean(column: String, where: Option[String] = None)
+  extends BigDecimalScanShareableAnalyzer[BigDecimalMeanState]("BigDecimal Mean", column)
+    with FilterableAnalyzer {
+
+  override def aggregationFunctions(): Seq[Column] = {
+    sum(conditionalSelection(column, where)) ::
+      count(conditionalSelection(column, where)).cast(LongType) :: Nil
+  }
+
+  override def fromAggregationResult(result: Row, offset: Int): Option[BigDecimalMeanState] = {
+
+    ifNoNullsIn(result, offset, howMany = 2) { _ =>
+      BigDecimalMeanState(result.getDecimal(offset), result.getLong(offset + 1))
+    }
+  }
+
+  override protected def additionalPreconditions(): Seq[StructType => Unit] = {
+    hasColumn(column) :: isDecimalType(column) :: Nil
+  }
+
+  override def filterCondition: Option[String] = where
+}
\ No newline at end of file
diff --git a/src/main/scala/com/amazon/deequ/analyzers/Minimum.scala b/src/main/scala/com/amazon/deequ/analyzers/Minimum.scala
index 68ef926cf..c98c2ed1b 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/Minimum.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/Minimum.scala
@@ -16,7 +16,9 @@
 
 package com.amazon.deequ.analyzers
 
-import com.amazon.deequ.analyzers.Preconditions.{hasColumn, isNumeric}
+import java.math.BigDecimal
+
+import com.amazon.deequ.analyzers.Preconditions.{hasColumn, isDecimalType, isNumeric}
 import org.apache.spark.sql.{Column, Row}
 import org.apache.spark.sql.functions.min
 import org.apache.spark.sql.types.{DoubleType, StructType}
@@ -54,3 +56,36 @@ case class Minimum(column: String, where: Option[String] = None)
 
   override def filterCondition: Option[String] = where
 }
+
+case class MinBigDecimalState(minValue: BigDecimal) extends BigDecimalValuedState[MinBigDecimalState] {
+
+  override def sum(other: MinBigDecimalState): MinBigDecimalState = {
+    MinBigDecimalState(minValue.min(other.minValue))
+  }
+
+  override def metricValue(): BigDecimal = {
+    minValue
+  }
+}
+
+case class MinimumBigDecimal(column: String, where: Option[String] = None)
+  extends BigDecimalScanShareableAnalyzer[MinBigDecimalState]("Minimum BigDecimal", column)
+    with FilterableAnalyzer {
+
+  override def aggregationFunctions(): Seq[Column] = {
+    min(conditionalSelection(column, where)) :: Nil
+  }
+
+
+  override def fromAggregationResult(result: Row, offset: Int): Option[MinBigDecimalState] = {
+    ifNoNullsIn(result, offset) { _ =>
+      MinBigDecimalState(result.getDecimal(offset))
+    }
+  }
+
+  override protected def additionalPreconditions(): Seq[StructType => Unit] = {
+    hasColumn(column) :: isDecimalType(column) :: Nil
+  }
+
+  override def filterCondition: Option[String] = where
+}
\ No newline at end of file
diff --git a/src/main/scala/com/amazon/deequ/analyzers/MinimumDateTime.scala b/src/main/scala/com/amazon/deequ/analyzers/MinimumDateTime.scala
index d5f7228fe..967e3f390 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/MinimumDateTime.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/MinimumDateTime.scala
@@ -23,14 +23,14 @@ import org.apache.spark.sql.types.{TimestampType, StructType}
 import Analyzers._
 import java.sql.Timestamp
 
-trait TimestampValuedState[S <: TimestampValuedState[S]] extends State[S] {
+trait DateTimeValuedState[S <: DateTimeValuedState[S]] extends State[S] {
   def metricValue(): Timestamp
 }
 
-case class MinTimestampState(minValue: Timestamp) extends TimestampValuedState[MinTimestampState] {
+case class MinDateTimeState(minValue: Timestamp) extends DateTimeValuedState[MinDateTimeState] {
 
-  override def sum(other: MinTimestampState): MinTimestampState = {
-    MinTimestampState(if(minValue.compareTo(other.minValue) < 0) minValue else other.minValue)
+  override def sum(other: MinDateTimeState): MinDateTimeState = {
+    MinDateTimeState(if(minValue.compareTo(other.minValue) < 0) minValue else other.minValue)
   }
 
   override def metricValue(): Timestamp = {
@@ -39,16 +39,16 @@ case class MinTimestampState(minValue: Timestamp) extends TimestampValuedState[M
 }
 
 case class MinimumDateTime(column: String, where: Option[String] = None)
-  extends TimestampScanShareableAnalyzer[MinTimestampState]("Minimum Date Time", column)
+  extends TimestampScanShareableAnalyzer[MinDateTimeState]("Minimum Date Time", column)
   with FilterableAnalyzer {
 
   override def aggregationFunctions(): Seq[Column] = {
     min(conditionalSelection(column, where)).cast(TimestampType) :: Nil
   }
 
-  override def fromAggregationResult(result: Row, offset: Int): Option[MinTimestampState] = {
+  override def fromAggregationResult(result: Row, offset: Int): Option[MinDateTimeState] = {
     ifNoNullsIn(result, offset) { _ =>
-        MinTimestampState(result.getTimestamp(offset))
+        MinDateTimeState(result.getTimestamp(offset))
     }
   }
 
diff --git a/src/main/scala/com/amazon/deequ/analyzers/Sum.scala b/src/main/scala/com/amazon/deequ/analyzers/Sum.scala
index 535c14209..5b9408cb7 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/Sum.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/Sum.scala
@@ -16,7 +16,9 @@
 
 package com.amazon.deequ.analyzers
 
-import com.amazon.deequ.analyzers.Preconditions.{hasColumn, isNumeric}
+import java.math.BigDecimal
+
+import com.amazon.deequ.analyzers.Preconditions.{hasColumn, isDecimalType, isNumeric}
 import org.apache.spark.sql.functions.sum
 import org.apache.spark.sql.types.{DoubleType, StructType}
 import org.apache.spark.sql.{Column, Row}
@@ -53,3 +55,35 @@ case class Sum(column: String, where: Option[String] = None)
 
   override def filterCondition: Option[String] = where
 }
+
+case class BigDecimalSumState(sum: BigDecimal) extends BigDecimalValuedState[BigDecimalSumState] {
+
+  override def sum(other: BigDecimalSumState): BigDecimalSumState = {
+    BigDecimalSumState(sum.add(other.sum))
+  }
+
+  override def metricValue(): BigDecimal = {
+    sum
+  }
+}
+
+case class BigDecimalSum(column: String, where: Option[String] = None)
+  extends BigDecimalScanShareableAnalyzer[BigDecimalSumState]("BigDecimal Sum", column)
+    with FilterableAnalyzer {
+
+  override def aggregationFunctions(): Seq[Column] = {
+    sum(conditionalSelection(column, where)) :: Nil
+  }
+
+  override def fromAggregationResult(result: Row, offset: Int): Option[BigDecimalSumState] = {
+    ifNoNullsIn(result, offset) { _ =>
+      BigDecimalSumState(result.getDecimal(offset))
+    }
+  }
+
+  override protected def additionalPreconditions(): Seq[StructType => Unit] = {
+    hasColumn(column) :: isDecimalType(column) :: Nil
+  }
+
+  override def filterCondition: Option[String] = where
+}
\ No newline at end of file
diff --git a/src/main/scala/com/amazon/deequ/constraints/Constraint.scala b/src/main/scala/com/amazon/deequ/constraints/Constraint.scala
index 495226cd1..afe09da4d 100644
--- a/src/main/scala/com/amazon/deequ/constraints/Constraint.scala
+++ b/src/main/scala/com/amazon/deequ/constraints/Constraint.scala
@@ -495,7 +495,7 @@ object Constraint {
 
     val minimum = MinimumDateTime(column, where)
 
-    val constraint = AnalysisBasedConstraint[MinTimestampState, Timestamp, Timestamp](minimum, assertion,
+    val constraint = AnalysisBasedConstraint[MinDateTimeState, Timestamp, Timestamp](minimum, assertion,
       hint = hint)
 
     new NamedConstraint(constraint, s"MinimumTimestampConstraint($minimum)")
@@ -510,7 +510,7 @@ object Constraint {
 
     val maximum = MaximumDateTime(column, where)
 
-    val constraint = AnalysisBasedConstraint[MaxTimestampState, Timestamp, Timestamp](maximum, assertion,
+    val constraint = AnalysisBasedConstraint[MaxDateTimeState, Timestamp, Timestamp](maximum, assertion,
       hint = hint)
 
     new NamedConstraint(constraint, s"MaximumTimestampConstraint($maximum)")
diff --git a/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
index abec67ae3..0db0bbdbb 100644
--- a/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
+++ b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
@@ -5,7 +5,7 @@ import java.sql.Timestamp
 
 import com.amazon.deequ.VerificationSuite
 import com.amazon.deequ.analyzers.runners.{AnalysisRunner, AnalyzerContext}
-import com.amazon.deequ.analyzers.{MaximumDateTime, MinimumDateTime}
+import com.amazon.deequ.analyzers.{MaximumDateTime, MinimumDateTime, Sum, BigDecimalSum, Mean, BigDecimalMean}
 import com.amazon.deequ.examples.ExampleUtils.{ordersAsDataframe, withSpark}
 import com.amazon.deequ.analyzers.runners.AnalyzerContext.successMetricsAsDataFrame
 import com.amazon.deequ.checks.{Check, CheckLevel, CheckStatus}
@@ -26,6 +26,10 @@ private[examples] object AnalyzerExample extends App {
       .onData(data)
       .addAnalyzer(MinimumDateTime("orderDate"))
       .addAnalyzer(MaximumDateTime("orderDate"))
+      .addAnalyzer(Sum("amount"))
+      .addAnalyzer(BigDecimalSum("amount"))
+      .addAnalyzer(Mean("amount"))
+      .addAnalyzer(BigDecimalMean("amount"))
       .run()
     }
 
diff --git a/src/main/scala/com/amazon/deequ/metrics/Metric.scala b/src/main/scala/com/amazon/deequ/metrics/Metric.scala
index 97e9665d1..ce6703093 100644
--- a/src/main/scala/com/amazon/deequ/metrics/Metric.scala
+++ b/src/main/scala/com/amazon/deequ/metrics/Metric.scala
@@ -69,7 +69,7 @@ case class KeyedDoubleMetric(
   }
 }
 
-case class TimestampMetric(
+case class DateTimeMetric(
     entity: Entity.Value,
     name: String,
     instance: String,
@@ -83,4 +83,20 @@ case class TimestampMetric(
       Seq(DoubleMetric(entity, "Timestamp milliseconds", instance, Failure(value.failed.get)))
     }
   }
+}
+
+case class BigDecimalMetric(
+    entity: Entity.Value,
+    name: String,
+    instance: String,
+    value: Try[BigDecimal])
+  extends Metric[BigDecimal] {
+
+  override def flatten(): Seq[DoubleMetric] =  {
+    if(value.isSuccess){
+      Seq(DoubleMetric(entity, "BigDecimal", instance, Success(value.get.toDouble)))
+    } else {
+      Seq(DoubleMetric(entity, "BigDecimal", instance, Failure(value.failed.get)))
+    }
+  }
 }
\ No newline at end of file

From c0e07074ca8dd89980c006b9825cf4515634c621 Mon Sep 17 00:00:00 2001
From: Yash0215 <yash15.chawada@gmail.com>
Date: Fri, 25 Sep 2020 19:44:47 +0530
Subject: [PATCH 05/14] Feature Added: > 'isContained' is made available for
 all Numeric Types and Char

---
 .../scala/com/amazon/deequ/checks/Check.scala | 48 ++++++++++++++++---
 1 file changed, 42 insertions(+), 6 deletions(-)

diff --git a/src/main/scala/com/amazon/deequ/checks/Check.scala b/src/main/scala/com/amazon/deequ/checks/Check.scala
index 5fb194449..8d048421b 100644
--- a/src/main/scala/com/amazon/deequ/checks/Check.scala
+++ b/src/main/scala/com/amazon/deequ/checks/Check.scala
@@ -926,6 +926,22 @@ case class Check(
     isContainedIn(column, allowedValues, Check.IsOne, None)
   }
 
+  /**
+   * Asserts that every non-null value in a column is contained in a set of predefined values
+   *for providing isContained availble for numeric types as well
+   *
+   * @param column Column to run the assertion on
+   * @param allowedValues allowed values for the column
+   * @return
+   */
+  def isContainedIn[T <: AnyVal](
+      column: String,
+      allowedValues: Array[T])
+  : CheckWithLastConstraintFilterable = {
+
+    isContainedIn(column, allowedValues, Check.IsOne, None)
+  }
+
   // We can't use default values here as you can't combine default values and overloading in Scala
   /**
     * Asserts that every non-null value in a column is contained in a set of predefined values
@@ -944,6 +960,15 @@ case class Check(
     isContainedIn(column, allowedValues, Check.IsOne, hint)
   }
 
+  def isContainedIn[T <: AnyVal](
+      column: String,
+      allowedValues: Array[T],
+      hint: Option[String])
+  : CheckWithLastConstraintFilterable = {
+
+    isContainedIn(column, allowedValues, Check.IsOne, hint)
+  }
+
   // We can't use default values here as you can't combine default values and overloading in Scala
   /**
     * Asserts that every non-null value in a column is contained in a set of predefined values
@@ -973,22 +998,32 @@ case class Check(
     * @param hint A hint to provide additional context why a constraint could have failed
     * @return
     */
-  def isContainedIn(
+  def isContainedIn[T](
       column: String,
-      allowedValues: Array[String],
+      allowedValues: Array[T],
       assertion: Double => Boolean,
       hint: Option[String])
     : CheckWithLastConstraintFilterable = {
 
-
-    val valueList = allowedValues
-      .map { _.replaceAll("'", "''") }
-      .mkString("'", "','", "'")
+    val valueList = getValueList(allowedValues)
 
     val predicate = s"`$column` IS NULL OR `$column` IN ($valueList)"
     satisfies(predicate, s"$column contained in ${allowedValues.mkString(",")}", assertion, hint)
   }
 
+
+  def getValueList[T](allowedValues: Array[_]): String = {
+    allowedValues match {
+      case allowedValues : Array[String] => allowedValues
+        .map {
+          _.toString.replaceAll("'", "''")
+        }
+        .mkString("'", "','", "'")
+      case allowedValues : Array[Char] => allowedValues.mkString("'", "','", "'")
+      case _ => allowedValues.mkString(",")
+    }
+  }
+
   /**
     * Asserts that the non-null values in a numeric column fall into the predefined interval
     *
@@ -1247,4 +1282,5 @@ object Check {
 
     detectedAnomalies.anomalies.isEmpty
   }
+
 }

From 692ce76a021eb650d3b8870dca3b47917e4305fe Mon Sep 17 00:00:00 2001
From: Yash0215 <yash15.chawada@gmail.com>
Date: Sat, 26 Sep 2020 14:53:20 +0530
Subject: [PATCH 06/14] Feature Added: > Analyzer - DateTimeDistribution   for
 analyzing distributions metrics (count, ratio) over some time intervals

---
 .../analyzers/DateTimeDistribution.scala      | 121 ++++++++++++++++++
 .../catalyst/DateTimeAggregation.scala        |  57 +++++++++
 .../analyzers/catalyst/DeequFunctions.scala   |  13 +-
 .../deequ/examples/AnalyzerExample.scala      |   7 +-
 4 files changed, 193 insertions(+), 5 deletions(-)
 create mode 100644 src/main/scala/com/amazon/deequ/analyzers/DateTimeDistribution.scala
 create mode 100644 src/main/scala/com/amazon/deequ/analyzers/catalyst/DateTimeAggregation.scala

diff --git a/src/main/scala/com/amazon/deequ/analyzers/DateTimeDistribution.scala b/src/main/scala/com/amazon/deequ/analyzers/DateTimeDistribution.scala
new file mode 100644
index 000000000..1b8cbae7f
--- /dev/null
+++ b/src/main/scala/com/amazon/deequ/analyzers/DateTimeDistribution.scala
@@ -0,0 +1,121 @@
+/**
+ * Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ * use this file except in compliance with the License. A copy of the License
+ * is located at
+ *
+ *     http://aws.amazon.com/apache2.0/
+ *
+ * or in the "license" file accompanying this file. This file is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ *
+ */
+
+package com.amazon.deequ.analyzers
+
+import java.sql.Timestamp
+import com.amazon.deequ.analyzers.Analyzers._
+import com.amazon.deequ.analyzers.Preconditions.{hasColumn, isDateType}
+import com.amazon.deequ.analyzers.runners.MetricCalculationException
+import com.amazon.deequ.metrics.{Distribution, DistributionValue, HistogramMetric}
+import org.apache.spark.sql.DeequFunctions.dateTimeDistribution
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Column, Row}
+
+import scala.util.{Failure, Success}
+
+object DistributionInterval extends Enumeration {
+  val QUARTER_HOUR, HOURLY, DAILY, WEEKLY, MONTHLY = Value
+}
+
+case class DateTimeDistributionState(distribution : Map[(Timestamp, Timestamp), Long])
+  extends State[DateTimeDistributionState] {
+
+  override def sum(other: DateTimeDistributionState): DateTimeDistributionState = {
+
+    DateTimeDistributionState(distribution ++ other.distribution.map { case (k, v) => k -> (v + distribution.getOrElse(k, 0l)) })
+  }
+}
+
+object  DateTimeDistributionState {
+
+  def computeStateFromResult(result: Map[Long, Long], frequency: Long): Map[(Timestamp, Timestamp), Long] = {
+    result.map({
+      case (x, y) => (new Timestamp(x), new Timestamp(x + frequency - 1l)) -> y
+    })
+  }
+
+  def toDistribution(histogram: DateTimeDistributionState): Distribution = {
+    val totalCount = histogram.distribution.foldLeft(0l)(_ + _._2)
+    Distribution(histogram.distribution.map({
+      case (x, y) => ("(" + x._1.toString + " to " + x._2.toString + ")") -> DistributionValue(y, y.toDouble/totalCount)
+    }), totalCount)
+  }
+}
+
+/**
+ *
+ * @param column: column on which distribution analysis is to be performed
+ * @param interval: interval of the distribution;
+ * @param where: optional filter condition
+ */
+case class DateTimeDistribution(
+    column: String,
+    interval: DistributionInterval.Value,
+    where: Option[String] = None)
+  extends ScanShareableAnalyzer[DateTimeDistributionState, HistogramMetric]
+    with FilterableAnalyzer {
+
+  /** Defines the aggregations to compute on the data */
+  override private[deequ] def aggregationFunctions() : Seq[Column]  = {
+    dateTimeDistribution(conditionalSelection(column, where), DateTimeDistribution.getDateTimeAggIntervalValue(interval)) :: Nil
+  }
+
+  /** Computes the state from the result of the aggregation functions */
+  override private[deequ] def fromAggregationResult(result: Row, offset: Int): Option[DateTimeDistributionState] = {
+    ifNoNullsIn(result, offset) { _ =>
+      DateTimeDistributionState(DateTimeDistributionState.computeStateFromResult(Map.empty[Long, Long] ++ result.getMap(0),
+        DateTimeDistribution.getDateTimeAggIntervalValue(interval)))
+    }
+  }
+
+  override def preconditions: Seq[StructType => Unit] = {
+    hasColumn(column) +: isDateType(column) +: super.preconditions
+  }
+
+  override def filterCondition: Option[String] = where
+
+  /**
+   * Compute the metric from the state (sufficient statistics)
+   * @param state wrapper holding a state of type S (required due to typing issues...)
+   * @return
+   */
+  override def computeMetricFrom(state:  Option[DateTimeDistributionState]): HistogramMetric = {
+    state match {
+      case Some(histogram) =>
+        HistogramMetric(column, Success(DateTimeDistributionState.toDistribution(histogram)))
+      case _ =>
+        toFailureMetric(emptyStateException(this))
+    }
+  }
+
+  override private[deequ] def toFailureMetric(failure:  Exception): HistogramMetric = {
+    HistogramMetric(column, Failure(MetricCalculationException.wrapIfNecessary(failure)))
+  }
+}
+
+object  DateTimeDistribution {
+
+  def getDateTimeAggIntervalValue(interval: DistributionInterval.Value) : Long = {
+    interval match {
+      case DistributionInterval.QUARTER_HOUR => 900000l    // 15 Minutes
+      case DistributionInterval.HOURLY => 3600000l         // 60 Minutes
+      case DistributionInterval.DAILY => 86400000          // 24 Hours
+      case _ => 604800000l                                  // 7 * 24 Hours
+    }
+  }
+
+}
diff --git a/src/main/scala/com/amazon/deequ/analyzers/catalyst/DateTimeAggregation.scala b/src/main/scala/com/amazon/deequ/analyzers/catalyst/DateTimeAggregation.scala
new file mode 100644
index 000000000..ff1d3355d
--- /dev/null
+++ b/src/main/scala/com/amazon/deequ/analyzers/catalyst/DateTimeAggregation.scala
@@ -0,0 +1,57 @@
+/**
+ * Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ * use this file except in compliance with the License. A copy of the License
+ * is located at
+ *
+ *     http://aws.amazon.com/apache2.0/
+ *
+ * or in the "license" file accompanying this file. This file is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ *
+ */
+
+package com.amazon.deequ.analyzers.catalyst
+
+import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
+import org.apache.spark.sql.{Row}
+import org.apache.spark.sql.types._
+
+
+ class DateTimeAggregation(frequency: Long) extends UserDefinedAggregateFunction {
+
+  override def inputSchema: StructType = StructType(StructField("value", TimestampType) :: Nil)
+
+  override def bufferSchema: StructType = StructType(StructField("map", DataTypes.createMapType(LongType, LongType)) :: Nil)
+
+  override def dataType: DataType = DataTypes.createMapType(LongType, LongType)
+
+  override def deterministic: Boolean = true
+
+  override def initialize(buffer: MutableAggregationBuffer): Unit = {
+    val map = Map.empty[Long, Long]
+    buffer.update(0, map)
+  }
+
+  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
+    if (!input.isNullAt(0)) {
+      val datetime = input.getTimestamp(0).getTime
+      val batchTime = datetime - (datetime % frequency)
+      var bufferMap = buffer(0).asInstanceOf[Map[Long, Long]]
+      buffer(0) = bufferMap + (batchTime -> (bufferMap.getOrElse(batchTime, 0l)+ 1l))
+    }
+  }
+
+  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
+    var bufferMap1 = buffer1(0).asInstanceOf[Map[Long, Long]]
+    var bufferMap2 = buffer2(0).asInstanceOf[Map[Long, Long]]
+    buffer1(0) = bufferMap1 ++ bufferMap2.map { case (k, v) => k -> (v + bufferMap1.getOrElse(k, 0l)) }
+  }
+
+  override def evaluate(buffer: Row): Any = {
+    buffer.getMap(0)
+  }
+}
diff --git a/src/main/scala/com/amazon/deequ/analyzers/catalyst/DeequFunctions.scala b/src/main/scala/com/amazon/deequ/analyzers/catalyst/DeequFunctions.scala
index e19448d76..8efbf1ccf 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/catalyst/DeequFunctions.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/catalyst/DeequFunctions.scala
@@ -17,7 +17,7 @@
 package org.apache.spark.sql
 
 
-import com.amazon.deequ.analyzers.KLLSketch
+import com.amazon.deequ.analyzers.catalyst.DateTimeAggregation
 import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateFunction, StatefulApproxQuantile, StatefulHyperloglogPlus}
 import org.apache.spark.sql.catalyst.expressions.Literal
 
@@ -81,6 +81,17 @@ object DeequFunctions {
     statefulDataType(column)
   }
 
+  /**
+   * return DataTime distribution aggregation function
+   * @param column: column on which aggragtion to be performed
+   * @param interval: interval of date time aggregation
+   * @return Column: aggregation function Column
+   * */
+  def dateTimeDistribution(column: Column, interval: Long): Column = {
+    val dateTimeAgg = new DateTimeAggregation(interval)
+    dateTimeAgg(column)
+  }
+
   def stateful_kll(
       column: Column,
       sketchSize: Int,
diff --git a/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
index 0db0bbdbb..450d5ca54 100644
--- a/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
+++ b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
@@ -3,13 +3,11 @@ package com.amazon.deequ.examples
 import java.math.BigDecimal
 import java.sql.Timestamp
 
-import com.amazon.deequ.VerificationSuite
+
 import com.amazon.deequ.analyzers.runners.{AnalysisRunner, AnalyzerContext}
-import com.amazon.deequ.analyzers.{MaximumDateTime, MinimumDateTime, Sum, BigDecimalSum, Mean, BigDecimalMean}
+import com.amazon.deequ.analyzers.{BigDecimalMean, BigDecimalSum, DateTimeDistribution, DistributionInterval, MaximumDateTime, Mean, MinimumDateTime, Sum}
 import com.amazon.deequ.examples.ExampleUtils.{ordersAsDataframe, withSpark}
 import com.amazon.deequ.analyzers.runners.AnalyzerContext.successMetricsAsDataFrame
-import com.amazon.deequ.checks.{Check, CheckLevel, CheckStatus}
-import com.amazon.deequ.constraints.ConstraintStatus
 
 private[examples] object AnalyzerExample extends App {
   withSpark { session =>
@@ -24,6 +22,7 @@ private[examples] object AnalyzerExample extends App {
 
     val analysisResult: AnalyzerContext = { AnalysisRunner
       .onData(data)
+      .addAnalyzer(DateTimeDistribution("orderDate", DistributionInterval.HOURLY))
       .addAnalyzer(MinimumDateTime("orderDate"))
       .addAnalyzer(MaximumDateTime("orderDate"))
       .addAnalyzer(Sum("amount"))

From e65a03d296202380db1879a5fb1017201400f950 Mon Sep 17 00:00:00 2001
From: Yash0215 <yash15.chawada@gmail.com>
Date: Sat, 26 Sep 2020 15:06:49 +0530
Subject: [PATCH 07/14] Fixes

---
 .../deequ/analyzers/DateTimeDistribution.scala     | 14 +++++++-------
 .../analyzers/catalyst/DateTimeAggregation.scala   | 14 +++++++-------
 2 files changed, 14 insertions(+), 14 deletions(-)

diff --git a/src/main/scala/com/amazon/deequ/analyzers/DateTimeDistribution.scala b/src/main/scala/com/amazon/deequ/analyzers/DateTimeDistribution.scala
index 1b8cbae7f..ac6691d1b 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/DateTimeDistribution.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/DateTimeDistribution.scala
@@ -36,7 +36,7 @@ case class DateTimeDistributionState(distribution : Map[(Timestamp, Timestamp),
 
   override def sum(other: DateTimeDistributionState): DateTimeDistributionState = {
 
-    DateTimeDistributionState(distribution ++ other.distribution.map { case (k, v) => k -> (v + distribution.getOrElse(k, 0l)) })
+    DateTimeDistributionState(distribution ++ other.distribution.map { case (k, v) => k -> (v + distribution.getOrElse(k, 0L)) })
   }
 }
 
@@ -44,12 +44,12 @@ object  DateTimeDistributionState {
 
   def computeStateFromResult(result: Map[Long, Long], frequency: Long): Map[(Timestamp, Timestamp), Long] = {
     result.map({
-      case (x, y) => (new Timestamp(x), new Timestamp(x + frequency - 1l)) -> y
+      case (x, y) => (new Timestamp(x), new Timestamp(x + frequency - 1L)) -> y
     })
   }
 
   def toDistribution(histogram: DateTimeDistributionState): Distribution = {
-    val totalCount = histogram.distribution.foldLeft(0l)(_ + _._2)
+    val totalCount = histogram.distribution.foldLeft(0L)(_ + _._2)
     Distribution(histogram.distribution.map({
       case (x, y) => ("(" + x._1.toString + " to " + x._2.toString + ")") -> DistributionValue(y, y.toDouble/totalCount)
     }), totalCount)
@@ -111,10 +111,10 @@ object  DateTimeDistribution {
 
   def getDateTimeAggIntervalValue(interval: DistributionInterval.Value) : Long = {
     interval match {
-      case DistributionInterval.QUARTER_HOUR => 900000l    // 15 Minutes
-      case DistributionInterval.HOURLY => 3600000l         // 60 Minutes
-      case DistributionInterval.DAILY => 86400000          // 24 Hours
-      case _ => 604800000l                                  // 7 * 24 Hours
+      case DistributionInterval.QUARTER_HOUR => 900000L   // 15 Minutes
+      case DistributionInterval.HOURLY => 3600000L        // 60 Minutes
+      case DistributionInterval.DAILY => 86400000L         // 24 Hours
+      case _ => 604800000L                                  // 7 * 24 Hours
     }
   }
 
diff --git a/src/main/scala/com/amazon/deequ/analyzers/catalyst/DateTimeAggregation.scala b/src/main/scala/com/amazon/deequ/analyzers/catalyst/DateTimeAggregation.scala
index ff1d3355d..3ed62716e 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/catalyst/DateTimeAggregation.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/catalyst/DateTimeAggregation.scala
@@ -17,11 +17,11 @@
 package com.amazon.deequ.analyzers.catalyst
 
 import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
-import org.apache.spark.sql.{Row}
+import org.apache.spark.sql.Row
 import org.apache.spark.sql.types._
 
 
- class DateTimeAggregation(frequency: Long) extends UserDefinedAggregateFunction {
+class DateTimeAggregation(frequency: Long) extends UserDefinedAggregateFunction {
 
   override def inputSchema: StructType = StructType(StructField("value", TimestampType) :: Nil)
 
@@ -40,15 +40,15 @@ import org.apache.spark.sql.types._
     if (!input.isNullAt(0)) {
       val datetime = input.getTimestamp(0).getTime
       val batchTime = datetime - (datetime % frequency)
-      var bufferMap = buffer(0).asInstanceOf[Map[Long, Long]]
-      buffer(0) = bufferMap + (batchTime -> (bufferMap.getOrElse(batchTime, 0l)+ 1l))
+      val bufferMap = buffer(0).asInstanceOf[Map[Long, Long]]
+      buffer(0) = bufferMap + (batchTime -> (bufferMap.getOrElse(batchTime, 0L)+ 1L))
     }
   }
 
   override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
-    var bufferMap1 = buffer1(0).asInstanceOf[Map[Long, Long]]
-    var bufferMap2 = buffer2(0).asInstanceOf[Map[Long, Long]]
-    buffer1(0) = bufferMap1 ++ bufferMap2.map { case (k, v) => k -> (v + bufferMap1.getOrElse(k, 0l)) }
+    val bufferMap1 = buffer1(0).asInstanceOf[Map[Long, Long]]
+    val bufferMap2 = buffer2(0).asInstanceOf[Map[Long, Long]]
+    buffer1(0) = bufferMap1 ++ bufferMap2.map { case (k, v) => k -> (v + bufferMap1.getOrElse(k, 0L)) }
   }
 
   override def evaluate(buffer: Row): Any = {

From 7ac28fb3ffba18e019adb15b86bc251d3534499c Mon Sep 17 00:00:00 2001
From: Yash0215 <yash15.chawada@gmail.com>
Date: Tue, 29 Sep 2020 19:32:14 +0530
Subject: [PATCH 08/14] Commit

---
 src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala | 1 +
 1 file changed, 1 insertion(+)

diff --git a/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
index 450d5ca54..3ae06de21 100644
--- a/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
+++ b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
@@ -29,6 +29,7 @@ private[examples] object AnalyzerExample extends App {
       .addAnalyzer(BigDecimalSum("amount"))
       .addAnalyzer(Mean("amount"))
       .addAnalyzer(BigDecimalMean("amount"))
+      .addAnalyzer(BigDecimalMean("amount"))
       .run()
     }
 

From 6799a211189d9e6d0868a6825f08ac1a0a951fb5 Mon Sep 17 00:00:00 2001
From: Yash0215 <yash15.chawada@gmail.com>
Date: Tue, 29 Sep 2020 19:34:06 +0530
Subject: [PATCH 09/14] Commit

---
 src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala | 1 -
 1 file changed, 1 deletion(-)

diff --git a/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
index 3ae06de21..450d5ca54 100644
--- a/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
+++ b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
@@ -29,7 +29,6 @@ private[examples] object AnalyzerExample extends App {
       .addAnalyzer(BigDecimalSum("amount"))
       .addAnalyzer(Mean("amount"))
       .addAnalyzer(BigDecimalMean("amount"))
-      .addAnalyzer(BigDecimalMean("amount"))
       .run()
     }
 

From 541585692de86786dcae377fc48099c28f985512 Mon Sep 17 00:00:00 2001
From: Yash0215 <yash15.chawada@gmail.com>
Date: Thu, 22 Oct 2020 19:48:41 +0530
Subject: [PATCH 10/14] Adding Licence

---
 .../amazon/deequ/examples/AnalyzerExample.scala  | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)

diff --git a/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
index 450d5ca54..e3813d5c1 100644
--- a/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
+++ b/src/main/scala/com/amazon/deequ/examples/AnalyzerExample.scala
@@ -1,3 +1,19 @@
+/**
+ * Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ * use this file except in compliance with the License. A copy of the License
+ * is located at
+ *
+ *     http://aws.amazon.com/apache2.0/
+ *
+ * or in the "license" file accompanying this file. This file is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ *
+ */
+
 package com.amazon.deequ.examples
 
 import java.math.BigDecimal

From 66aebaf6dc1a5f71cd707d9d7e0bdcc6ee2c81ce Mon Sep 17 00:00:00 2001
From: Yash0215 <yash15.chawada@gmail.com>
Date: Fri, 23 Oct 2020 11:28:13 +0530
Subject: [PATCH 11/14] Formatting

---
 .../analyzers/DateTimeDistribution.scala      | 56 +++++++++++--------
 .../scala/com/amazon/deequ/checks/Check.scala | 17 +++---
 .../amazon/deequ/constraints/Constraint.scala |  8 +--
 .../com/amazon/deequ/metrics/Metric.scala     | 13 +++--
 4 files changed, 53 insertions(+), 41 deletions(-)

diff --git a/src/main/scala/com/amazon/deequ/analyzers/DateTimeDistribution.scala b/src/main/scala/com/amazon/deequ/analyzers/DateTimeDistribution.scala
index ac6691d1b..361f32d09 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/DateTimeDistribution.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/DateTimeDistribution.scala
@@ -5,7 +5,7 @@
  * use this file except in compliance with the License. A copy of the License
  * is located at
  *
- *     http://aws.amazon.com/apache2.0/
+ * http://aws.amazon.com/apache2.0/
  *
  * or in the "license" file accompanying this file. This file is distributed on
  * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
@@ -31,18 +31,21 @@ object DistributionInterval extends Enumeration {
   val QUARTER_HOUR, HOURLY, DAILY, WEEKLY, MONTHLY = Value
 }
 
-case class DateTimeDistributionState(distribution : Map[(Timestamp, Timestamp), Long])
+case class DateTimeDistributionState(distribution: Map[(Timestamp, Timestamp), Long])
   extends State[DateTimeDistributionState] {
 
   override def sum(other: DateTimeDistributionState): DateTimeDistributionState = {
 
-    DateTimeDistributionState(distribution ++ other.distribution.map { case (k, v) => k -> (v + distribution.getOrElse(k, 0L)) })
+    DateTimeDistributionState(distribution ++ other.distribution.map {
+      case (k, v) => k -> (v + distribution.getOrElse(k, 0L))
+    })
   }
 }
 
-object  DateTimeDistributionState {
+object DateTimeDistributionState {
 
-  def computeStateFromResult(result: Map[Long, Long], frequency: Long): Map[(Timestamp, Timestamp), Long] = {
+  def computeStateFromResult(result: Map[Long, Long], frequency: Long):
+  Map[(Timestamp, Timestamp), Long] = {
     result.map({
       case (x, y) => (new Timestamp(x), new Timestamp(x + frequency - 1L)) -> y
     })
@@ -51,33 +54,37 @@ object  DateTimeDistributionState {
   def toDistribution(histogram: DateTimeDistributionState): Distribution = {
     val totalCount = histogram.distribution.foldLeft(0L)(_ + _._2)
     Distribution(histogram.distribution.map({
-      case (x, y) => ("(" + x._1.toString + " to " + x._2.toString + ")") -> DistributionValue(y, y.toDouble/totalCount)
+      case (x, y) => ("(" + x._1.toString + " to " + x._2.toString + ")") ->
+        DistributionValue(y, y.toDouble / totalCount)
     }), totalCount)
   }
 }
 
 /**
  *
- * @param column: column on which distribution analysis is to be performed
- * @param interval: interval of the distribution;
- * @param where: optional filter condition
+ * @param column   : column on which distribution analysis is to be performed
+ * @param interval : interval of the distribution;
+ * @param where    : optional filter condition
  */
 case class DateTimeDistribution(
-    column: String,
-    interval: DistributionInterval.Value,
-    where: Option[String] = None)
+                                 column: String,
+                                 interval: DistributionInterval.Value,
+                                 where: Option[String] = None)
   extends ScanShareableAnalyzer[DateTimeDistributionState, HistogramMetric]
     with FilterableAnalyzer {
 
   /** Defines the aggregations to compute on the data */
-  override private[deequ] def aggregationFunctions() : Seq[Column]  = {
-    dateTimeDistribution(conditionalSelection(column, where), DateTimeDistribution.getDateTimeAggIntervalValue(interval)) :: Nil
+  override private[deequ] def aggregationFunctions(): Seq[Column] = {
+    dateTimeDistribution(conditionalSelection(column, where),
+      DateTimeDistribution.getDateTimeAggIntervalValue(interval)) :: Nil
   }
 
   /** Computes the state from the result of the aggregation functions */
-  override private[deequ] def fromAggregationResult(result: Row, offset: Int): Option[DateTimeDistributionState] = {
+  override private[deequ] def fromAggregationResult(result: Row, offset: Int):
+  Option[DateTimeDistributionState] = {
     ifNoNullsIn(result, offset) { _ =>
-      DateTimeDistributionState(DateTimeDistributionState.computeStateFromResult(Map.empty[Long, Long] ++ result.getMap(0),
+      DateTimeDistributionState(
+        DateTimeDistributionState.computeStateFromResult(Map.empty[Long, Long] ++ result.getMap(0),
         DateTimeDistribution.getDateTimeAggIntervalValue(interval)))
     }
   }
@@ -90,10 +97,11 @@ case class DateTimeDistribution(
 
   /**
    * Compute the metric from the state (sufficient statistics)
+   *
    * @param state wrapper holding a state of type S (required due to typing issues...)
    * @return
    */
-  override def computeMetricFrom(state:  Option[DateTimeDistributionState]): HistogramMetric = {
+  override def computeMetricFrom(state: Option[DateTimeDistributionState]): HistogramMetric = {
     state match {
       case Some(histogram) =>
         HistogramMetric(column, Success(DateTimeDistributionState.toDistribution(histogram)))
@@ -102,19 +110,19 @@ case class DateTimeDistribution(
     }
   }
 
-  override private[deequ] def toFailureMetric(failure:  Exception): HistogramMetric = {
+  override private[deequ] def toFailureMetric(failure: Exception): HistogramMetric = {
     HistogramMetric(column, Failure(MetricCalculationException.wrapIfNecessary(failure)))
   }
 }
 
-object  DateTimeDistribution {
+object DateTimeDistribution {
 
-  def getDateTimeAggIntervalValue(interval: DistributionInterval.Value) : Long = {
+  def getDateTimeAggIntervalValue(interval: DistributionInterval.Value): Long = {
     interval match {
-      case DistributionInterval.QUARTER_HOUR => 900000L   // 15 Minutes
-      case DistributionInterval.HOURLY => 3600000L        // 60 Minutes
-      case DistributionInterval.DAILY => 86400000L         // 24 Hours
-      case _ => 604800000L                                  // 7 * 24 Hours
+      case DistributionInterval.QUARTER_HOUR => 900000L // 15 Minutes
+      case DistributionInterval.HOURLY => 3600000L // 60 Minutes
+      case DistributionInterval.DAILY => 86400000L // 24 Hours
+      case _ => 604800000L // 7 * 24 Hours
     }
   }
 
diff --git a/src/main/scala/com/amazon/deequ/checks/Check.scala b/src/main/scala/com/amazon/deequ/checks/Check.scala
index 8d048421b..9baaf8d6c 100644
--- a/src/main/scala/com/amazon/deequ/checks/Check.scala
+++ b/src/main/scala/com/amazon/deequ/checks/Check.scala
@@ -1016,7 +1016,7 @@ case class Check(
     allowedValues match {
       case allowedValues : Array[String] => allowedValues
         .map {
-          _.toString.replaceAll("'", "''")
+          _.replaceAll("'", "''")
         }
         .mkString("'", "','", "'")
       case allowedValues : Array[Char] => allowedValues.mkString("'", "','", "'")
@@ -1054,10 +1054,10 @@ case class Check(
   }
 
   /**
-   * Creates a constraint that asserts a minimum java.sql.Timestamp
+   * Creates a constraint that asserts a minimum Timestamp
    *
    * @param column – Column to run the assertion on
-   * @param assertion – Function that receives a java.sql.Timestamp input parameter and returns a boolean
+   * @param assertion – Function that receives a Timestamp input parameter and returns a boolean
    * @param hint – A hint to provide additional context why a constraint could have failed
    * @return
    */
@@ -1074,7 +1074,7 @@ case class Check(
    * Creates a constraint that asserts a maximum java.sql.Timestamp
    *
    * @param column – Column to run the assertion on
-   * @param assertion – Function that receives a java.sql.Timestamp input parameter and returns a boolean
+   * @param assertion – Function that receives a Timestamp input parameter and returns a boolean
    * @param hint – A hint to provide additional context why a constraint could have failed
    * @return
    */
@@ -1089,7 +1089,8 @@ case class Check(
 
   /**
    *
-   * Asserts that, in each row, the value of column (DateType or TimestampType) is less than the given datetime (Timestamp)
+   * Asserts that, in each row, the value of column (DateType or TimestampType)
+   * is less than the given datetime (Timestamp)
    *
    * @param column Column to run the assertion on
    * @param datetime value of Timestamp to run assert
@@ -1111,7 +1112,8 @@ case class Check(
 
   /**
    *
-   * Asserts that, in each row, the value of column (DateType or TimestampType) is greater than the given datetime (Timestamp)
+   * Asserts that, in each row, the value of column (DateType or TimestampType)
+   * is greater than the given datetime (Timestamp)
    *
    * @param column Column to run the assertion on
    * @param datetime value of Timestamp to run assert
@@ -1153,7 +1155,8 @@ case class Check(
 
   /**
    *
-   * Asserts that, in each row, the value of column (DateType or TimestampType) contains a future date
+   * Asserts that, in each row, the value of column (DateType or TimestampType)
+   * contains a future date
    *
    * @param column Column to run the assertion on
    * @param assertion Function that receives a Timestamp input parameter and returns a boolean
diff --git a/src/main/scala/com/amazon/deequ/constraints/Constraint.scala b/src/main/scala/com/amazon/deequ/constraints/Constraint.scala
index afe09da4d..0ef06dfc1 100644
--- a/src/main/scala/com/amazon/deequ/constraints/Constraint.scala
+++ b/src/main/scala/com/amazon/deequ/constraints/Constraint.scala
@@ -495,8 +495,8 @@ object Constraint {
 
     val minimum = MinimumDateTime(column, where)
 
-    val constraint = AnalysisBasedConstraint[MinDateTimeState, Timestamp, Timestamp](minimum, assertion,
-      hint = hint)
+    val constraint = AnalysisBasedConstraint[MinDateTimeState, Timestamp,
+      Timestamp](minimum, assertion, hint = hint)
 
     new NamedConstraint(constraint, s"MinimumTimestampConstraint($minimum)")
   }
@@ -510,8 +510,8 @@ object Constraint {
 
     val maximum = MaximumDateTime(column, where)
 
-    val constraint = AnalysisBasedConstraint[MaxDateTimeState, Timestamp, Timestamp](maximum, assertion,
-      hint = hint)
+    val constraint = AnalysisBasedConstraint[MaxDateTimeState, Timestamp,
+      Timestamp](maximum, assertion, hint = hint)
 
     new NamedConstraint(constraint, s"MaximumTimestampConstraint($maximum)")
   }
diff --git a/src/main/scala/com/amazon/deequ/metrics/Metric.scala b/src/main/scala/com/amazon/deequ/metrics/Metric.scala
index ce6703093..fc5f996a6 100644
--- a/src/main/scala/com/amazon/deequ/metrics/Metric.scala
+++ b/src/main/scala/com/amazon/deequ/metrics/Metric.scala
@@ -76,9 +76,10 @@ case class DateTimeMetric(
     value: Try[Timestamp])
   extends Metric[Timestamp] {
 
-  override def flatten(): Seq[DoubleMetric] =  {
-    if(value.isSuccess){
-      Seq(DoubleMetric(entity, "Timestamp milliseconds", instance, Success(value.get.getTime.toDouble)))
+  override def flatten(): Seq[DoubleMetric] = {
+    if(value.isSuccess) {
+      Seq(DoubleMetric(entity, "Timestamp milliseconds", instance,
+        Success(value.get.getTime.toDouble)))
     } else {
       Seq(DoubleMetric(entity, "Timestamp milliseconds", instance, Failure(value.failed.get)))
     }
@@ -92,11 +93,11 @@ case class BigDecimalMetric(
     value: Try[BigDecimal])
   extends Metric[BigDecimal] {
 
-  override def flatten(): Seq[DoubleMetric] =  {
-    if(value.isSuccess){
+  override def flatten(): Seq[DoubleMetric] = {
+    if(value.isSuccess) {
       Seq(DoubleMetric(entity, "BigDecimal", instance, Success(value.get.toDouble)))
     } else {
       Seq(DoubleMetric(entity, "BigDecimal", instance, Failure(value.failed.get)))
     }
   }
-}
\ No newline at end of file
+}

From 43034040c5ce87d9971f1579eff2c2fd7b786923 Mon Sep 17 00:00:00 2001
From: Yash0215 <yash15.chawada@gmail.com>
Date: Fri, 23 Oct 2020 12:27:45 +0530
Subject: [PATCH 12/14] Formatting

---
 src/main/scala/com/amazon/deequ/analyzers/Maximum.scala  | 5 +++--
 .../com/amazon/deequ/analyzers/MaximumDateTime.scala     | 4 ++--
 src/main/scala/com/amazon/deequ/analyzers/Mean.scala     | 5 +++--
 src/main/scala/com/amazon/deequ/analyzers/Minimum.scala  | 5 +++--
 .../com/amazon/deequ/analyzers/MinimumDateTime.scala     | 2 +-
 src/main/scala/com/amazon/deequ/analyzers/Sum.scala      | 2 +-
 .../deequ/analyzers/catalyst/DateTimeAggregation.scala   | 9 ++++++---
 src/main/scala/com/amazon/deequ/checks/Check.scala       | 2 +-
 src/main/scala/com/amazon/deequ/metrics/Metric.scala     | 4 ++--
 9 files changed, 22 insertions(+), 16 deletions(-)

diff --git a/src/main/scala/com/amazon/deequ/analyzers/Maximum.scala b/src/main/scala/com/amazon/deequ/analyzers/Maximum.scala
index cdf086031..30b45d825 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/Maximum.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/Maximum.scala
@@ -57,7 +57,8 @@ case class Maximum(column: String, where: Option[String] = None)
   override def filterCondition: Option[String] = where
 }
 
-case class MaxBigDecimalState(minValue: BigDecimal) extends BigDecimalValuedState[MaxBigDecimalState] {
+case class MaxBigDecimalState(minValue: BigDecimal)
+  extends BigDecimalValuedState[MaxBigDecimalState] {
 
   override def sum(other: MaxBigDecimalState): MaxBigDecimalState = {
     MaxBigDecimalState(minValue.max(other.minValue))
@@ -88,4 +89,4 @@ case class MaximumBigDecimal(column: String, where: Option[String] = None)
   }
 
   override def filterCondition: Option[String] = where
-}
\ No newline at end of file
+}
diff --git a/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala b/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala
index eb26bbc74..b27b3d759 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/MaximumDateTime.scala
@@ -26,7 +26,7 @@ import java.sql.Timestamp
 case class MaxDateTimeState(maxValue: Timestamp) extends DateTimeValuedState[MaxDateTimeState] {
 
   override def sum(other: MaxDateTimeState): MaxDateTimeState = {
-    MaxDateTimeState(if(maxValue.compareTo(other.maxValue) > 0) maxValue else other.maxValue)
+    MaxDateTimeState(if (maxValue.compareTo(other.maxValue) > 0) maxValue else other.maxValue)
   }
 
   override def metricValue(): Timestamp = {
@@ -53,4 +53,4 @@ case class MaximumDateTime(column: String, where: Option[String] = None)
   }
 
   override def filterCondition: Option[String] = where
-}
\ No newline at end of file
+}
diff --git a/src/main/scala/com/amazon/deequ/analyzers/Mean.scala b/src/main/scala/com/amazon/deequ/analyzers/Mean.scala
index 44792b92d..b0451cf67 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/Mean.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/Mean.scala
@@ -58,7 +58,8 @@ case class Mean(column: String, where: Option[String] = None)
   override def filterCondition: Option[String] = where
 }
 
-case class BigDecimalMeanState(sum: BigDecimal, count: Long) extends BigDecimalValuedState[BigDecimalMeanState] {
+case class BigDecimalMeanState(sum: BigDecimal, count: Long)
+  extends BigDecimalValuedState[BigDecimalMeanState] {
 
   override def sum(other: BigDecimalMeanState): BigDecimalMeanState = {
     BigDecimalMeanState(sum.add(other.sum), count + other.count)
@@ -90,4 +91,4 @@ case class BigDecimalMean(column: String, where: Option[String] = None)
   }
 
   override def filterCondition: Option[String] = where
-}
\ No newline at end of file
+}
diff --git a/src/main/scala/com/amazon/deequ/analyzers/Minimum.scala b/src/main/scala/com/amazon/deequ/analyzers/Minimum.scala
index c98c2ed1b..cd43ae3a5 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/Minimum.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/Minimum.scala
@@ -57,7 +57,8 @@ case class Minimum(column: String, where: Option[String] = None)
   override def filterCondition: Option[String] = where
 }
 
-case class MinBigDecimalState(minValue: BigDecimal) extends BigDecimalValuedState[MinBigDecimalState] {
+case class MinBigDecimalState(minValue: BigDecimal)
+  extends BigDecimalValuedState[MinBigDecimalState] {
 
   override def sum(other: MinBigDecimalState): MinBigDecimalState = {
     MinBigDecimalState(minValue.min(other.minValue))
@@ -88,4 +89,4 @@ case class MinimumBigDecimal(column: String, where: Option[String] = None)
   }
 
   override def filterCondition: Option[String] = where
-}
\ No newline at end of file
+}
diff --git a/src/main/scala/com/amazon/deequ/analyzers/MinimumDateTime.scala b/src/main/scala/com/amazon/deequ/analyzers/MinimumDateTime.scala
index 967e3f390..e1a0f2b39 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/MinimumDateTime.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/MinimumDateTime.scala
@@ -30,7 +30,7 @@ trait DateTimeValuedState[S <: DateTimeValuedState[S]] extends State[S] {
 case class MinDateTimeState(minValue: Timestamp) extends DateTimeValuedState[MinDateTimeState] {
 
   override def sum(other: MinDateTimeState): MinDateTimeState = {
-    MinDateTimeState(if(minValue.compareTo(other.minValue) < 0) minValue else other.minValue)
+    MinDateTimeState(if (minValue.compareTo(other.minValue) < 0) minValue else other.minValue)
   }
 
   override def metricValue(): Timestamp = {
diff --git a/src/main/scala/com/amazon/deequ/analyzers/Sum.scala b/src/main/scala/com/amazon/deequ/analyzers/Sum.scala
index 5b9408cb7..8127561fe 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/Sum.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/Sum.scala
@@ -86,4 +86,4 @@ case class BigDecimalSum(column: String, where: Option[String] = None)
   }
 
   override def filterCondition: Option[String] = where
-}
\ No newline at end of file
+}
diff --git a/src/main/scala/com/amazon/deequ/analyzers/catalyst/DateTimeAggregation.scala b/src/main/scala/com/amazon/deequ/analyzers/catalyst/DateTimeAggregation.scala
index 3ed62716e..2290ecf5a 100644
--- a/src/main/scala/com/amazon/deequ/analyzers/catalyst/DateTimeAggregation.scala
+++ b/src/main/scala/com/amazon/deequ/analyzers/catalyst/DateTimeAggregation.scala
@@ -25,7 +25,8 @@ class DateTimeAggregation(frequency: Long) extends UserDefinedAggregateFunction
 
   override def inputSchema: StructType = StructType(StructField("value", TimestampType) :: Nil)
 
-  override def bufferSchema: StructType = StructType(StructField("map", DataTypes.createMapType(LongType, LongType)) :: Nil)
+  override def bufferSchema: StructType = StructType(StructField("map",
+    DataTypes.createMapType(LongType, LongType)) :: Nil)
 
   override def dataType: DataType = DataTypes.createMapType(LongType, LongType)
 
@@ -41,14 +42,16 @@ class DateTimeAggregation(frequency: Long) extends UserDefinedAggregateFunction
       val datetime = input.getTimestamp(0).getTime
       val batchTime = datetime - (datetime % frequency)
       val bufferMap = buffer(0).asInstanceOf[Map[Long, Long]]
-      buffer(0) = bufferMap + (batchTime -> (bufferMap.getOrElse(batchTime, 0L)+ 1L))
+      buffer(0) = bufferMap + (batchTime -> (bufferMap.getOrElse(batchTime, 0L) + 1L))
     }
   }
 
   override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
     val bufferMap1 = buffer1(0).asInstanceOf[Map[Long, Long]]
     val bufferMap2 = buffer2(0).asInstanceOf[Map[Long, Long]]
-    buffer1(0) = bufferMap1 ++ bufferMap2.map { case (k, v) => k -> (v + bufferMap1.getOrElse(k, 0L)) }
+    buffer1(0) = bufferMap1 ++ bufferMap2.map {
+      case (k, v) => k -> (v + bufferMap1.getOrElse(k, 0L))
+    }
   }
 
   override def evaluate(buffer: Row): Any = {
diff --git a/src/main/scala/com/amazon/deequ/checks/Check.scala b/src/main/scala/com/amazon/deequ/checks/Check.scala
index 9baaf8d6c..2172496b8 100644
--- a/src/main/scala/com/amazon/deequ/checks/Check.scala
+++ b/src/main/scala/com/amazon/deequ/checks/Check.scala
@@ -1050,7 +1050,7 @@ case class Check(
     val predicate = s"`$column` IS NULL OR " +
       s"(`$column` $leftOperand $lowerBound AND `$column` $rightOperand $upperBound)"
 
-    satisfies(predicate, s"$column between $lowerBound and $upperBound", hint = hint)
+    satisfies(predicate, s"`$column` between $lowerBound and $upperBound", hint = hint)
   }
 
   /**
diff --git a/src/main/scala/com/amazon/deequ/metrics/Metric.scala b/src/main/scala/com/amazon/deequ/metrics/Metric.scala
index fc5f996a6..d5036d535 100644
--- a/src/main/scala/com/amazon/deequ/metrics/Metric.scala
+++ b/src/main/scala/com/amazon/deequ/metrics/Metric.scala
@@ -77,7 +77,7 @@ case class DateTimeMetric(
   extends Metric[Timestamp] {
 
   override def flatten(): Seq[DoubleMetric] = {
-    if(value.isSuccess) {
+    if (value.isSuccess) {
       Seq(DoubleMetric(entity, "Timestamp milliseconds", instance,
         Success(value.get.getTime.toDouble)))
     } else {
@@ -94,7 +94,7 @@ case class BigDecimalMetric(
   extends Metric[BigDecimal] {
 
   override def flatten(): Seq[DoubleMetric] = {
-    if(value.isSuccess) {
+    if (value.isSuccess) {
       Seq(DoubleMetric(entity, "BigDecimal", instance, Success(value.get.toDouble)))
     } else {
       Seq(DoubleMetric(entity, "BigDecimal", instance, Failure(value.failed.get)))

From 172be49c42df09e5c1fd1053c752519523037b84 Mon Sep 17 00:00:00 2001
From: Yash0215 <yash15.chawada@gmail.com>
Date: Fri, 23 Oct 2020 13:01:05 +0530
Subject: [PATCH 13/14] Formatting

---
 .../scala/com/amazon/deequ/checks/Check.scala | 19 +------------------
 1 file changed, 1 insertion(+), 18 deletions(-)

diff --git a/src/main/scala/com/amazon/deequ/checks/Check.scala b/src/main/scala/com/amazon/deequ/checks/Check.scala
index 2172496b8..ecf543baa 100644
--- a/src/main/scala/com/amazon/deequ/checks/Check.scala
+++ b/src/main/scala/com/amazon/deequ/checks/Check.scala
@@ -1052,15 +1052,7 @@ case class Check(
 
     satisfies(predicate, s"`$column` between $lowerBound and $upperBound", hint = hint)
   }
-
-  /**
-   * Creates a constraint that asserts a minimum Timestamp
-   *
-   * @param column – Column to run the assertion on
-   * @param assertion – Function that receives a Timestamp input parameter and returns a boolean
-   * @param hint – A hint to provide additional context why a constraint could have failed
-   * @return
-   */
+  
   def hasMinTimestamp(
       column: String,
       assertion: Timestamp => Boolean,
@@ -1070,14 +1062,6 @@ case class Check(
     addFilterableConstraint { filter => minTimestampConstraint(column, assertion, filter, hint) }
   }
 
-  /**
-   * Creates a constraint that asserts a maximum java.sql.Timestamp
-   *
-   * @param column – Column to run the assertion on
-   * @param assertion – Function that receives a Timestamp input parameter and returns a boolean
-   * @param hint – A hint to provide additional context why a constraint could have failed
-   * @return
-   */
   def hasMaxTimestamp(
       column: String,
       assertion: Timestamp => Boolean,
@@ -1088,7 +1072,6 @@ case class Check(
   }
 
   /**
-   *
    * Asserts that, in each row, the value of column (DateType or TimestampType)
    * is less than the given datetime (Timestamp)
    *

From 4c9df65e0af72f53701debdc45c7e41c7ad536ea Mon Sep 17 00:00:00 2001
From: Yash0215 <yash15.chawada@gmail.com>
Date: Fri, 23 Oct 2020 13:33:07 +0530
Subject: [PATCH 14/14] Formatting

---
 src/main/scala/com/amazon/deequ/checks/Check.scala | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/main/scala/com/amazon/deequ/checks/Check.scala b/src/main/scala/com/amazon/deequ/checks/Check.scala
index ecf543baa..c78ebff75 100644
--- a/src/main/scala/com/amazon/deequ/checks/Check.scala
+++ b/src/main/scala/com/amazon/deequ/checks/Check.scala
@@ -1052,7 +1052,7 @@ case class Check(
 
     satisfies(predicate, s"`$column` between $lowerBound and $upperBound", hint = hint)
   }
-  
+
   def hasMinTimestamp(
       column: String,
       assertion: Timestamp => Boolean,