From e3feb2a8f6df0da3debb55b6d49553696a23b3e2 Mon Sep 17 00:00:00 2001
From: Enrico Minack <github@enrico.minack.dev>
Date: Wed, 23 Aug 2023 13:05:36 +0200
Subject: [PATCH 1/2] Add key order sensitivity to map comparator

---
 DIFF.md                                       |  2 +-
 .../spark/diff/DiffComparators.scala          |  2 +-
 .../diff/comparator/MapDiffComparator.scala   | 15 ++++----
 .../spark/diff/DiffComparatorSuite.scala      | 34 ++++++++++---------
 4 files changed, 29 insertions(+), 24 deletions(-)

diff --git a/DIFF.md b/DIFF.md
index 1c03a0f6..e4ecae4c 100644
--- a/DIFF.md
+++ b/DIFF.md
@@ -306,7 +306,7 @@ The following alternative comparators are provided:
 |`DiffComparators.epsilon(epsilon)`|Two values are equal when they are at most `epsilon` apart.<br/><br/>The comparator can be configured to use `epsilon` as an absolute (`.asAbsolute()`) threshold, or as relative (`.asRelative()`) to the larger value. Further, the threshold itself can be considered equal (`.asInclusive()`) or not equal (`.asExclusive()`):<ul><li>`DiffComparators.epsilon(epsilon).asAbsolute().asInclusive()`:<br/>`x` and `y` are equal iff `abs(x - y) ≤ epsilon`</li><li>`DiffComparators.epsilon(epsilon).asAbsolute().asExclusive()`:<br/>`x` and `y` are equal iff `abs(x - y) < epsilon`</li><li>`DiffComparators.epsilon(epsilon).asRelative().asInclusive()`:<br/>`x` and `y` are equal iff `abs(x - y) ≤ epsilon * max(abs(x), abs(y))`</li><li>`DiffComparators.epsilon(epsilon).asRelative().asExclusive()`:<br/>`x` and `y` are equal iff `abs(x - y) < epsilon * max(abs(x), abs(y))`</li></ul>|
 |`DiffComparators.string()`|Two `StringType` values are compared while ignoring white space differences. For this comparison, sequences of whitespaces are collapesed into single whitespaces, leading and trailing whitespaces are removed. With `DiffComparators.string(false)`, string values are compared with the default comparator.|
 |`DiffComparators.duration(duration)`|Two `DateType` or `TimestampType` values are equal when they are at most `duration` apart. That duration is an instance of `java.time.Duration`.<br/><br/>The comparator can be configured to consider `duration` as equal (`.asInclusive()`) or not equal (`.asExclusive()`):<ul><li>`DiffComparators.duration(duration).asInclusive()`:<br/>`x` and `y` are equal iff `x - y ≤ duration`</li><li>`DiffComparators.duration(duration).asExclusive()`:<br/>`x` and `y` are equal iff `x - y < duration`</li></lu>|
-|`DiffComparators.map[K,V]()`|Two `Map[K,V]` values are equal when they match in all their keys and values.|
+|`DiffComparators.map[K,V](keyOrderSensitive)`|Two `Map[K,V]` values are equal when they match in all their keys and values. With `keyOrderSensitive=true`, the order of the keys matters, with `keyOrderSensitive=false` (default), the order of keys is ignored.|
 
 An example:
 
diff --git a/src/main/scala/uk/co/gresearch/spark/diff/DiffComparators.scala b/src/main/scala/uk/co/gresearch/spark/diff/DiffComparators.scala
index 7f7a2cac..a97c9169 100644
--- a/src/main/scala/uk/co/gresearch/spark/diff/DiffComparators.scala
+++ b/src/main/scala/uk/co/gresearch/spark/diff/DiffComparators.scala
@@ -96,5 +96,5 @@ object DiffComparators {
   /**
    * This comparator compares two `Map[K,V]` values. They are equal when they match in all their keys and values.
    */
-  def map[K: Encoder, V: Encoder](): DiffComparator = MapDiffComparator[K, V]()
+  def map[K: Encoder, V: Encoder](keyOrderSensitive: Boolean = false): DiffComparator = MapDiffComparator[K, V](keyOrderSensitive)
 }
diff --git a/src/main/scala/uk/co/gresearch/spark/diff/comparator/MapDiffComparator.scala b/src/main/scala/uk/co/gresearch/spark/diff/comparator/MapDiffComparator.scala
index b8be8532..3f52eec7 100644
--- a/src/main/scala/uk/co/gresearch/spark/diff/comparator/MapDiffComparator.scala
+++ b/src/main/scala/uk/co/gresearch/spark/diff/comparator/MapDiffComparator.scala
@@ -27,11 +27,14 @@ case class MapDiffComparator[K, V](private val comparator: EquivDiffComparator[U
   override def equiv(left: Column, right: Column): Column = comparator.equiv(left, right)
 }
 
-private case class MapDiffEquiv[K: ClassTag, V](keyType: DataType, valueType: DataType) extends math.Equiv[UnsafeMapData] {
+private case class MapDiffEquiv[K: ClassTag, V](keyType: DataType, valueType: DataType, keyOrderSensitive: Boolean) extends math.Equiv[UnsafeMapData] {
   override def equiv(left: UnsafeMapData, right: UnsafeMapData): Boolean = {
 
-    val leftKeysIndices: Map[K, Int] = left.keyArray().toArray(keyType).zipWithIndex.toMap
-    val rightKeysIndices: Map[K, Int] = right.keyArray().toArray(keyType).zipWithIndex.toMap
+    val leftKeys: Array[K] = left.keyArray().toArray(keyType)
+    val rightKeys: Array[K] = right.keyArray().toArray(keyType)
+
+    val leftKeysIndices: Map[K, Int] = leftKeys.zipWithIndex.toMap
+    val rightKeysIndices: Map[K, Int] = rightKeys.zipWithIndex.toMap
 
     val leftValues = left.valueArray()
     val rightValues = right.valueArray()
@@ -46,16 +49,16 @@ private case class MapDiffEquiv[K: ClassTag, V](keyType: DataType, valueType: Da
       }
 
     left.numElements() == right.numElements() &&
-      leftKeysIndices.keySet.diff(rightKeysIndices.keySet).isEmpty &&
+      (keyOrderSensitive && leftKeys.sameElements(rightKeys) || !keyOrderSensitive && leftKeys.toSet.diff(rightKeys.toSet).isEmpty) &&
       valuesAreEqual.forall(identity)
   }
 }
 
 case object MapDiffComparator {
-  def apply[K: Encoder, V: Encoder](): MapDiffComparator[K, V] = {
+  def apply[K: Encoder, V: Encoder](keyOrderSensitive: Boolean = false): MapDiffComparator[K, V] = {
     val keyType = encoderFor[K].schema.fields(0).dataType
     val valueType = encoderFor[V].schema.fields(0).dataType
-    val equiv = MapDiffEquiv(keyType, valueType)
+    val equiv = MapDiffEquiv(keyType, valueType, keyOrderSensitive)
     val dataType = MapType(keyType, valueType)
     val comparator = InputTypedEquivDiffComparator[UnsafeMapData](equiv, dataType)
     MapDiffComparator[K, V](comparator)
diff --git a/src/test/scala/uk/co/gresearch/spark/diff/DiffComparatorSuite.scala b/src/test/scala/uk/co/gresearch/spark/diff/DiffComparatorSuite.scala
index d087dc74..c2401bdd 100644
--- a/src/test/scala/uk/co/gresearch/spark/diff/DiffComparatorSuite.scala
+++ b/src/test/scala/uk/co/gresearch/spark/diff/DiffComparatorSuite.scala
@@ -355,22 +355,24 @@ class DiffComparatorSuite extends AnyFunSuite with SparkTestSession {
     }
   }
 
-  Seq("true", "false").foreach { codegen =>
-    test(s"map comparator - codegen enabled=$codegen") {
-      withSQLConf(
-        SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen,
-        SQLConf.CODEGEN_FALLBACK.key -> "false"
-      ) {
-        val options = DiffOptions.default.withComparator(DiffComparators.map[Int, Long](), "map")
-
-        val actual = leftMaps.diff(rightMaps, options, "id").orderBy($"id").collect()
-        val diffs = Seq((1, "N"), (2, "C"), (3, "C"), (4, "D"), (5, "I"), (6, "N"), (7, "C")).toDF("id", "diff")
-        val expected = leftMaps.withColumnRenamed("map", "left_map")
-          .join(rightMaps.withColumnRenamed("map", "right_map"), Seq("id"), "fullouter")
-          .join(diffs, "id")
-          .select($"diff", $"id", $"left_map", $"right_map")
-          .orderBy($"id").collect()
-        assert(actual === expected)
+  Seq(true, false).foreach { sensitive =>
+    Seq("true", "false").foreach { codegen =>
+      test(s"map comparator - keyOrderSensitive=$sensitive - codegen enabled=$codegen") {
+        withSQLConf(
+          SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen,
+          SQLConf.CODEGEN_FALLBACK.key -> "false"
+        ) {
+          val options = DiffOptions.default.withComparator(DiffComparators.map[Int, Long](sensitive), "map")
+
+          val actual = leftMaps.diff(rightMaps, options, "id").orderBy($"id").collect()
+          val diffs = Seq((1, "N"), (2, "C"), (3, "C"), (4, "D"), (5, "I"), (6, if (sensitive) "C" else "N"), (7, "C")).toDF("id", "diff")
+          val expected = leftMaps.withColumnRenamed("map", "left_map")
+            .join(rightMaps.withColumnRenamed("map", "right_map"), Seq("id"), "fullouter")
+            .join(diffs, "id")
+            .select($"diff", $"id", $"left_map", $"right_map")
+            .orderBy($"id").collect()
+          assert(actual === expected)
+        }
       }
     }
   }

From 6be3887fd79ea951c528e88590e4bb7dcbf573e6 Mon Sep 17 00:00:00 2001
From: Enrico Minack <github@enrico.minack.dev>
Date: Wed, 23 Aug 2023 14:05:23 +0200
Subject: [PATCH 2/2] Replace default value with overloaded methods to keep
 backward compatibility

---
 .../uk/co/gresearch/spark/diff/DiffComparators.scala  | 11 ++++++++++-
 .../spark/diff/comparator/MapDiffComparator.scala     |  6 +++++-
 2 files changed, 15 insertions(+), 2 deletions(-)

diff --git a/src/main/scala/uk/co/gresearch/spark/diff/DiffComparators.scala b/src/main/scala/uk/co/gresearch/spark/diff/DiffComparators.scala
index a97c9169..1c9e2f1d 100644
--- a/src/main/scala/uk/co/gresearch/spark/diff/DiffComparators.scala
+++ b/src/main/scala/uk/co/gresearch/spark/diff/DiffComparators.scala
@@ -96,5 +96,14 @@ object DiffComparators {
   /**
    * This comparator compares two `Map[K,V]` values. They are equal when they match in all their keys and values.
    */
-  def map[K: Encoder, V: Encoder](keyOrderSensitive: Boolean = false): DiffComparator = MapDiffComparator[K, V](keyOrderSensitive)
+  def map[K: Encoder, V: Encoder](): DiffComparator = MapDiffComparator[K, V](keyOrderSensitive = false)
+
+  // for backward compatibility to v2.4.0 up to v2.8.0
+  // replace with default value in above map when moving to v3
+  /**
+   * This comparator compares two `Map[K,V]` values. They are equal when they match in all their keys and values.
+   *
+   * @param keyOrderSensitive comparator compares key order if true
+   */
+  def map[K: Encoder, V: Encoder](keyOrderSensitive: Boolean): DiffComparator = MapDiffComparator[K, V](keyOrderSensitive)
 }
diff --git a/src/main/scala/uk/co/gresearch/spark/diff/comparator/MapDiffComparator.scala b/src/main/scala/uk/co/gresearch/spark/diff/comparator/MapDiffComparator.scala
index 3f52eec7..d2ca79fc 100644
--- a/src/main/scala/uk/co/gresearch/spark/diff/comparator/MapDiffComparator.scala
+++ b/src/main/scala/uk/co/gresearch/spark/diff/comparator/MapDiffComparator.scala
@@ -55,7 +55,7 @@ private case class MapDiffEquiv[K: ClassTag, V](keyType: DataType, valueType: Da
 }
 
 case object MapDiffComparator {
-  def apply[K: Encoder, V: Encoder](keyOrderSensitive: Boolean = false): MapDiffComparator[K, V] = {
+  def apply[K: Encoder, V: Encoder](keyOrderSensitive: Boolean): MapDiffComparator[K, V] = {
     val keyType = encoderFor[K].schema.fields(0).dataType
     val valueType = encoderFor[V].schema.fields(0).dataType
     val equiv = MapDiffEquiv(keyType, valueType, keyOrderSensitive)
@@ -63,4 +63,8 @@ case object MapDiffComparator {
     val comparator = InputTypedEquivDiffComparator[UnsafeMapData](equiv, dataType)
     MapDiffComparator[K, V](comparator)
   }
+
+  // for backward compatibility to v2.4.0 up to v2.8.0
+  // replace with default value in above apply when moving to v3
+  def apply[K: Encoder, V: Encoder](): MapDiffComparator[K, V] = apply(keyOrderSensitive = false)
 }