not cast for same type compare

Author: adrian-wang

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/HiveTypeCoercion.scala

@@ -220,8 +220,7 @@ trait HiveTypeCoercion {
       case a: BinaryArithmetic if a.right.dataType == StringType =>
         a.makeCopy(Array(a.left, Cast(a.right, DoubleType)))
 
-      // we should cast all timestamp/date/string compare into string compare,
-      // even if both sides are of same type, as Hive use xxxwritable to compare.
+      // we should cast all timestamp/date/string compare into string compare
       case p: BinaryPredicate if p.left.dataType == StringType
         && p.right.dataType == DateType =>
         p.makeCopy(Array(p.left, Cast(p.right, StringType)))
@@ -240,13 +239,6 @@ trait HiveTypeCoercion {
       case p: BinaryPredicate if p.left.dataType == DateType
         && p.right.dataType == TimestampType =>
         p.makeCopy(Array(Cast(p.left, StringType), Cast(p.right, StringType)))
-      // same type
-      case p: BinaryPredicate if p.left.dataType == DateType
-        && p.right.dataType == DateType =>
-        p.makeCopy(Array(Cast(p.left, StringType), Cast(p.right, StringType)))
-      case p: BinaryPredicate if p.left.dataType == TimestampType
-        && p.right.dataType == TimestampType =>
-        p.makeCopy(Array(Cast(p.left, StringType), Cast(p.right, StringType)))
 
       case p: BinaryPredicate if p.left.dataType == StringType && p.right.dataType != StringType =>
         p.makeCopy(Array(Cast(p.left, DoubleType), p.right))

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

@@ -174,7 +174,7 @@ case class Cast(child: Expression, dataType: DataType) extends UnaryExpression w
   // Date cannot be cast to double, according to hive
   private[this] def dateToDouble(d: Date) = null
 
-  // Converts Timestamp to string according to Hive TimestampWritable convention
+  // Converts Date to string according to Hive DateWritable convention
   private[this] def dateToString(d: Date): String = {
     Cast.threadLocalDateFormat.get.format(d)
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/dataTypes.scala

@@ -256,7 +256,7 @@ case object DateType extends NativeType {
   @transient private[sql] lazy val tag = ScalaReflectionLock.synchronized { typeTag[JvmType] }
 
   private[sql] val ordering = new Ordering[JvmType] {
-    def compare(x: Date, y: Date) = x.compareTo(y)
+    def compare(x: Date, y: Date) = x.toString.compareTo(y.toString)
   }
 
   def simpleString: String = "date"

sql/core/src/main/scala/org/apache/spark/sql/columnar/ColumnStats.scala

@@ -198,8 +198,8 @@ private[sql] class DateColumnStats extends ColumnStats {
   override def gatherStats(row: Row, ordinal: Int) {
     if (!row.isNullAt(ordinal)) {
       val value = row(ordinal).asInstanceOf[Date]
-      if (upper == null || value.compareTo(upper) > 0) upper = value
-      if (lower == null || value.compareTo(lower) < 0) lower = value
+      if (upper == null || value.toString.compareTo(upper.toString) > 0) upper = value
+      if (lower == null || value.toString.compareTo(lower.toString) < 0) lower = value
     } else {
       nullCount += 1
     }