Validator: Add support for timestamp tolerance, by default 0ms (#64)

old behaviour is preserved by default

Author: tarzanek

config.yaml.example

@@ -197,3 +197,5 @@ validation:
   failuresToFetch: 100
   # What difference should we allow between floating point numbers?
   floatingPointTolerance: 0.001
+  # What difference in ms should we allow between timestamps?
+  timestampMsTolerance: 0

src/main/scala/com/scylladb/migrator/Validator.scala

@@ -109,6 +109,7 @@ object Validator {
             l,
             r,
             config.validation.floatingPointTolerance,
+            config.validation.timestampMsTolerance,
             config.validation.ttlToleranceMillis,
             config.validation.writetimeToleranceMillis,
             config.validation.compareTimestamps

src/main/scala/com/scylladb/migrator/config/Validation.scala

@@ -7,7 +7,8 @@ case class Validation(compareTimestamps: Boolean,
                       ttlToleranceMillis: Long,
                       writetimeToleranceMillis: Long,
                       failuresToFetch: Int,
-                      floatingPointTolerance: Double)
+                      floatingPointTolerance: Double,
+                      timestampMsTolerance: Double)
 object Validation {
   implicit val encoder: Encoder[Validation] = deriveEncoder[Validation]
   implicit val decoder: Decoder[Validation] = deriveDecoder[Validation]

src/main/scala/com/scylladb/migrator/readers/Cassandra.scala

@@ -27,8 +27,8 @@ object Cassandra {
   def determineCopyType(tableDef: TableDef,
                         preserveTimesRequest: Boolean): Either[Throwable, CopyType] =
     if (tableDef.columnTypes.exists(_.isCollection) && preserveTimesRequest)
-      Left(
-        new Exception("TTL/Writetime preservation is unsupported for tables with collection types. Please check in your config the option 'preserveTimestamps' and set it to false to continue."))
+      Left(new Exception(
+        "TTL/Writetime preservation is unsupported for tables with collection types. Please check in your config the option 'preserveTimestamps' and set it to false to continue."))
     else if (preserveTimesRequest && tableDef.regularColumns.nonEmpty)
       Right(CopyType.WithTimestampPreservation)
     else if (preserveTimesRequest && tableDef.regularColumns.isEmpty) {

src/main/scala/com/scylladb/migrator/validation/RowComparisonFailure.scala

@@ -3,6 +3,8 @@ package com.scylladb.migrator.validation
 import com.datastax.spark.connector.CassandraRow
 import com.google.common.math.DoubleMath
 
+import java.time.temporal.ChronoUnit
+
 case class RowComparisonFailure(row: CassandraRow,
                                 other: Option[CassandraRow],
                                 items: List[RowComparisonFailure.Item]) {
@@ -41,6 +43,7 @@ object RowComparisonFailure {
   def compareRows(left: CassandraRow,
                   right: Option[CassandraRow],
                   floatingPointTolerance: Double,
+                  timestampMsTolerance: Double,
                   ttlToleranceMillis: Long,
                   writetimeToleranceMillis: Long,
                   compareTimestamps: Boolean): Option[RowComparisonFailure] =
@@ -64,6 +67,10 @@ object RowComparisonFailure {
             rightValue = rightMap.get(name)
 
             result = (rightValue, leftValue) match {
+              // All timestamp types need to be compared with a configured tolerance
+              case (Some(l: java.time.Instant), Some(r: java.time.Instant))
+                  if timestampMsTolerance > 0 =>
+                Math.abs(r.until(l, ChronoUnit.MILLIS)) > timestampMsTolerance
               // All floating-point-like types need to be compared with a configured tolerance
               case (Some(l: Float), Some(r: Float)) =>
                 !DoubleMath.fuzzyEquals(l, r, floatingPointTolerance)