[SPARK-7952][SPARK-7984][SQL] equality check between boolean type and numeric type is broken.

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

@@ -76,7 +76,7 @@ trait HiveTypeCoercion {
     WidenTypes ::
     PromoteStrings ::
     DecimalPrecision ::
-    BooleanComparisons ::
+    BooleanEqualization ::
     StringToIntegralCasts ::
     FunctionArgumentConversion ::
     CaseWhenCoercion ::
@@ -482,30 +482,49 @@ trait HiveTypeCoercion {
   }
 
   /**
-   * Changes Boolean values to Bytes so that expressions like true < false can be Evaluated.
+   * Changes numeric values to booleans so that expressions like true = 1 can be evaluated.
    */
-  object BooleanComparisons extends Rule[LogicalPlan] {
-    val trueValues = Seq(1, 1L, 1.toByte, 1.toShort, new java.math.BigDecimal(1)).map(Literal(_))
-    val falseValues = Seq(0, 0L, 0.toByte, 0.toShort, new java.math.BigDecimal(0)).map(Literal(_))
+  object BooleanEqualization extends Rule[LogicalPlan] {
+    val trueValue = Literal(new java.math.BigDecimal(1))
+    val falseValue = Literal(new java.math.BigDecimal(0))
+
+    private def buildCaseKeyWhen(booleanExpr: Expression, numericExpr: Expression) = {
+      CaseKeyWhen(Cast(numericExpr, DecimalType.Unlimited),
+        Seq(
+          trueValue, booleanExpr,
+          falseValue, Not(booleanExpr),
+          Literal(false)))
+    }
+
+    private def transform(booleanExpr: Expression, numericExpr: Expression) = {
+      CaseWhen(Seq(
+        Or(IsNull(booleanExpr), IsNull(numericExpr)), Literal.create(null, BooleanType),
+        buildCaseKeyWhen(booleanExpr, numericExpr)
+      ))
+    }
+
+    private def transformNullSafe(booleanExpr: Expression, numericExpr: Expression) = {
+      CaseWhen(Seq(
+        And(IsNull(booleanExpr), IsNull(numericExpr)), Literal(true),
+        Or(IsNull(booleanExpr), IsNull(numericExpr)), Literal(false),
+        buildCaseKeyWhen(booleanExpr, numericExpr)
+      ))
+    }
 
     def apply(plan: LogicalPlan): LogicalPlan = plan transformAllExpressions {
       // Skip nodes who's children have not been resolved yet.
       case e if !e.childrenResolved => e
 
-      // Hive treats (true = 1) as true and (false = 0) as true.
-      case EqualTo(l @ BooleanType(), r) if trueValues.contains(r) => l
-      case EqualTo(l, r @ BooleanType()) if trueValues.contains(l) => r
-      case EqualTo(l @ BooleanType(), r) if falseValues.contains(r) => Not(l)
-      case EqualTo(l, r @ BooleanType()) if falseValues.contains(l) => Not(r)
-
-      // No need to change other EqualTo operators as that actually makes sense for boolean types.
-      case e: EqualTo => e
-      // No need to change the EqualNullSafe operators, too
-      case e: EqualNullSafe => e
-      // Otherwise turn them to Byte types so that there exists and ordering.
-      case p: BinaryComparison if p.left.dataType == BooleanType &&
-                                  p.right.dataType == BooleanType =>
-        p.makeCopy(Array(Cast(p.left, ByteType), Cast(p.right, ByteType)))
+      // Hive treats (true = 1) as true and (false = 0) as true,
+      // all other cases are considered as false.
+      case EqualTo(l @ BooleanType(), r) if r.dataType.isInstanceOf[NumericType] =>
+        transform(l, r)
+      case EqualTo(l, r @ BooleanType()) if l.dataType.isInstanceOf[NumericType] =>
+        transform(r, l)
+      case EqualNullSafe(l @ BooleanType(), r) if r.dataType.isInstanceOf[NumericType] =>
+        transformNullSafe(l, r)
+      case EqualNullSafe(l, r @ BooleanType()) if l.dataType.isInstanceOf[NumericType] =>
+        transformNullSafe(r, l)
     }
   }
 

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

@@ -1331,4 +1331,25 @@ class SQLQuerySuite extends QueryTest with BeforeAndAfterAll {
 
     checkAnswer(sql("SELECT a.`c.b`, `b.$q`[0].`a@!.q`, `q.w`.`w.i&`[0] FROM t"), Row(1, 1, 1))
   }
+
+  test("SPARK-7952: fix the equality check between boolean and numeric types") {
+    val df = Seq(
+      (1, true),
+      (0, false),
+      (2, true),
+      (2, false),
+      (null, true),
+      (null, false),
+      (0, null),
+      (1, null),
+      (null, null)
+    ).map { case (i, b) =>
+      (i.asInstanceOf[Integer], b.asInstanceOf[java.lang.Boolean])
+    }.toDF("i", "b")
+
+    checkAnswer(df.select('i === 'b),
+      Seq(true, true, false, false, null, null, null, null, null).map(Row(_)))
+    checkAnswer(df.select('i <=> 'b),
+      Seq(true, true, false, false, false, false, false, false, true).map(Row(_)))
+  }
 }