unify equals() and hashCode()

Author: Davies Liu

sql/catalyst/src/main/java/org/apache/spark/sql/BaseRow.java

@@ -155,39 +155,6 @@ public int fieldIndex(String name) {
     throw new UnsupportedOperationException();
   }
 
-  /**
-   * A generic version of Row.equals(Row), which is used for tests.
-   */
-  @Override
-  public boolean equals(Object other) {
-    if (other instanceof Row) {
-      Row row = (Row) other;
-      int n = size();
-      if (n != row.size()) {
-        return false;
-      }
-      for (int i = 0; i < n; i ++) {
-        if (isNullAt(i) != row.isNullAt(i)) {
-          return false;
-        }
-        if (!isNullAt(i)) {
-          Object o1 = get(i);
-          Object o2 = row.get(i);
-          if (o1 instanceof byte[]) {
-            // handle equals() of byte[]
-            if (!(o2 instanceof byte[]) || !java.util.Arrays.equals((byte[])o1, (byte[])o2)) {
-              return false;
-            }
-          } else if (!o1.equals(o2)) {
-            return false;
-          }
-        }
-      }
-      return true;
-    }
-    return false;
-  }
-
   @Override
   public InternalRow copy() {
     final int n = size();
@@ -227,15 +194,4 @@ public String mkString(String sep) {
   public String mkString(String start, String sep, String end) {
     return toSeq().mkString(start, sep, end);
   }
-
-  /*
-   * Returns hash code based on bytes in `arr`
-   * */
-  protected int bytesHashCode(byte[] arr) {
-    int hash = 0;
-    for (int i = 0; i < arr.length; i++) {
-      hash = hash * 37 + (int)arr[i];
-    }
-    return hash;
-  }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/Row.scala

@@ -17,8 +17,6 @@
 
 package org.apache.spark.sql
 
-import scala.util.hashing.MurmurHash3
-
 import org.apache.spark.sql.catalyst.expressions.GenericRow
 import org.apache.spark.sql.types.StructType
 
@@ -365,36 +363,6 @@ trait Row extends Serializable {
     false
   }
 
-  override def equals(that: Any): Boolean = that match {
-    case null => false
-    case that: Row =>
-      if (this.length != that.length) {
-        return false
-      }
-      var i = 0
-      val len = this.length
-      while (i < len) {
-        if (apply(i) != that.apply(i)) {
-          return false
-        }
-        i += 1
-      }
-      true
-    case _ => false
-  }
-
-  override def hashCode: Int = {
-    // Using Scala's Seq hash code implementation.
-    var n = 0
-    var h = MurmurHash3.seqSeed
-    val len = length
-    while (n < len) {
-      h = MurmurHash3.mix(h, apply(n).##)
-      n += 1
-    }
-    MurmurHash3.finalizeHash(h, n)
-  }
-
   /* ---------------------- utility methods for Scala ---------------------- */
 
   /**

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/InternalRow.scala

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.catalyst
 
 import org.apache.spark.sql.Row
-import org.apache.spark.sql.catalyst.expressions.GenericRow
+import org.apache.spark.sql.catalyst.expressions._
 
 /**
  * An abstract class for row used internal in Spark SQL, which only contain the columns as
@@ -27,6 +27,62 @@ import org.apache.spark.sql.catalyst.expressions.GenericRow
 abstract class InternalRow extends Row {
   // A default implementation to change the return type
   override def copy(): InternalRow = {this}
+
+  // A default version (slow), used for tests
+  override def equals(o: Any): Boolean = o match {
+    case other: InternalRow =>
+      if (length != other.length) {
+        return false
+      }
+
+      for (i <- 0 until length) {
+        if (isNullAt(i) != other.isNullAt(i)) {
+          return false
+        }
+        if (!isNullAt(i)) {
+          val o1 = apply(i)
+          val o2 = other.apply(i)
+          if (o1.isInstanceOf[Array[Byte]]) {
+            if (!o2.isInstanceOf[Array[Byte]] ||
+              !java.util.Arrays.equals(o1.asInstanceOf[Array[Byte]], o2.asInstanceOf[Array[Byte]])) {
+              return false
+            }
+          } else if (o1 != o2) {
+            return false
+          }
+        }
+      }
+      true
+    case _ => false
+  }
+
+  // Custom hashCode function that matches the efficient code generated version.
+  override def hashCode: Int = {
+    var result: Int = 37
+
+    for (i <- 0 until length) {
+      val update: Int =
+        if (isNullAt(i)) {
+          0
+        } else {
+          apply(i) match {
+            case b: Boolean => if (b) 0 else 1
+            case b: Byte => b.toInt
+            case s: Short => s.toInt
+            case i: Int => i
+            case l: Long => (l ^ (l >>> 32)).toInt
+            case f: Float => java.lang.Float.floatToIntBits(f)
+            case d: Double =>
+              val b = java.lang.Double.doubleToLongBits(d)
+              (b ^ (b >>> 32)).toInt
+            case a: Array[Byte] => java.util.Arrays.hashCode(a)
+            case other => other.hashCode()
+          }
+        }
+      result = 37 * result + update
+    }
+    result
+  }
 }
 
 object InternalRow {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/GenerateProjection.scala

@@ -127,7 +127,7 @@ object GenerateProjection extends CodeGenerator[Seq[Expression], Projection] {
         case FloatType => s"Float.floatToIntBits($col)"
         case DoubleType =>
             s"(int)(Double.doubleToLongBits($col) ^ (Double.doubleToLongBits($col) >>> 32))"
-        case BinaryType => s"bytesHashCode($col)"
+        case BinaryType => s"java.util.Arrays.hashCode($col)"
         case _ => s"$col.hashCode()"
       }
       s"isNullAt($i) ? 0 : ($nonNull)"

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/rows.scala

@@ -121,70 +121,6 @@ class GenericRow(protected[sql] val values: Array[Any]) extends InternalRow {
     }
   }
 
-  // TODO(davies): add getDate and getDecimal
-
-  // Custom hashCode function that matches the efficient code generated version.
-  override def hashCode: Int = {
-    var result: Int = 37
-
-    var i = 0
-    while (i < values.length) {
-      val update: Int =
-        if (isNullAt(i)) {
-          0
-        } else {
-          apply(i) match {
-            case b: Boolean => if (b) 0 else 1
-            case b: Byte => b.toInt
-            case s: Short => s.toInt
-            case i: Int => i
-            case l: Long => (l ^ (l >>> 32)).toInt
-            case f: Float => java.lang.Float.floatToIntBits(f)
-            case d: Double =>
-              val b = java.lang.Double.doubleToLongBits(d)
-              (b ^ (b >>> 32)).toInt
-            case a: Array[Byte] => a.map(_.toInt).fold(0)(_ * 37 + _)
-            case other => other.hashCode()
-          }
-        }
-      result = 37 * result + update
-      i += 1
-    }
-    result
-  }
-
-  override def equals(o: Any): Boolean = o match {
-    case other: InternalRow =>
-      if (values.length != other.length) {
-        return false
-      }
-
-      var i = 0
-      while (i < values.length) {
-        if (isNullAt(i) != other.isNullAt(i)) {
-          return false
-        }
-        if (!isNullAt(i)) {
-          val o1 = apply(i)
-          val o2 = other.apply(i)
-          if (o1.isInstanceOf[Array[Byte]]) {
-            val b1 = o1.asInstanceOf[Array[Byte]]
-            if (!o2.isInstanceOf[Array[Byte]] ||
-              java.util.Arrays.equals(b1, o2.asInstanceOf[Array[Byte]])) {
-              return false
-            }
-
-          } else if (apply(i) != other.apply(i)) {
-            return false
-          }
-        }
-        i += 1
-      }
-      true
-
-    case _ => false
-  }
-
   override def copy(): InternalRow = this
 }