[SPARK-40315][SQL] Add equals() and hashCode() to ArrayBasedMapData

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ArrayBasedMapData.scala

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.catalyst.util
 
+import scala.util.hashing.MurmurHash3
+
 import java.util.{Map => JavaMap}
 
 /**
@@ -35,6 +37,29 @@ class ArrayBasedMapData(val keyArray: ArrayData, val valueArray: ArrayData) exte
   override def toString: String = {
     s"keys: $keyArray, values: $valueArray"
   }
+
+  override def equals(obj: Any): Boolean = {
+    if (obj == null && this == null) {
+      return true
+    }
+
+    if (obj == null || !obj.isInstanceOf[ArrayBasedMapData]) {
+      return false
+    }
+
+    val other = obj.asInstanceOf[ArrayBasedMapData]
+
+    keyArray.equals(other.keyArray) && valueArray.equals(other.valueArray)
+  }
+
+  // Hash this class as a Product of two hashCodes. We don't know the DataType which prevents us
+  // from getting individual rows for hashing as a Map.
+  override def hashCode(): Int = {
+    val seed = MurmurHash3.productSeed
+    val keyHash = scala.util.hashing.MurmurHash3.mix(seed, keyArray.hashCode())
+    val valueHash = scala.util.hashing.MurmurHash3.mix(keyHash, valueArray.hashCode())
+    scala.util.hashing.MurmurHash3.finalizeHash(valueHash, 2)
+  }
 }
 
 object ArrayBasedMapData {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/ArrayBasedMapBuilderSuite.scala

@@ -142,4 +142,36 @@ class ArrayBasedMapBuilderSuite extends SparkFunSuite with SQLHelper {
         Map(new GenericArrayData(Seq(1, 1)) -> 3, new GenericArrayData(Seq(2, 2)) -> 2))
     }
   }
+
+  test("SPARK-40315: simple equal() and hashCode() semantics") {
+    val dataToAdd: Map[Int, Int] = Map(0 -> -7, 1 -> 3, 10 -> 4, 20 -> 5)
+    val builder1 = new ArrayBasedMapBuilder(IntegerType, IntegerType)
+    val builder2 = new ArrayBasedMapBuilder(IntegerType, IntegerType)
+    val builder3 = new ArrayBasedMapBuilder(IntegerType, IntegerType)
+    dataToAdd.foreach { case (key, value) =>
+      builder1.put(key, value)
+      builder2.put(key, value)
+      // Replace the value by something slightly different in builder3 for one of the keys.
+      if (key == 20) {
+        builder3.put(key, value - 1)
+      } else {
+        builder3.put(key, value)
+      }
+    }
+    val arrayBasedMapData1 = builder1.build()
+    val arrayBasedMapData2 = builder2.build()
+    val arrayBasedMapData3 = builder3.build()
+
+    // We expect two objects to be equal and to have the same hashCode if they have the same
+    // elements.
+    assert(arrayBasedMapData1 == arrayBasedMapData2)
+    assert(arrayBasedMapData1.equals(arrayBasedMapData2))
+    assert(arrayBasedMapData1.hashCode() == arrayBasedMapData2.hashCode())
+
+    // If two objects have different elements, we expect them to not be equal and their hashCode
+    // to be different.
+    assert(arrayBasedMapData1 != arrayBasedMapData3)
+    assert(!arrayBasedMapData1.equals(arrayBasedMapData3))
+    assert(arrayBasedMapData1.hashCode() != arrayBasedMapData3.hashCode())
+  }
 }