[SPARK-21646][SQL] Add new type coercion to compatible with Hive

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

@@ -116,19 +116,21 @@ object TypeCoercion {
    * other is a Timestamp by making the target type to be String.
    */
   val findCommonTypeForBinaryComparison: (DataType, DataType) => Option[DataType] = {
-    // We should cast all relative timestamp/date/string comparison into string comparisons
-    // This behaves as a user would expect because timestamp strings sort lexicographically.
-    // i.e. TimeStamp(2013-01-01 00:00 ...) < "2014" = true
-    case (StringType, DateType) => Some(StringType)
-    case (DateType, StringType) => Some(StringType)
-    case (StringType, TimestampType) => Some(StringType)
-    case (TimestampType, StringType) => Some(StringType)
-    case (TimestampType, DateType) => Some(StringType)
-    case (DateType, TimestampType) => Some(StringType)
+    // We should follow hive:
+    // https://github.com/apache/hive/blob/rel/storage-release-2.4.0/ql/src/java/
+    // org/apache/hadoop/hive/ql/exec/FunctionRegistry.java#L781
+    case (StringType, DateType) => Some(DateType)
+    case (DateType, StringType) => Some(DateType)
+    case (StringType, TimestampType) => Some(TimestampType)
+    case (TimestampType, StringType) => Some(TimestampType)
+    case (TimestampType, DateType) => Some(TimestampType)
+    case (DateType, TimestampType) => Some(TimestampType)
     case (StringType, NullType) => Some(StringType)
     case (NullType, StringType) => Some(StringType)
+    case (StringType | TimestampType, r: NumericType) => Some(DoubleType)
+    case (l: NumericType, StringType | TimestampType) => Some(DoubleType)
     case (l: StringType, r: AtomicType) if r != StringType => Some(r)
-    case (l: AtomicType, r: StringType) if (l != StringType) => Some(l)
+    case (l: AtomicType, r: StringType) if l != StringType => Some(l)
     case (l, r) => None
   }
 
@@ -352,7 +354,6 @@ object TypeCoercion {
         p.makeCopy(Array(Cast(left, TimestampType), right))
       case p @ Equality(left @ TimestampType(), right @ StringType()) =>
         p.makeCopy(Array(left, Cast(right, TimestampType)))
-
       case p @ BinaryComparison(left, right)
         if findCommonTypeForBinaryComparison(left.dataType, right.dataType).isDefined =>
         val commonType = findCommonTypeForBinaryComparison(left.dataType, right.dataType).get

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionSuite.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.analysis
 
-import java.sql.Timestamp
+import java.sql.{Date, Timestamp}
 
 import org.apache.spark.sql.catalyst.analysis.TypeCoercion._
 import org.apache.spark.sql.catalyst.dsl.expressions._
@@ -1101,13 +1101,34 @@ class TypeCoercionSuite extends AnalysisTest {
   test("binary comparison with string promotion") {
     ruleTest(PromoteStrings,
       GreaterThan(Literal("123"), Literal(1)),
-      GreaterThan(Cast(Literal("123"), IntegerType), Literal(1)))
+      GreaterThan(Cast(Literal("123"), DoubleType), Cast(Literal(1), DoubleType)))
     ruleTest(PromoteStrings,
       LessThan(Literal(true), Literal("123")),
       LessThan(Literal(true), Cast(Literal("123"), BooleanType)))
     ruleTest(PromoteStrings,
       EqualTo(Literal(Array(1, 2)), Literal("123")),
       EqualTo(Literal(Array(1, 2)), Literal("123")))
+    ruleTest(PromoteStrings,
+      GreaterThan(Literal("123"), Literal(1L)),
+      GreaterThan(Cast(Literal("123"), DoubleType), Cast(Literal(1L), DoubleType)))
+    ruleTest(PromoteStrings,
+      GreaterThan(Literal("123"), Literal(0.1)),
+      GreaterThan(Cast(Literal("123"), DoubleType), Literal(0.1)))
+
+    val date1 = Date.valueOf("2017-07-21")
+    val timestamp1 = Timestamp.valueOf("2017-07-21 23:42:12.123")
+    ruleTest(PromoteStrings,
+      GreaterThan(Literal(date1), Literal("2017-07-01")),
+      GreaterThan(Literal(date1), Cast(Literal("2017-07-01"), DateType)))
+    ruleTest(PromoteStrings,
+      GreaterThan(Literal(timestamp1), Literal("2017-07-01")),
+      GreaterThan(Literal(timestamp1), Cast(Literal("2017-07-01"), TimestampType)))
+    ruleTest(PromoteStrings,
+      GreaterThan(Literal(timestamp1), Cast(Literal("2017-07-01"), DateType)),
+      GreaterThan(Literal(timestamp1), Cast(Cast(Literal("2017-07-01"), DateType), TimestampType)))
+    ruleTest(PromoteStrings,
+      GreaterThan(Literal(timestamp1), Literal(1L)),
+      GreaterThan(Cast(Literal(timestamp1), DoubleType), Cast(Literal(1L), DoubleType)))
   }
 
   test("cast WindowFrame boundaries to the type they operate upon") {

sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala

@@ -1966,7 +1966,7 @@ class DataFrameSuite extends QueryTest with SharedSQLContext {
 
   test("SPARK-17913: compare long and string type column may return confusing result") {
     val df = Seq(123L -> "123", 19157170390056973L -> "19157170390056971").toDF("i", "j")
-    checkAnswer(df.select($"i" === $"j"), Row(true) :: Row(false) :: Nil)
+    checkAnswer(df.select($"i" === $"j"), Row(true) :: Row(true) :: Nil)
   }
 
   test("SPARK-19691 Calculating percentile of decimal column fails with ClassCastException") {

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -20,7 +20,7 @@ package org.apache.spark.sql
 import java.io.File
 import java.math.MathContext
 import java.net.{MalformedURLException, URL}
-import java.sql.Timestamp
+import java.sql.{Date, Timestamp}
 import java.util.concurrent.atomic.AtomicBoolean
 
 import org.apache.spark.{AccumulatorSuite, SparkException}
@@ -2677,4 +2677,67 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
       checkAnswer(df, Row(1, 1, 1))
     }
   }
+
+  test("SPARK-21646: CommonTypeForBinaryComparison: StringType vs NumericType") {
+    withTempView("v") {
+      val str1 = Long.MaxValue.toString + "1"
+      val str2 = Int.MaxValue.toString + "1"
+      val str3 = "10"
+      Seq(str1, str2, str3).toDF("c1").createOrReplaceTempView("v")
+      checkAnswer(sql("SELECT c1 from v where c1 > 0"), Row(str1) :: Row(str2) :: Row(str3) :: Nil)
+      checkAnswer(sql("SELECT c1 from v where c1 > 0L"), Row(str1) :: Row(str2) :: Row(str3) :: Nil)
+    }
+  }
+
+  test("SPARK-21646: CommonTypeForBinaryComparison: DoubleType vs IntegerType") {
+    withTempView("v") {
+      Seq(("0", 1), ("-0.4", 2)).toDF("a", "b").createOrReplaceTempView("v")
+      checkAnswer(sql("SELECT a FROM v WHERE a=0"), Seq(Row("0")))
+      checkAnswer(sql("SELECT a FROM v WHERE a=0L"), Seq(Row("0")))
+      checkAnswer(sql("SELECT a FROM v WHERE a=0.0"), Seq(Row("0")))
+      checkAnswer(sql("SELECT a FROM v WHERE a=-0.4"), Seq(Row("-0.4")))
+    }
+  }
+
+  test("SPARK-21646: CommonTypeForBinaryComparison: StringType vs DateType") {
+    withTempView("v") {
+      val v1 = Date.valueOf("2017-09-22")
+      val v2 = Date.valueOf("2017-09-09")
+      Seq(v1, v2).toDF("c1").createTempView("v")
+      checkAnswer(sql("select * from v where c1 > '2017-8-1'"), Row(v1) :: Row(v2) :: Nil)
+      checkAnswer(sql("select * from v where c1 > cast('2017-8-1' as date)"),
+        Row(v1) :: Row(v2) :: Nil)
+    }
+  }
+
+  test("SPARK-21646: CommonTypeForBinaryComparison: StringType vs TimestampType") {
+    withTempView("v") {
+      val v1 = Timestamp.valueOf("2017-07-21 23:42:12.123")
+      val v2 = Timestamp.valueOf("2017-08-21 23:42:12.123")
+      val df = Seq(v1, v2).toDF("c1").createTempView("v")
+      checkAnswer(sql("select * from v where c1 > '2017-8-1'"), Row(v2) :: Nil)
+      checkAnswer(sql("select * from v where c1 > cast('2017-8-1' as timestamp)"), Row(v2) :: Nil)
+    }
+  }
+
+  test("SPARK-21646: CommonTypeForBinaryComparison: TimestampType vs DateType") {
+    withTempView("v") {
+      val v1 = Timestamp.valueOf("2017-07-21 23:42:12.123")
+      val v2 = Timestamp.valueOf("2017-08-21 23:42:12.123")
+      val df = Seq(v1, v2).toDF("c1").createTempView("v")
+      checkAnswer(sql("select * from v where c1 > cast('2017-8-1' as date)"), Row(v2) :: Nil)
+      checkAnswer(sql("select * from v where c1 > cast('2017-8-1' as timestamp)"), Row(v2) :: Nil)
+    }
+  }
+
+  test("SPARK-21646: CommonTypeForBinaryComparison: TimestampType vs NumericType") {
+    withTempView("v") {
+      val v1 = Timestamp.valueOf("2017-07-21 23:42:12.123")
+      val v2 = Timestamp.valueOf("2017-08-21 23:42:12.123")
+      val df = Seq(v1, v2).toDF("c1").createTempView("v")
+      checkAnswer(sql("select * from v where c1 > 1"), Row(v1) :: Row(v2) :: Nil)
+      checkAnswer(sql("select * from v where c1 > cast(cast('2010-08-01' as timestamp) as double)"),
+        Row(v1) :: Row(v2) :: Nil)
+    }
+  }
 }